Several programs need the same camera

Run a detector, a recorder and alerts on one camera, each in a stream of its own, so each can change or fail without touching the others. A security and operations example.

A camera in a lobby, on a loading dock or over a workbench usually ends up with several jobs: a model that counts or detects, a recording, an alert when something happens, and a person or an agent who wants to look now and then. Each job wants the camera, and a camera can only be opened by one program at a time. So the jobs get bundled into one application, where changing the model means restarting the recording, and a crash in the detector takes everything down.

With Tatolab, one stream owns the camera and exposes its video. Each job is a stream of its own that reads it.

lobby_camera:    CameraSource ─────────────────────────────▶ video (exposed)
lobby_counts:    video ─▶ ObjectDetector ─▶ ZoneCounter ─┬─▶ occupancy (exposed)
                                                         └─▶ CrowdingAlert ─▶ your webhook
lobby_recorder:  video ─▶ H264Encoder ─▶ Mp4Sink ─▶ recordings/lobby-….mp4

The streams

All three live in one project’s stream.py. They could just as well be three projects, owned by three people:

from datetime import datetime
from pathlib import Path

from tatolab.stream import CameraSource, H264Encoder, Mp4Sink, StreamBuilder, stream

from nodes.crowding_alert import CrowdingAlert
from nodes.object_detector import ObjectDetector
from nodes.zone_counter import ZoneCounter

RECORDINGS = Path(__file__).resolve().parent / "recordings"


@stream
def lobby_camera(stream_builder: StreamBuilder) -> None:
    """Opens the lobby camera and shares it with the other streams on this machine."""
    camera = stream_builder.add(CameraSource, name="camera", config={"max_width": 1280, "max_height": 720})
    stream_builder.expose(camera.output("video"))


@stream
def lobby_counts(stream_builder: StreamBuilder) -> None:
    """Counts people in the waiting area, and sends an alert when it gets crowded."""
    detector = stream_builder.add(ObjectDetector, name="detector", config={"confidence": 0.45})
    counter = stream_builder.add(ZoneCounter, name="waiting_area", config={"zone": [0, 240, 640, 720], "alert_above": 5})
    alert = stream_builder.add(CrowdingAlert, config={"webhook_url": "https://chat.example.com/hooks/lobby"})
    stream_builder.connect(stream_builder.remote_output("lobby_camera/camera/video"), detector.input("video"))
    stream_builder.connect(detector.output("detections"), counter.input("detections"))
    stream_builder.connect(counter.output("occupancy"), alert.input("occupancy"))
    stream_builder.expose(counter.output("occupancy"))


@stream
def lobby_recorder(stream_builder: StreamBuilder) -> None:
    """Records the lobby camera, to a new file each time it starts."""
    path = RECORDINGS / f"lobby-{datetime.now():%Y%m%d-%H%M%S}.mp4"
    encoder = stream_builder.add(H264Encoder, config={"bitrate_bps": 2_000_000})
    recording = stream_builder.add(Mp4Sink, config={"path": str(path)})
    stream_builder.connect(stream_builder.remote_output("lobby_camera/camera/video"), encoder.input("video"))
    stream_builder.connect(encoder.output("encoded_video"), recording.input("tracks"))

ObjectDetector is the YOLO node from Use your model. The counter checks which detections are inside a zone:

"""Counts detections inside a zone, and flags when too many are there."""

from dataclasses import dataclass, field
from typing import Annotated

from tatolab.stream import RuntimeContextLimitedAccess, log, node


@dataclass
class ZoneCounterConfig:
    zone: Annotated[list[float], "The zone as [left, top, right, bottom], in pixels."] = field(
        default_factory=lambda: [0.0, 0.0, 640.0, 720.0]
    )
    label: Annotated[str, "Which kind of detection to count."] = "person"
    alert_above: Annotated[int, "Flag the zone as crowded above this many."] = 3


@node
class ZoneCounter:
    """Counts detections whose center is inside a zone, and flags crowding."""

    @node.input(delivery_profile="newest")
    def detections(self) -> dict: ...

    @node.output(description="Per frame: how many are in the zone, and whether it's crowded")
    def occupancy(self) -> dict: ...

    def __init__(self, config: ZoneCounterConfig) -> None:
        self.left, self.top, self.right, self.bottom = config.zone
        self.label = config.label
        self.alert_above = config.alert_above
        self.crowded = False

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        detections = ctx.inputs.read("detections")
        if detections is None:
            return
        inside = 0
        for found in detections["objects"]:
            if found["label"] != self.label:
                continue
            x0, y0, x1, y1 = found["box"]
            center_x, center_y = (x0 + x1) / 2, (y0 + y1) / 2
            if self.left <= center_x <= self.right and self.top <= center_y <= self.bottom:
                inside += 1

        crowded = inside > self.alert_above
        ctx.outputs.write("occupancy", {"count": inside, "crowded": crowded}, timestamp_ns=detections["timestamp_ns"])
        if crowded and not self.crowded:
            log.warn("waiting area crowded", count=inside)
        elif self.crowded and not crowded:
            log.info("waiting area clear", count=inside)
        self.crowded = crowded

And the alert posts to a chat webhook when the zone becomes crowded:

"""Posts to a webhook when the waiting area becomes crowded."""

import json
import urllib.request
from dataclasses import dataclass

from tatolab.stream import RuntimeContextLimitedAccess, log, node


@dataclass
class CrowdingAlertConfig:
    webhook_url: str = ""


@node
class CrowdingAlert:
    """Posts a message to a webhook each time the zone becomes crowded."""

    @node.input(delivery_profile="ordered")
    def occupancy(self) -> dict: ...

    def __init__(self, config: CrowdingAlertConfig) -> None:
        self.webhook_url = config.webhook_url
        self.crowded = False

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (reading := ctx.inputs.read("occupancy")) is not None:
            if reading["crowded"] and not self.crowded:
                self.post(f"Waiting area crowded: {reading['count']} people")
            self.crowded = reading["crowded"]

    def post(self, text: str) -> None:
        request = urllib.request.Request(
            self.webhook_url,
            data=json.dumps({"text": text}).encode(),
            headers={"Content-Type": "application/json"},
        )
        try:
            urllib.request.urlopen(request, timeout=5).close()
        except OSError as failure:
            log.warn("alert not sent", error=str(failure))

Run them

Create the recordings/ folder first; Mp4Sink creates files, not folders. Then keep the camera and the counts, and record when you want to:

mkdir -p recordings
tatolab run -d stream.py:lobby_camera
tatolab run -d stream.py:lobby_counts
tatolab run stream.py:lobby_recorder      # attached: records until Ctrl-C

What you get from splitting it up:

Change the model without touching the recording. tatolab stop lobby_counts, edit, tatolab run -d stream.py:lobby_counts. The camera and the recorder never notice.
A detector crash doesn’t stop the recording. It’s in another stream, and its nodes run in their own processes.
Anyone can add a reader. A dashboard, a second model or an agent reads lobby_camera/camera/video or lobby_counts/waiting_area/occupancy without changing these streams.

Recording around the clock

The recorder is run attached on purpose: a kept stream reloads its recorded graph after every restart, and Mp4Sink empties its file when it loads, so a kept recorder would overwrite its last file after a reboot. For continuous recording, start the recorder again on a schedule, each run with a new file:

while true; do timeout 1h tatolab run stream.py:lobby_recorder; done

When timeout stops tatolab run, the stream unloads and the file is closed cleanly. Run the loop as a service if it should survive reboots, and delete old files on your own retention schedule.

What leaves the machine

Only what you send: the alert’s webhook message, and anything a node of yours posts. The video stays on the machine unless an agent looks at frames: every frame an agent exchanges goes to its model provider. Decide whether that’s acceptable before connecting an agent to a camera that sees people, and follow the rules for video surveillance where you operate.

Today’s limits

IP cameras aren’t built in. CameraSource reads cameras the operating system sees. An RTSP camera needs a source node you write, for example with PyAV, and the docs don’t show a complete one yet. ONVIF, VMS integration and event clips with pre-roll aren’t built in either.
USB webcam frame rates on Linux. CameraSource reads uncompressed video only, so most USB 2.0 webcams deliver 5 to 10 frames per second in HD. The example asks for 720p; lower it, or use a USB 3 camera, if you need more.
A crashed node stays down until its stream loads again. Alert on silence too: if occupancy stops arriving, the detector isn’t running.
Plain MP4 files. There’s no wall-clock time inside the file beyond its name. If recordings may be used as evidence, add your own log of times, checksums and export procedure.

Share between streams: exposure and addresses.
Use your model: the detector.
Recordings and restarts: why the recorder runs attached.
Deploy to a machine: run it on a box near the camera.