Writing a rule

Almost every rule reduces to one question: has this object been somewhere, for long enough?

On the facade that is one line, and the two clocks are handled for you:

@app.on_detection("person", zone="driveway", dwell="$dwell_s", cooldown=60)
def loitering(event):
    event.alert(f"Person loitering on {event.camera}", severity="high")

dwell measures time spent satisfying this rule's filters — with a zone, that is time in the zone, not time on camera — and fires once per presence episode. forget= (default max(30s, dwell)) is the gap that ends an episode and re-arms it, so the second person of the day alerts too.

The rest of this page is what that wraps, for a rule that needs to do it by hand: geometry answers the where, keyed_state the how long, and an Alert is what you do about it.

Where — zones and tripwires

Coordinates are normalized (0–1 of the frame) throughout, matching the platform's NormalizedBBox, so a rule written against one camera resolution works on all of them. The catalog's zone editor emits normalized vertices for the same reason.

def in_driveway(detection: dict) -> bool:
    """A detection's bbox centre against a polygon."""
    centre = bbox_center(detection.get("bbox", {}), 1, 1)
    return DRIVEWAY.contains(centre)


# ── Tripwires: did the object cross a line, and which way? ──────────

ENTRANCE = Tripwire.from_config("entrance", [0.0, 0.5], [1.0, 0.5])


def crossed(previous: Point, current: Point) -> str | None:
    """`"a_to_b"`, `"b_to_a"`, or None. Direction is what separates
    "12 people entered" from "12 people milled about the door"."""

A tripwire adds direction, which is what separates "12 people entered" from "12 people milled about the door".

How long — keyed TTL state

class Dwell:
    """The loitering pattern in nine lines.

    TTL is in seconds of EVENT time — whatever timeline you pass to
    `touch(at=...)`. A key not touched within the TTL is garbage-
    collected, which is what ends a presence episode and re-arms the
    latch.
    """

    def __init__(self, threshold_s: float = 30.0) -> None:
        self.threshold = threshold_s
        # ttl comfortably longer than the gap between events, shorter
        # than "the object really left".
        self.present = keyed_state(ttl=10.0)

    def saw(self, camera_id: str, track_id: str, at: float) -> float | None:
        """Returns the dwell time the first moment it crosses the
        threshold, then None until the object leaves and returns."""
        record = self.present.touch((camera_id, track_id), at=at)
        if record.age < self.threshold or record.alerted:
            return None
        record.alerted = True                 # the latch: once per episode
        return record.age

    def note(self, camera_id: str, track_id: str, **values) -> None:
        """`record.data` is a free-form scratchpad per key — a phase, a
        counter, the last zone the object was in."""
        self.present[(camera_id, track_id)].data.update(values)

Two things to get right:

• TTL is event time. Pass at= from the event's own timestamp, not the wall clock, or a replayed batch will skew every timer.
• Latch, then let the TTL re-arm it. record.alerted fires once per presence episode; when the key is not touched for the TTL it is garbage-collected, which is what makes the next appearance a new episode.

What to fire

Keep the title to one line an operator can act on, put the numbers in evidence, and thread the correlation_id so the alert joins the inference, the audit line and the evidence frame in one chain.

def a_well_formed_alert(camera_id: str, correlation_id: str) -> Alert:
    """`alert_id` and `fired_at` fill themselves in; `source` comes from
    the app's identity, which the archetype sets around every handler
    call. What is worth your attention is the rest."""
    return Alert(
        title=f"Loitering on {camera_id}",          # one line, in the inbox
        description="A person has been by the cars for 40 seconds.",
        camera_id=camera_id,
        severity="high",                            # low|medium|high|critical
        # Thread the platform's correlation id so the alert joins the
        # inference, the audit line and the evidence frame in one chain.
        correlation_id=correlation_id,
        # Anything a consumer might route or filter on. Keep it flat and
        # JSON-serializable.
        evidence={"dwell_s": 40.2, "label": "person", "confidence": 0.91},
        # Cheap, greppable routing hints.
        tags=["loitering", "night"],
    )

Full example: 08_state_and_geometry.py, 18_alerts_and_channels.py.