--- title: "Real-time object detection with Python and YOLO: RTSP camera to event" author: "Abdulaziz Akyol" author_url: https://www.abdulazizakyol.com/en/about/ url: https://www.abdulazizakyol.com/en/blog/real-time-object-detection-with-python-and-yolo-from-rtsp-camera-to-event/ language: en published: 2026-09-25 categories: ["Computer vision", "Software development"] tags: ["YOLO", "Python", "OpenCV", "RTSP", "ByteTrack", "MQTT", "Ultralytics"] translation: https://www.abdulazizakyol.com/blog/python-ve-yolo-ile-gercek-zamanli-nesne-tespiti-rtsp-kameradan-olaya/ description: "Real-time detection from an RTSP camera with Python, OpenCV and YOLO: ByteTrack tracking, line counts, zone alerts and JSON events to MQTT, full script." --- # Real-time object detection with Python and YOLO: RTSP camera to event By [Abdulaziz Akyol](https://www.abdulazizakyol.com/en/about/) · 2026-09-25 ## Key takeaways - Reading the RTSP stream in a separate thread and keeping only the newest frame prevents buffer-induced latency from piling up and makes reconnecting with exponential backoff straightforward. - In Ultralytics YOLO, model.track(persist=True, tracker="bytetrack.yaml") gives every object an identity that survives across frames; counting and dwell-time rules operate on those identities, not on raw detections. - A line crossing is a track ID moving from one side of a line segment to the other; a zone intrusion is the foot point staying inside a polygon for longer than a set time. - An event is a small JSON document with camera ID, UTC timestamp, rule and confidence score, carries no image and is published to MQTT with QoS 1. - Ultralytics YOLO is licensed under AGPL-3.0; commercial use without open-sourcing the whole project requires an Ultralytics Enterprise License. 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](https://www.abdulazizakyol.com/en/blog/video-analytics-on-existing-cameras-from-rtsp-to-events/). 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](https://www.ultralytics.com/license), 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. ```bash 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: ```python 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. ```python 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. ```python 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. ```python 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: ```json { "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](https://www.abdulazizakyol.com/en/blog/secure-iot-with-mqtt-python-tls-and-access-control/). ## 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](https://www.abdulazizakyol.com/en/blog/accelerating-edge-ai-models-onnx-tensorrt-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. ```python #!/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: ```bash export RTSP_URL='rtsp://:@10.0.0.21:554/stream2' export MQTT_HOST=broker.local MQTT_CA=/etc/cx/ca.crt MQTT_USER=cam-07 MQTT_PASS='' 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](https://www.abdulazizakyol.com/en/blog/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. ## Frequently asked questions ### How do you read an RTSP camera stream in Python? Open it with OpenCV's cv2.VideoCapture(url, cv2.CAP_FFMPEG). Force TCP with OPENCV_FFMPEG_CAPTURE_OPTIONS="rtsp_transport;tcp" and pass open and read timeouts with CAP_PROP_OPEN_TIMEOUT_MSEC and CAP_PROP_READ_TIMEOUT_MSEC. To avoid accumulating latency, read in a separate thread and keep only the latest frame. ### How do you track objects with YOLO? In Ultralytics, call model.track(frame, persist=True, tracker="bytetrack.yaml") on every frame. persist=True keeps the tracker state from previous frames, and track IDs are read from results[0].boxes.id. Set the tracker explicitly, because the default can change between releases. ### How do you count people crossing a line with YOLO? For each track ID, compute which side of the line the foot point is on from the sign of a cross product. When the sign flips and the point projects onto the line segment, count a crossing. Ignoring points within a few pixels of the line and adding a short per-ID cooldown prevents double counts caused by jitter. ### Can Ultralytics YOLO be used for free in a commercial project? Ultralytics YOLO is licensed under AGPL-3.0, which requires releasing the source code of the entire derivative work, and according to Ultralytics trained models fall under the same licence by default. Using it in a closed-source product, SaaS or edge device requires an Enterprise License. ### How can you reduce GPU load in real-time object detection? Decode every frame but do not infer on every frame: 8 to 15 processed frames per second is enough for most rules. Camera sub-streams, a smaller input size, a smaller model scale and compiling with TensorRT in FP16 or INT8 are the other levers. ## Sources 1. [Ultralytics Docs — Multi-Object Tracking (Track mode)](https://docs.ultralytics.com/modes/track/) 2. [Ultralytics — Licensing (AGPL-3.0 and Enterprise)](https://www.ultralytics.com/license) 3. [OpenCV — cv::VideoCapture Class Reference](https://docs.opencv.org/4.x/d8/dfe/classcv_1_1VideoCapture.html) 4. [Eclipse Paho MQTT Python Client — Migrations (2.0)](https://eclipse.dev/paho/files/paho.mqtt.python/html/migrations.html) 5. [Law No. 6698 on the Protection of Personal Data (KVKK), official text](https://www.mevzuat.gov.tr/mevzuatmetin/1.5.6698.pdf) --- Abdulaziz Akyol is the founder of CX Teknoloji (AI, computer vision and IoT). Canonical version: https://www.abdulazizakyol.com/en/blog/real-time-object-detection-with-python-and-yolo-from-rtsp-camera-to-event/