lambda-recipes.md

Lambda Recipes

Short patterns you can copy into real lambdas.

1) Periodic health check

import jbapi

def query():
    try:
        gps = jbapi.get_sensor("/sensor/gps/fix")
    except TimeoutError:
        return

    if gps is None:
        return

    if gps["status"]["status"] < 0:
        jbapi.trigger("gps_fault", {"preroll": 5, "timeout": 10}, {"reason": "no_fix"})

2) On-topic trigger

When your lambda should run only when a topic publishes, set the trigger in config:

function:
  name: query
  execute_on:
    type: OnTopic
    topic: /sensor/camera/image_raw

Make sure the same topic appears in subscriptions.

3) Event-based recording with postroll

import jbapi

TRIGGER = "near_miss"

def query():
    if is_near_miss():
        jbapi.trigger(TRIGGER, {"preroll": 3, "timeout": 8}, {"type": "near_miss"})

    if should_stop_recording():
        jbapi.stop_trigger(TRIGGER, {"postroll": 2})

4) Image + ONNX inference

import jbapi
import numpy as np

def query():
    image = jbapi.get_image("/camera/front")
    if image is None:
        return

    input_tensor = preprocess(image).astype(np.float32)
    outputs = jbapi.run_onnx("detector", {"input": input_tensor})
    score = float(outputs["score"].max())

    if score > 0.9:
        jbapi.trigger("rare_object", {"preroll": 2, "timeout": 6}, {"score": score})

Note: the string passed to run_onnx must match the assets[].name in your config.

5) Defensive topic reads

import jbapi

def query():
    try:
        data = jbapi.get_sensor("/sensor/imu/data")
    except TimeoutError:
        return
    except ValueError:
        # Topic not subscribed or invalid
        return

    if data is None:
        return