Python ve YOLO ile gerçek zamanlı nesne tespiti: RTSP kameradan olaya

Gerçek zamanlı nesne tespiti, bir kamera akışındaki karelerde nesnelerin sınıfını ve konumunu milisaniyeler içinde bulan işlemdir; tek başına bir sonuç değil, “kaç kişi girdi” ya da “yasak alanda biri var” gibi olaylara giden zincirin ortasındaki halkadır. Bu yazıda Python, OpenCV ve Ultralytics YOLO ile RTSP kameradan MQTT’ye giden zinciri çalışır bir betikle, parça parça kuruyoruz.

Zincirin mimarisini ve sahada nerede tıkandığını RTSP akışından olaya giden zincir yazısında anlattım. Burada aynı zinciri koda döküyorum: hangi satır neden orada, hangi varsayılan sizi sahada yanıltır.

Ne kuracağız?

Betiğin yaptığı iş beş adımdan oluşuyor:

1. RTSP akışını ayrı bir iş parçacığında okumak ve yalnızca en son kareyi tutmak.
2. Seçilen karelerde YOLO ile kişi tespiti yapmak.
3. ByteTrack ile her kişiye kareler boyunca korunan bir iz kimliği vermek.
4. İzler üzerinde iki kural çalıştırmak: çizgi geçişi (giriş/çıkış sayımı) ve bölge ihlali (yasak alanda belirli süreden uzun kalma).
5. Kural tetiklenince görüntü içermeyen bir JSON olayı MQTT’ye ve istenirse bir webhook’a göndermek.

Bileşen	Kütüphane	Görev
Akış okuma	OpenCV (FFmpeg arka ucu)	RTSP’yi açmak, kareyi çözmek, yeniden bağlanmak
Tespit ve takip	Ultralytics YOLO + ByteTrack	Kutular ve iz kimlikleri
Kurallar	Saf Python + cv2.pointPolygonTest	Çizgi ve poligon mantığı
Olay çıkışı	paho-mqtt 2.x, urllib	MQTT yayını ve webhook

Kodu yazmadan önce: lisans

Ultralytics YOLO AGPL-3.0 lisansıyla dağıtılıyor. Ultralytics’in lisans sayfasına göre bu lisansla uyum, uygulamanın tamamı dahil türetilmiş işin kaynak kodunu yayımlamak anlamına geliyor ve eğitilmiş modeller de varsayılan olarak AGPL-3.0’a tabi. Kapalı kaynak bir ürün, SaaS ya da uç cihaza gömülü bir sistem geliştiriyorsanız kurumsal lisans gerekiyor.

Benim önerim, lisans kararını ilk hafta vermek. Prototip AGPL ile rahatça yapılır; ama müşteriye teslim edilecek sistemde lisans sonradan fark edilirse ya kaynak kodu açmak ya da modeli baştan değiştirmek gerekir. Başka bir dedektör seçerseniz kodun lisansıyla ağırlıkların lisansını ayrı ayrı kontrol edin.

Adım 1: Ortamı kurmak

Python 3.10 ve üstü yeterli. ultralytics paketi PyTorch’u ve OpenCV’yi bağımlılık olarak getiriyor; MQTT için paho-mqtt ekliyoruz.

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])"

Son satır önemli: OpenCV’nin RTSP’yi FFmpeg arka ucuyla açabilmesi için derlemede FFmpeg desteğinin YES görünmesi gerekir. Her platformun hazır paketinde bu destek olmayabilir; sahaya çıkmadan önce hedef makinede kontrol edin.

Adım 2: RTSP akışını güvenilir okumak

İlk karar ana akış mı alt akış mı. Kişi sayma ve bölge ihlali gibi işler genellikle alt akışla çalışır; hem ağ trafiği hem kod çözme yükü düşer.

İkinci karar taşıma protokolü. OpenCV’nin FFmpeg arka ucu, OPENCV_FFMPEG_CAPTURE_OPTIONS ortam değişkenini anahtar;değer|anahtar;değer biçiminde okur. Değişken boşsa TCP’yi tercih eden bir bayrak kullanıyor; rtsp_transport;tcp yazmak davranışı açık ve sürümden bağımsız kılıyor. Değişkeni cv2 içe aktarılmadan önce ayarlayın.

Üçüncüsü zaman aşımı. Kamera kapandığında read() çağrısının ne kadar bekleyeceğini CAP_PROP_OPEN_TIMEOUT_MSEC ve CAP_PROP_READ_TIMEOUT_MSEC belirler; bu iki özellik yalnızca FFmpeg ve GStreamer arka uçlarında geçerli ve açılışta parametre olarak verilmesi gerekiyor. Verilmezse OpenCV kaynak kodundaki varsayılan 30 saniye. Sahada 30 saniyelik sessiz bir bekleme, “sistem dondu” şikâyeti demektir.

En sık yapılan hata gecikmeyle ilgili. cap.read() kareleri sırayla verir; işlemeniz kameranın kare hızından yavaşsa aradaki fark birikir ve bir süre sonra birkaç saniye önceki görüntüyü işliyor olursunuz. CAP_PROP_BUFFERSIZE burada çözüm değil: OpenCV kaynak kodunda FFmpeg arka ucu bu özelliği hiç işlemiyor, özellik V4L2 gibi yerel kamera arka uçları için var. Çözüm, okumayı ayrı bir iş parçacığına almak ve yalnızca son kareyi saklamak:

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                      # dış döngü üstel beklemeyle yeniden bağlanır
    with lock:
        latest_frame, seq = frame, seq + 1

H.264 ve H.265’te kareler birbirine bağımlı olduğu için okuyucu her kareyi çözmek zorunda; ama çıkarım yalnızca son kare üzerinde yapılır. Bağlantı koptuğunda okuyucu 1 saniyeden başlayıp 30 saniyeye kadar ikiye katlanan aralıklarla yeniden dener. RTSP adresindeki kullanıcı adı ve parolayı koda yazmayın, ortam değişkeninden okuyun ve adresi loglara basmayın.

Adım 3: YOLO ile tespit, ByteTrack ile takip

Ultralytics’in güncel model ailesi YOLO26; n, s, m, l, x ölçekleri var. Sahada genellikle en küçük ölçekle başlayıp doğruluk yetmiyorsa büyütmek doğru sıra.

from ultralytics import YOLO

model = YOLO("yolo26n.pt")         # her kamera için ayrı model nesnesi
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()

Dört ayrıntı önemli:

• persist=True, izleyiciye bu karenin aynı akışın bir sonraki karesi olduğunu söyler. Ultralytics dokümanı bunun ilgisiz görüntülerde ya da farklı akışlarda kullanılmamasını açıkça belirtiyor; bu yüzden birden fazla kamerayı tek model nesnesinden geçirmeyin, izleyici durumu karışır.
• İzleyiciyi açıkça yazın. ultralytics 8.4.63 ile yeni izleyiciler geldi ve tracker verilmediğinde kullanılan varsayılan artık TrackTrack. Sürüm yükseltmesinde davranışın sessizce değişmesini istemiyorsanız bytetrack.yaml gibi sabit bir değer verin.
• Karede iz yoksa boxes.id boş gelir; is_track kontrolü bu durumu yakalar.
• bytetrack.yaml içindeki track_buffer (varsayılan 30), kaybolan bir izin kaç kare boyunca bekletileceğidir ve izleyiciye verilen kareler üzerinden sayılır. Saniyede 8 kare işliyorsanız 30 kare yaklaşık 3,75 saniyeye denk gelir. Kare atlama oranını değiştirdiğinizde bu değeri de gözden geçirin.

Adım 4: Çizgi geçişi ile sayma

Sayımın temeli basit geometri: noktanın bir doğrunun hangi tarafında olduğu, çapraz çarpımın işaretiyle bulunur. Bir iz kimliğinin işareti değiştiğinde çizgi geçilmiş demektir. Pratikte üç ek koşul gerekiyor:

• Hangi nokta? Eğik bakan kamerada kutunun alt ortası (ayak noktası), tepeden bakan kamerada kutunun merkezi. Kutunun merkezi eğik açıda kişinin boyuna göre kayar.
• Doğru parçası, sonsuz doğru değil. Noktanın izdüşümü çizginin iki ucu arasında değilse geçiş sayılmaz; yoksa kapının yanındaki koridordan geçen de sayılır.
• Titreşim. Kutu her karede birkaç piksel oynar. Çizgiye HYSTERESIS_PX mesafesinden yakın noktaları yok saymak ve aynı kimlik için 1 saniyelik bekleme koymak çift sayımı keser.

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"

Yönü, çizginin uç noktalarının sırası belirler. İlk testte giriş ile çıkış ters çıkarsa A ile B’yi yer değiştirmeniz yeterli. Koordinatları 0–1 aralığında (normalize) tutmak, kamera alt akıştan ana akışa geçtiğinde çizgiyi yeniden çizme zahmetini ortadan kaldırıyor. Sayımın sahada ne kadar doğru olduğunu ölçmek ayrı bir iş: kabul testini, elle sayılmış gerçek geçişlere karşı yapın.

Adım 5: Bölge (poligon) ihlali ve süre kuralı

Bölge kuralında OpenCV’nin pointPolygonTest fonksiyonu işi görüyor: üçüncü parametre False iken içerideyse pozitif, kenardaysa sıfır, dışarıdaysa negatif döner.

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)                     # aynı giriş için tek olay
        emit(make_event("zone.intrusion", ...))
else:
    entered_at.pop(tid, None); alerted.discard(tid)

Süre eşiği iki işe yarıyor: bölgenin kenarından geçen kişiyi ve tek karelik yanlış tespitleri eliyor. Aynı kişi bölgede kaldığı sürece tek olay üretilir; çıkıp yeniden girerse sayaç sıfırdan başlar.

Adım 6: Olayı JSON olarak MQTT’ye ya da webhook’a göndermek

Olay şeması, IoT ve kamera analitiğini aynı olay hattında birleştirdiğimiz mimarideki şemanın aynısı; yalnızca tekrar eden teslimatları ayıklamak için bir event_id ekledim:

{
  "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}
}

Konu cx/{site}/{kamera}/{olay tipi} biçiminde. paho-mqtt 2.x’te istemciyi oluştururken geri çağırma API sürümünü belirtmek gerekiyor; CallbackAPIVersion.VERSION2 ile on_connect imzası (client, userdata, flags, reason_code, properties) oluyor ve reason_code bir nesne. Bağlantının reddedilip reddedilmediğini reason_code.is_failure ile kontrol ediyorum.

Üç tasarım kararı var:

• Çıkarım döngüsü ağı beklemez. Olaylar bir kuyruğa girer, ayrı bir iş parçacığı MQTT’ye yayınlar ve webhook’u zaman aşımı ile yeniden denemeli çağırır. Webhook sunucusu yavaşladığında kamera döngüsü yavaşlamaz.
• QoS 1 ve bellek sınırı. paho, bağlantı yokken QoS 1 mesajları bellekte tutar ve bağlanınca gönderir. max_queued_messages_set(1000) bu kuyruğa üst sınır koyar. QoS 1 “en az bir kez” teslim ettiği için tüketici tarafı event_id ile tekrarları ayıklamalı.
• Durum konusu. cx/{site}/{kamera}/status konusuna retain bayraklı online yazılır; son vasiyet (Last Will) mesajı offline. Panel, kameranın analitik servisinin ayakta olup olmadığını buradan görür.

Broker tarafında TLS, kullanıcı doğrulama ve konu bazlı yetki olmadan bu yapı sahaya çıkmamalı; ayrıntıları MQTT ile güvenli IoT yazısında anlattım.

Adım 7: Kare atlama ve GPU bütçesi

Betik kareleri zaman tabanlı seçiyor: PROCESS_FPS=8 ise en fazla 125 milisaniyede bir kare işleniyor, arada gelenler atlanıyor. “Her N. kare” yerine zamana bakmak, kamera kare hızı değiştiğinde ya da akış takıldığında da tutarlı sonuç veriyor.

Bütçe hesabı basit bir çarpım. 12 kamerayı saniyede 8 kareyle işlemek saniyede 96 çıkarım demek; tek GPU’da sırayla çalışıyorsa kare başına yaklaşık 10,4 milisaniyelik bütçe kalıyor. Model bu sürenin üstündeyse kaldıraçlar şunlar:

Kaldıraç	Etkisi	Bedeli
PROCESS_FPS düşürmek	Doğrusal kazanç	Hızlı olaylar kaçabilir, track_buffer gözden geçirilmeli
Alt akış / küçük imgsz	Çözme ve çıkarım ucuzlar	Uzak ve küçük nesnelerde doğruluk düşer
Küçük model ölçeği	Çıkarım hızlanır	Doğruluk sahada ölçülmeli
TensorRT FP16/INT8	Aynı modelle daha düşük gecikme	Derleme ve doğrulama adımı eklenir
Kameraları toplu (batch) işlemek	GPU daha verimli kullanılır	Tek kamera gecikmesi artar, kod karmaşıklaşır

TensorRT derleme ve INT8 tarafını ONNX, TensorRT ve INT8 yazısında ayrıntılı işledim. Betik, dakikada bir işlenen kare hızını, yeniden bağlanma sayısını ve kuyruk uzunluğunu logluyor; hedef hızın altına düşüldüğünü ilk bu satırdan görürsünüz.

Tam betik

Aşağıdaki betik yukarıdaki parçaları birleştiriyor. Ayarlar ortam değişkenlerinden okunuyor; çizgi ve bölge dosyanın başında 0–1 koordinatlarıyla tanımlı.

#!/usr/bin/env python3
"""RTSP kamera -> YOLO + ByteTrack -> çizgi geçişi / bölge ihlali -> JSON olay (MQTT ve/veya webhook).

Kare diske yazılmaz, olaylara görüntü eklenmez. Ayarlar ortam değişkenlerinden okunur.
"""
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

# FFmpeg arka ucuna RTSP'yi TCP üzerinden açmasını söyle (cv2 içe aktarılmadan önce ayarlanmalı).
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")

# --- Ayarlar -----------------------------------------------------------------
RTSP_URL = os.environ["RTSP_URL"]  # rtsp://kullanici:parola@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")  # TensorRT için .engine dosyası da olur
DEVICE = os.getenv("DEVICE", "0")  # "0" = ilk GPU, "cpu" = işlemci
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" = ayak noktası, "center" = kutu merkezi

# Geometri 0-1 arası (normalize) koordinatlarla: çözünürlük değişse de geçerli kalır.
LINE = ((0.10, 0.60), (0.90, 0.60))  # sayım çizgisi A -> B
ZONE = [(0.60, 0.20), (0.95, 0.20), (0.95, 0.55), (0.60, 0.55)]  # yasak bölge
ZONE_MIN_SECONDS = 5.0  # bölgede bu süreyi aşan iz için olay üret
HYSTERESIS_PX = 4.0  # çizgiye bu mesafeden yakın noktalar yan değiştirmiş sayılmaz

MQTT_HOST = os.getenv("MQTT_HOST")  # boşsa MQTT kapalı
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_CA = os.getenv("MQTT_CA")  # örn. /etc/cx/ca.crt; tanımlıysa TLS açılır
MQTT_USER = os.getenv("MQTT_USER")
MQTT_PASS = os.getenv("MQTT_PASS")
WEBHOOK_URL = os.getenv("WEBHOOK_URL")  # boşsa webhook kapalı

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


# --- 1) RTSP okuyucu: ayrı iş parçacığı, yalnızca son kare ------------------------
class LatestFrameReader(threading.Thread):
    """Akışı sürekli okur, yalnızca en son kareyi tutar; kopunca üstel bekleme ile yeniden bağlanır."""

    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("Akış açılamadı; %.0f sn sonra tekrar denenecek", backoff)
                self._stop_evt.wait(backoff)
                backoff = min(backoff * 2, 30.0)
                continue
            log.info("Akış açıldı")
            backoff = 1.0
            while not self._stop_evt.is_set():
                ok, frame = cap.read()
                if not ok:
                    log.warning("Kare okunamadı; yeniden bağlanılıyor")
                    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) Kurallar: çizgi geçişi ve bölge ihlali ----------------------------------
def signed_distance(p, a, b) -> float:
    """p noktasının AB doğrusuna işaretli uzaklığı (piksel). İşaret, hangi tarafta olduğunu verir."""
    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:
    """p'nin AB doğrusuna izdüşümü, doğru parçasının içinde mi (sonsuz doğruda değil)?"""
    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:  # çizginin hemen üstündeyken titreşimi yok say
            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) Olay çıkışı: kuyruk + gönderici iş parçacığı ------------------------------
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)  # bağlantı yokken bellekte bekleyecek en fazla mesaj

    def on_connect(c, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.error("MQTT bağlantısı reddedildi: %s", reason_code)
            return
        log.info("MQTT bağlı")
        c.publish(f"{TOPIC_BASE}/status", "online", qos=1, retain=True)

    def on_disconnect(c, userdata, flags, reason_code, properties):
        log.warning("MQTT bağlantısı koptu: %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()  # ağ döngüsü ve otomatik yeniden bağlanma arka planda
    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 denemesi %d başarısız: %s", i + 1, exc)
        time.sleep(2 ** i)
    log.error("Webhook teslim edilemedi, olay atlandı")


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) Ana döngü --------------------------------------------------------------
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)  # her kamera için ayrı model nesnesi: izleyici durumu buna bağlı
    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("Olay kuyruğu dolu; olay atlandı: %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:  # yeni kare yok
            time.sleep(0.005)
            continue
        last_seq, next_run = seq, now + period  # kare atlama: saniyede en çok PROCESS_FPS kare

        if line is None:  # geometriyi ilk karenin çözünürlüğüne göre piksele çevir
            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:  # iz kimliği yoksa (boş kare) atla
            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]:  # eski izleri temizle
            last_seen.pop(tid)
            line.forget(tid)
            zone.forget(tid)

        processed += 1
        if now - stats_t0 >= 60:
            log.info("işlenen fps=%.1f yeniden_bağlanma=%d kuyruk=%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:  # düzgün kapanışta LWT tetiklenmez; durumu kendimiz yazarız
        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()

Çalıştırmak için:

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

Üretime alırken dikkat edilecekler

KVKK ve görüntü saklamamak. Kamera görüntüsü kişisel veridir. Betik kareyi diske yazmıyor, olaya görüntü eklemiyor, logda yalnızca sayılar var. Geliştirme sırasında kutulu görüntüyü ekranda görmek isteyeceksiniz; bunu yalnızca test ortamında, gerekli bilgilendirme yapılmış alanlarda yapın ve üretim imajına koymayın. Hukuki çerçeveyi KVKK ve görüntü işleme yazısında anlattım.

Saat. Olay zaman damgası UTC ve sunucu saatinden geliyor. Sunucu NTP’ye bağlı değilse olayları başka sistemlerin kayıtlarıyla eşleştiremezsiniz.

Süreç yönetimi. Betik SIGTERM sinyalini yakalayıp düzgün kapanıyor ve durum konusuna offline yazıyor; systemd ya da konteyner altında çalıştırmaya hazır. Beklenmedik bir çökme durumunda aynı bilgiyi broker, son vasiyet mesajıyla yayınlıyor.

Kamera başına süreç. Her kamera için ayrı süreç hatayı yalıtır ve izleyici durumlarını karıştırmaz; bedeli, her sürecin modeli GPU belleğine ayrıca yüklemesidir. Kamera sayısı arttıkça bellek ölçümü yapıp gerekirse birden fazla kamerayı tek süreçte toplu işlemeye geçin.

Doğruluk. Hazır COCO modeli “person” sınıfında iyi bir başlangıç, ama kendi sahanızın ışığında, açısında ve gece modunda ölçülmeden kabul edilmemeli.

Kontrol listesi

• Lisans kararı verildi (AGPL-3.0 uyumu ya da kurumsal lisans).
• OpenCV derlemesinde FFmpeg desteği hedef makinede doğrulandı.
• RTSP TCP’ye zorlandı, açma/okuma zaman aşımları verildi, yeniden bağlanma test edildi (kamera kablosu çekilerek).
• Okuma ayrı iş parçacığında, yalnızca son kare tutuluyor.
• İzleyici açıkça belirtildi, track_buffer işlenen kare hızına göre ayarlandı.
• Çizgi yönü ve bölge sınırları sahada gerçek geçişlerle doğrulandı.
• Olay şeması sabitlendi; tüketici event_id ile tekrarları ayıklıyor.
• MQTT bağlantısı TLS ve kullanıcı doğrulamalı, konu yetkileri kısıtlı.
• Kare diske yazılmıyor, olaylarda ve loglarda görüntü ya da parola yok.
• İşlenen kare hızı, yeniden bağlanma ve kuyruk uzunluğu izleniyor.