Real-time object detection with Python and YOLO: RTSP camera to event

Real-time object detection is the process of finding the class and position of objects in the frames of a camera stream within milliseconds. On its own it is not a result; it is the middle link in a chain that ends in events such as “how many people came in” or “someone is in the restricted area”. In this article we build the chain from an RTSP camera to MQTT with Python, OpenCV and Ultralytics YOLO, piece by piece, ending with a working script.

I described the architecture of that chain, and where it tends to break in the field, in the chain from RTSP stream to event. Here I turn the same chain into code: why each line is there, and which defaults will mislead you on site.

What are we building?

The script does five things:

1. Reads the RTSP stream in a separate thread and keeps only the newest frame.
2. Runs YOLO person detection on selected frames.
3. Uses ByteTrack to give each person an identity that persists across frames.
4. Runs two rules on the tracks: line crossing (in/out counting) and zone intrusion (staying in a restricted area longer than a threshold).
5. When a rule fires, sends a JSON event with no image to MQTT and, optionally, to a webhook.

Component	Library	Job
Stream intake	OpenCV (FFmpeg backend)	Open RTSP, decode frames, reconnect
Detection and tracking	Ultralytics YOLO + ByteTrack	Boxes and track IDs
Rules	Plain Python + cv2.pointPolygonTest	Line and polygon logic
Event output	paho-mqtt 2.x, urllib	MQTT publish and webhook

Before writing code: the licence

Ultralytics YOLO is distributed under AGPL-3.0. According to Ultralytics’ licensing page, complying with it means publishing the source code of the entire derivative work, including the larger application, and trained models fall under AGPL-3.0 by default. If you are building a closed-source product, a SaaS or a system embedded in an edge device, you need an Enterprise License.

My advice is to settle the licence question in the first week. A prototype can happily run on AGPL; but if the licence is noticed only when the system is about to be handed over, you either open the source or swap the model. If you choose a different detector, check the licence of the code and of the weights separately.

Step 1: Set up the environment

Python 3.10 or newer is enough. The ultralytics package pulls in PyTorch and OpenCV as dependencies; we add paho-mqtt for MQTT.

python3 -m venv .venv && source .venv/bin/activate
pip install ultralytics paho-mqtt
python -c "import torch; print('CUDA:', torch.cuda.is_available())"
python -c "import cv2; print([s.strip() for s in cv2.getBuildInformation().splitlines() if 'FFMPEG' in s])"

The last line matters: for OpenCV to open RTSP through its FFmpeg backend, the build information must show FFmpeg support as YES. Not every platform’s prebuilt package includes it, so check on the target machine before going on site.

Step 2: Read the RTSP stream reliably

The first decision is main stream or sub-stream. People counting and zone intrusion usually work on the sub-stream, which lowers both network traffic and decoding load.

The second is the transport. OpenCV’s FFmpeg backend reads the OPENCV_FFMPEG_CAPTURE_OPTIONS environment variable in key;value|key;value form. When the variable is empty it uses a flag that prefers TCP; writing rtsp_transport;tcp makes the behaviour explicit and independent of the version. Set it before cv2 is imported.

The third is timeouts. How long read() waits when a camera goes away is set by CAP_PROP_OPEN_TIMEOUT_MSEC and CAP_PROP_READ_TIMEOUT_MSEC; both apply only to the FFmpeg and GStreamer backends and must be passed as parameters when opening. If you omit them, the default in OpenCV’s source is 30 seconds. On site, a silent 30-second wait translates into a “the system froze” complaint.

The most common mistake is about latency. cap.read() delivers frames in order; if your processing is slower than the camera’s frame rate, the gap accumulates and after a while you are processing an image from several seconds ago. CAP_PROP_BUFFERSIZE is not the fix: in OpenCV’s source the FFmpeg backend does not handle that property at all; it exists for local capture backends such as V4L2. The fix is to move reading into its own thread and keep only the newest frame:

cap = cv2.VideoCapture(url, cv2.CAP_FFMPEG, [
    cv2.CAP_PROP_OPEN_TIMEOUT_MSEC, 5000,
    cv2.CAP_PROP_READ_TIMEOUT_MSEC, 5000,
])
while not stop.is_set():
    ok, frame = cap.read()
    if not ok:
        break                      # the outer loop reconnects with exponential backoff
    with lock:
        latest_frame, seq = frame, seq + 1

Because H.264 and H.265 frames depend on each other, the reader has to decode every frame; inference, however, runs only on the latest one. When the connection drops, the reader retries at intervals that start at 1 second and double up to 30 seconds. Do not hard-code the username and password in the RTSP URL; read them from an environment variable and never print the URL to logs.

Step 3: Detect with YOLO, track with ByteTrack

Ultralytics’ current model family is YOLO26, available in n, s, m, l and x scales. On site the right order is usually to start with the smallest scale and grow only if accuracy falls short.

from ultralytics import YOLO

model = YOLO("yolo26n.pt")         # one model object per camera
result = model.track(frame, persist=True, tracker="bytetrack.yaml",
                     classes=[0], conf=0.35, device=0, verbose=False)[0]
if result.boxes and result.boxes.is_track:
    ids = result.boxes.id.int().cpu().tolist()
    boxes = result.boxes.xyxy.cpu().tolist()

Four details matter:

• persist=True tells the tracker that this frame is the next one in the same stream. The Ultralytics documentation explicitly says not to use it across unrelated images or different streams, so do not push several cameras through one model object; their tracker states would get mixed.
• Name the tracker explicitly. New trackers arrived with ultralytics 8.4.63, and the default used when you pass no tracker is now TrackTrack. If you do not want behaviour to change silently on an upgrade, pin a value such as bytetrack.yaml.
• When a frame has no tracks, boxes.id is empty; the is_track check catches that.
• track_buffer in bytetrack.yaml (default 30) is how many frames a lost track is kept alive, counted in frames passed to the tracker. At 8 processed frames per second, 30 frames is about 3.75 seconds. Revisit it whenever you change the frame-skip rate.

Step 4: Count with a line crossing

Counting rests on simple geometry: which side of a line a point lies on is given by the sign of a cross product. When a track ID’s sign flips, it has crossed the line. In practice three extra conditions are needed:

• Which point? With an angled camera use the bottom centre of the box (the foot point); with a top-down camera use the box centre. At an angle the box centre shifts with the person’s height.
• A segment, not an infinite line. If the point does not project between the two ends of the line, it is not a crossing; otherwise someone walking along the corridor next to the door would be counted too.
• Jitter. Boxes wobble by a few pixels every frame. Ignoring points closer than HYSTERESIS_PX to the line and adding a 1-second cooldown per ID stops double counts.

def signed_distance(p, a, b):
    abx, aby = b[0] - a[0], b[1] - a[1]
    cross = abx * (p[1] - a[1]) - aby * (p[0] - a[0])
    return cross / max((abx * abx + aby * aby) ** 0.5, 1e-9)

d = signed_distance(p, a, b)
if abs(d) >= HYSTERESIS_PX:
    side = 1 if d > 0 else -1
    if prev_side[tid] not in (None, side) and projects_onto_segment(p, a, b):
        direction = "in" if side > 0 else "out"

The order of the line’s end points decides the direction. If “in” and “out” come out reversed in the first test, just swap A and B. Keeping coordinates normalised to 0–1 saves you from redrawing the line when the camera moves from sub-stream to main stream. Measuring how accurate the count really is on site is a separate job: run the acceptance test against real crossings counted by hand.

Step 5: Zone (polygon) intrusion with a time rule

For the zone rule, OpenCV’s pointPolygonTest does the work: with the third argument set to False it returns a positive value inside, zero on the edge and a negative value outside.

inside = cv2.pointPolygonTest(poly, (float(x), float(y)), False) >= 0
if inside:
    t0 = entered_at.setdefault(tid, now)
    if tid not in alerted and now - t0 >= ZONE_MIN_SECONDS:
        alerted.add(tid)                     # one event per entry
        emit(make_event("zone.intrusion", ...))
else:
    entered_at.pop(tid, None); alerted.discard(tid)

The time threshold does two jobs: it filters out people brushing past the edge of the zone and single-frame false detections. As long as the same person stays inside, only one event is produced; if they leave and come back, the timer starts again.

Step 6: Send the event as JSON to MQTT or a webhook

The event schema is the same one we use when combining IoT and camera analytics on a single event bus; I only added an event_id so that repeated deliveries can be filtered out:

{
  "event_id": "5f0c2a4e-8a51-4f0e-9d5c-2f1b7f3c9e10",
  "source": "cam-07",
  "type": "line.cross",
  "ts": "2026-09-25T09:14:03.120Z",
  "site": "gebze-01",
  "zone": "entrance",
  "confidence": 0.912,
  "rule": "line-entrance",
  "payload": {"direction": "in", "track_id": 42, "object": "person", "count_in": 118, "count_out": 97}
}

The topic follows cx/{site}/{camera}/{event type}. In paho-mqtt 2.x you must state the callback API version when creating the client; with CallbackAPIVersion.VERSION2, on_connect has the signature (client, userdata, flags, reason_code, properties) and reason_code is an object. I check whether the connection was refused with reason_code.is_failure.

Three design decisions:

• The inference loop never waits on the network. Events go into a queue; a separate thread publishes to MQTT and calls the webhook with a timeout and retries. When the webhook server slows down, the camera loop does not.
• QoS 1 with a memory cap. While disconnected, paho keeps QoS 1 messages in memory and sends them once it reconnects. max_queued_messages_set(1000) puts an upper bound on that queue. Because QoS 1 means “at least once”, consumers should deduplicate on event_id.
• A status topic. A retained online message is written to cx/{site}/{camera}/status, and the Last Will message is offline. A dashboard can see from there whether a camera’s analytics service is alive.

This setup should not go into the field without TLS, client authentication and per-topic authorisation on the broker; I cover the details in secure IoT with MQTT.

Step 7: Frame skipping and the GPU budget

The script picks frames based on time: with PROCESS_FPS=8 it processes at most one frame every 125 milliseconds and skips whatever arrives in between. Looking at the clock rather than “every Nth frame” gives consistent results even when the camera’s frame rate changes or the stream stutters.

The budget is a simple multiplication. Processing 12 cameras at 8 frames per second means 96 inferences per second; if they run one after another on a single GPU, that leaves roughly 10.4 milliseconds per frame. If the model takes longer than that, these are the levers:

Lever	Effect	Cost
Lower PROCESS_FPS	Linear gain	Fast events may be missed; revisit track_buffer
Sub-stream / smaller imgsz	Cheaper decoding and inference	Lower accuracy on small, distant objects
Smaller model scale	Faster inference	Accuracy must be measured on site
TensorRT FP16/INT8	Lower latency with the same model	Adds a build and validation step
Batching cameras together	Better GPU utilisation	Higher per-camera latency, more complex code

I cover TensorRT builds and INT8 in detail in ONNX, TensorRT and INT8. The script logs the processed frame rate, the reconnect count and the queue length once a minute; that line is where you will first notice falling below the target rate.

The full script

The script below combines the pieces above. Settings are read from environment variables; the line and the zone are defined at the top of the file in 0–1 coordinates.

#!/usr/bin/env python3
"""RTSP camera -> YOLO + ByteTrack -> line crossing / zone intrusion -> JSON event (MQTT and/or webhook).

No frame is written to disk and no image is attached to events. Settings come from environment variables.
"""
from __future__ import annotations

import json
import logging
import os
import queue
import signal
import threading
import time
import urllib.request
import uuid
from datetime import datetime, timezone

# Tell the FFmpeg backend to open RTSP over TCP (must be set before cv2 is imported).
os.environ.setdefault("OPENCV_FFMPEG_CAPTURE_OPTIONS", "rtsp_transport;tcp")

import cv2  # noqa: E402
import numpy as np  # noqa: E402
import paho.mqtt.client as mqtt  # noqa: E402
from ultralytics import YOLO  # noqa: E402

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("cam")

# --- Settings ----------------------------------------------------------------
RTSP_URL = os.environ["RTSP_URL"]  # rtsp://user:password@10.0.0.21:554/stream2
SITE = os.getenv("SITE", "gebze-01")
CAMERA_ID = os.getenv("CAMERA_ID", "cam-07")
MODEL_PATH = os.getenv("MODEL_PATH", "yolo26n.pt")  # a TensorRT .engine file also works
DEVICE = os.getenv("DEVICE", "0")  # "0" = first GPU, "cpu" = processor
PROCESS_FPS = float(os.getenv("PROCESS_FPS", "8"))
CONF = float(os.getenv("CONF", "0.35"))
CLASSES = [0]  # COCO: 0 = person
ANCHOR = os.getenv("ANCHOR", "bottom")  # "bottom" = foot point, "center" = box centre

# Geometry in normalised 0-1 coordinates: stays valid if the resolution changes.
LINE = ((0.10, 0.60), (0.90, 0.60))  # counting line A -> B
ZONE = [(0.60, 0.20), (0.95, 0.20), (0.95, 0.55), (0.60, 0.55)]  # restricted zone
ZONE_MIN_SECONDS = 5.0  # emit an event for tracks that stay longer than this
HYSTERESIS_PX = 4.0  # points closer than this to the line never count as a side change

MQTT_HOST = os.getenv("MQTT_HOST")  # MQTT disabled if empty
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_CA = os.getenv("MQTT_CA")  # e.g. /etc/cx/ca.crt; enables TLS when set
MQTT_USER = os.getenv("MQTT_USER")
MQTT_PASS = os.getenv("MQTT_PASS")
WEBHOOK_URL = os.getenv("WEBHOOK_URL")  # webhook disabled if empty

TOPIC_BASE = f"cx/{SITE}/{CAMERA_ID}"


# --- 1) RTSP reader: separate thread, latest frame only -------------------------
class LatestFrameReader(threading.Thread):
    """Reads the stream continuously, keeps only the newest frame, reconnects with exponential backoff."""

    def __init__(self, url: str, open_timeout_ms: int = 5000, read_timeout_ms: int = 5000):
        super().__init__(daemon=True, name="rtsp-reader")
        self.url = url
        self.params = [
            cv2.CAP_PROP_OPEN_TIMEOUT_MSEC, open_timeout_ms,
            cv2.CAP_PROP_READ_TIMEOUT_MSEC, read_timeout_ms,
        ]
        self._lock = threading.Lock()
        self._frame: np.ndarray | None = None
        self._seq = 0
        self._stop_evt = threading.Event()
        self.reconnects = 0

    def run(self) -> None:
        backoff = 1.0
        while not self._stop_evt.is_set():
            cap = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG, self.params)
            if not cap.isOpened():
                log.warning("Could not open stream; retrying in %.0f s", backoff)
                self._stop_evt.wait(backoff)
                backoff = min(backoff * 2, 30.0)
                continue
            log.info("Stream opened")
            backoff = 1.0
            while not self._stop_evt.is_set():
                ok, frame = cap.read()
                if not ok:
                    log.warning("Frame read failed; reconnecting")
                    break
                with self._lock:
                    self._frame = frame
                    self._seq += 1
            cap.release()
            self.reconnects += 1

    def latest(self) -> tuple[int, np.ndarray | None]:
        with self._lock:
            return self._seq, self._frame

    def stop(self) -> None:
        self._stop_evt.set()


# --- 2) Rules: line crossing and zone intrusion ---------------------------------
def signed_distance(p, a, b) -> float:
    """Signed distance (pixels) from p to line AB. The sign tells which side p is on."""
    abx, aby = b[0] - a[0], b[1] - a[1]
    cross = abx * (p[1] - a[1]) - aby * (p[0] - a[0])
    return cross / max((abx * abx + aby * aby) ** 0.5, 1e-9)


def projects_onto_segment(p, a, b) -> bool:
    """Does the projection of p fall within segment AB (not just the infinite line)?"""
    abx, aby = b[0] - a[0], b[1] - a[1]
    t = ((p[0] - a[0]) * abx + (p[1] - a[1]) * aby) / max(abx * abx + aby * aby, 1e-9)
    return 0.0 <= t <= 1.0


class LineCounter:
    def __init__(self, a, b, hysteresis_px: float = HYSTERESIS_PX, cooldown_s: float = 1.0):
        self.a, self.b = a, b
        self.h = hysteresis_px
        self.cooldown_s = cooldown_s
        self.last_side: dict[int, int] = {}
        self.last_cross: dict[int, float] = {}
        self.count_in = 0
        self.count_out = 0

    def update(self, tid: int, p, now: float) -> str | None:
        d = signed_distance(p, self.a, self.b)
        if abs(d) < self.h:  # ignore jitter right on the line
            return None
        side = 1 if d > 0 else -1
        prev = self.last_side.get(tid)
        self.last_side[tid] = side
        if prev is None or prev == side or not projects_onto_segment(p, self.a, self.b):
            return None
        if now - self.last_cross.get(tid, 0.0) < self.cooldown_s:
            return None
        self.last_cross[tid] = now
        if prev < 0 < side:
            self.count_in += 1
            return "in"
        self.count_out += 1
        return "out"

    def forget(self, tid: int) -> None:
        self.last_side.pop(tid, None)
        self.last_cross.pop(tid, None)


class ZoneWatcher:
    def __init__(self, polygon_px: np.ndarray, min_seconds: float):
        self.poly = polygon_px.reshape(-1, 1, 2).astype(np.int32)
        self.min_seconds = min_seconds
        self.entered_at: dict[int, float] = {}
        self.alerted: set[int] = set()

    def update(self, tid: int, p, now: float) -> float | None:
        inside = cv2.pointPolygonTest(self.poly, (float(p[0]), float(p[1])), False) >= 0
        if not inside:
            self.forget(tid)
            return None
        t0 = self.entered_at.setdefault(tid, now)
        if tid not in self.alerted and now - t0 >= self.min_seconds:
            self.alerted.add(tid)
            return now - t0
        return None

    def forget(self, tid: int) -> None:
        self.entered_at.pop(tid, None)
        self.alerted.discard(tid)


# --- 3) Event output: queue + sender thread -------------------------------------
def make_event(etype: str, rule: str, zone: str, confidence: float, payload: dict) -> dict:
    return {
        "event_id": str(uuid.uuid4()),
        "source": CAMERA_ID,
        "type": etype,
        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z"),
        "site": SITE,
        "zone": zone,
        "confidence": round(confidence, 3),
        "rule": rule,
        "payload": payload,
    }


def build_mqtt() -> mqtt.Client | None:
    if not MQTT_HOST:
        return None
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{SITE}-{CAMERA_ID}",
                         protocol=mqtt.MQTTv5)
    if MQTT_CA:
        client.tls_set(ca_certs=MQTT_CA)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.will_set(f"{TOPIC_BASE}/status", "offline", qos=1, retain=True)
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.max_queued_messages_set(1000)  # max messages held in memory while disconnected

    def on_connect(c, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.error("MQTT connection refused: %s", reason_code)
            return
        log.info("MQTT connected")
        c.publish(f"{TOPIC_BASE}/status", "online", qos=1, retain=True)

    def on_disconnect(c, userdata, flags, reason_code, properties):
        log.warning("MQTT disconnected: %s", reason_code)

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()  # network loop and automatic reconnect in the background
    return client


def post_webhook(body: bytes, attempts: int = 3) -> None:
    for i in range(attempts):
        try:
            req = urllib.request.Request(WEBHOOK_URL, data=body, method="POST",
                                         headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=3) as resp:
                if 200 <= resp.status < 300:
                    return
        except OSError as exc:
            log.warning("Webhook attempt %d failed: %s", i + 1, exc)
        time.sleep(2 ** i)
    log.error("Webhook delivery failed, event dropped")


def sender_loop(events: queue.Queue, client: mqtt.Client | None, stop: threading.Event) -> None:
    while not stop.is_set() or not events.empty():
        try:
            event = events.get(timeout=0.5)
        except queue.Empty:
            continue
        body = json.dumps(event, ensure_ascii=False).encode("utf-8")
        if client is not None:
            client.publish(f"{TOPIC_BASE}/{event['type']}", body, qos=1)
        if WEBHOOK_URL:
            post_webhook(body)


# --- 4) Main loop ---------------------------------------------------------------
def to_px(points, w: int, h: int) -> list[tuple[float, float]]:
    return [(x * w, y * h) for x, y in points]


def anchor_point(xyxy) -> tuple[float, float]:
    x1, y1, x2, y2 = (float(v) for v in xyxy)
    return ((x1 + x2) / 2, y2) if ANCHOR == "bottom" else ((x1 + x2) / 2, (y1 + y2) / 2)


def main() -> None:
    stop = threading.Event()
    signal.signal(signal.SIGTERM, lambda *_: stop.set())
    signal.signal(signal.SIGINT, lambda *_: stop.set())

    model = YOLO(MODEL_PATH)  # one model object per camera: tracker state lives in it
    reader = LatestFrameReader(RTSP_URL)
    reader.start()

    events: queue.Queue = queue.Queue(maxsize=1000)
    client = build_mqtt()
    sender = threading.Thread(target=sender_loop, args=(events, client, stop), daemon=True)
    sender.start()

    def emit(event: dict) -> None:
        try:
            events.put_nowait(event)
        except queue.Full:
            log.error("Event queue full; event dropped: %s", event["type"])

    line = zone = None
    last_seen: dict[int, float] = {}
    period = 1.0 / PROCESS_FPS
    next_run = time.monotonic()
    last_seq = -1
    processed, stats_t0 = 0, time.monotonic()

    while not stop.is_set():
        now = time.monotonic()
        if now < next_run:
            time.sleep(min(next_run - now, 0.01))
            continue
        seq, frame = reader.latest()
        if frame is None or seq == last_seq:  # no new frame
            time.sleep(0.005)
            continue
        last_seq, next_run = seq, now + period  # frame skipping: at most PROCESS_FPS frames per second

        if line is None:  # convert geometry to pixels using the first frame's size
            h, w = frame.shape[:2]
            a, b = to_px(LINE, w, h)
            line = LineCounter(a, b)
            zone = ZoneWatcher(np.array(to_px(ZONE, w, h)), ZONE_MIN_SECONDS)

        result = model.track(frame, persist=True, tracker="bytetrack.yaml", classes=CLASSES,
                             conf=CONF, device=DEVICE, verbose=False)[0]
        boxes = result.boxes
        if boxes and boxes.is_track:  # skip when there are no track IDs (empty frame)
            ids = boxes.id.int().cpu().tolist()
            confs = boxes.conf.cpu().tolist()
            for tid, xyxy, conf in zip(ids, boxes.xyxy.cpu().tolist(), confs):
                last_seen[tid] = now
                p = anchor_point(xyxy)
                direction = line.update(tid, p, now)
                if direction:
                    emit(make_event("line.cross", "line-entrance", "entrance", conf,
                                    {"direction": direction, "track_id": tid, "object": "person",
                                     "count_in": line.count_in, "count_out": line.count_out}))
                dwell = zone.update(tid, p, now)
                if dwell is not None:
                    emit(make_event("zone.intrusion", "zone-restricted", "restricted", conf,
                                    {"duration_s": round(dwell, 1), "track_id": tid,
                                     "object": "person"}))

        for tid in [t for t, ts in last_seen.items() if now - ts > 10.0]:  # forget stale tracks
            last_seen.pop(tid)
            line.forget(tid)
            zone.forget(tid)

        processed += 1
        if now - stats_t0 >= 60:
            log.info("processed_fps=%.1f reconnects=%d queue=%d",
                     processed / (now - stats_t0), reader.reconnects, events.qsize())
            processed, stats_t0 = 0, now

    reader.stop()
    sender.join(timeout=5)
    if client is not None:  # a clean shutdown does not fire the LWT; publish the status ourselves
        try:
            client.publish(f"{TOPIC_BASE}/status", "offline", qos=1, retain=True).wait_for_publish(3)
        except (RuntimeError, ValueError):
            pass
        client.disconnect()
        client.loop_stop()


if __name__ == "__main__":
    main()

To run it:

export RTSP_URL='rtsp://<user>:<password>@10.0.0.21:554/stream2'
export MQTT_HOST=broker.local MQTT_CA=/etc/cx/ca.crt MQTT_USER=cam-07 MQTT_PASS='<password>'
export MODEL_PATH=yolo26n.pt PROCESS_FPS=8
python rtsp_yolo_events.py

What to watch when going to production

KVKK and not storing images. KVKK is Turkey’s Personal Data Protection Law No. 6698, broadly comparable to the GDPR, and camera images count as personal data under it. The script writes no frames to disk, attaches no image to events, and logs only numbers. During development you will want to see annotated frames on screen; do that only in a test environment, in areas where people have been properly informed, and keep it out of the production image. I cover the legal side in GDPR and video analytics without face recognition.

Time. Event timestamps are UTC and come from the server clock. If the server is not synchronised with NTP, you cannot match events against records from other systems.

Process management. The script catches SIGTERM, shuts down cleanly and writes offline to the status topic, so it is ready to run under systemd or in a container. If it crashes unexpectedly, the broker publishes the same information through the Last Will message.

One process per camera. A separate process per camera isolates failures and keeps tracker states apart; the price is that each process loads the model into GPU memory separately. As the camera count grows, measure memory and, if needed, move to batching several cameras in one process.

Accuracy. The off-the-shelf COCO model is a good start for the “person” class, but it should not be accepted before it is measured under your site’s lighting, camera angles and night mode.

Checklist

• The licence decision is made (AGPL-3.0 compliance or an Enterprise License).
• FFmpeg support in the OpenCV build is verified on the target machine.
• RTSP is forced to TCP, open/read timeouts are set, reconnection is tested (by pulling the camera cable).
• Reading runs in a separate thread and only the latest frame is kept.
• The tracker is set explicitly and track_buffer matches the processed frame rate.
• Line direction and zone boundaries are verified with real crossings on site.
• The event schema is fixed; consumers deduplicate on event_id.
• The MQTT connection uses TLS and authentication, with restricted topic permissions.
• No frames are written to disk; no images or passwords appear in events or logs.
• Processed frame rate, reconnects and queue length are monitored.