--- title: "Several programs need the same camera" description: "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." type: "docs-page" url: "https://tatolab.com/docs/use-cases/video-analytics" site: "Tatolab" section: "Use cases" --- # 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. ```txt 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: ```python title="stream.py" 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](https://tatolab.com/docs/build/models). The counter checks which detections are inside a zone: ```python title="nodes/zone_counter.py" """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: ```python title="nodes/crowding_alert.py" """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: ```sh 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: ```sh 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. ## Related guides - [Share between streams](https://tatolab.com/docs/share/between-streams): exposure and addresses. - [Use your model](https://tatolab.com/docs/build/models): the detector. - [Recordings and restarts](https://tatolab.com/docs/build/recording#recordings-and-restarts): why the recorder runs attached. - [Deploy to a machine](https://tatolab.com/docs/run/deploy): run it on a box near the camera.