Smart CCTV System

Intrusion detection + alert।

🎬 গল্প দিয়ে শুরু

সাধারণ CCTV শুধু record করে — অপরাধ ঘটার পর video খুঁজে বের করতে হয়। Smart CCTV ঘটনার সাথে সাথে detect, alert, এবং clip save করে। এই প্রজেক্টে একটি intrusion + suspicious activity system বানাবো।

Features

Virtual fence (polygon ROI) — define no-go zone।
Person/vehicle detection (YOLOv8) only — false alarm কমে।
Time-based rule — রাত ১০টা–ভোর ৬টা strict mode।
Telegram/Email/SMS alert with snapshot।
Event clip save (10s before + 10s after)।
Web dashboard — live tiles + event timeline।

Architecture

text

RTSP cam(s) ─► Frame Queue (drop-oldest)
            ─► YOLOv8 (TensorRT, batched)
            ─► Tracker (ByteTrack) — unique IDs
            ─► Rule Engine (ROI ∩ class ∩ time)
                 ├─ Alert  → Telegram / Webhook
                 ├─ Clip   → S3 / MinIO (ring buffer)
                 └─ Event  → Postgres + WebSocket UI

ROI polygon dramatic check

python

import cv2, numpy as np
ROI = np.array([[100,400],[1180,400],[1180,700],[100,700]])

def inside(x, y):
    return cv2.pointPolygonTest(ROI, (int(x), int(y)), False) >= 0

# detection result থেকে foot point নিন (bbox bottom-center)
fx, fy = (x1+x2)/2, y2
if inside(fx, fy) and cls == "person":
    trigger("intrusion", track_id, frame)

Detection + Tracking (YOLOv8 + ByteTrack)

python

from ultralytics import YOLO
model = YOLO("yolov8n.engine")

for r in model.track(source=rtsp_url, persist=True,
                     tracker="bytetrack.yaml", stream=True,
                     classes=[0, 2, 3, 5, 7]):   # person, car, motorbike, bus, truck
    for box, tid, cls in zip(r.boxes.xyxy, r.boxes.id, r.boxes.cls):
        x1,y1,x2,y2 = box.cpu().int().tolist()
        if inside((x1+x2)/2, y2) and int(cls) == 0:
            rule.evaluate(int(tid), r.orig_img)

Rule engine — debounce ও cooldown

python

import time
class Rule:
    def __init__(self, cooldown=30):
        self.last = {}
        self.cd   = cooldown
    def evaluate(self, tid, frame):
        now = time.time()
        if now - self.last.get(tid, 0) < self.cd:
            return                  # same person spamming
        self.last[tid] = now
        save_clip(frame_buffer)     # 10s before/after
        send_alert(frame, tid)

Cooldown

একই ব্যক্তি ROI-তে দাঁড়িয়ে থাকলেও প্রতি ৩০s-এ একবার alert — operator spam হবে না।

Telegram alert (১০ লাইন)

python

import requests, cv2
TOKEN, CHAT = os.environ["TG_TOKEN"], os.environ["TG_CHAT"]

def send_alert(frame, tid):
    ok, buf = cv2.imencode(".jpg", frame)
    requests.post(
        f"https://api.telegram.org/bot{TOKEN}/sendPhoto",
        data={"chat_id": CHAT,
              "caption": f"⚠️ Intrusion — track #{tid}"},
        files={"photo": ("a.jpg", buf.tobytes(), "image/jpeg")},
        timeout=5)

Ring buffer (pre-event clip)

python

from collections import deque
PRE = 25 * 10                          # 10s @ 25fps
buf = deque(maxlen=PRE)

# প্রতি frame-এ
buf.append(frame.copy())

def save_clip(post_frames):
    frames = list(buf) + post_frames
    w = cv2.VideoWriter(f"clip_{int(time.time())}.mp4",
        cv2.VideoWriter_fourcc(*"mp4v"), 25, frames[0].shape[1::-1])
    for f in frames: w.write(f)
    w.release()

Suspicious activity (loitering)

Track ID-এর position 30s-এর জন্য store।
Bounding box-এর variance খুব কম + same ROI → loitering।
Running detection — সময়ের সাথে position change rate threshold।
Crowd density — frame-এ person count > N।

Deployment topology

text

[Jetson Orin]  per 4 cameras  ─► local detection
        └── MQTT/HTTP events ──► Central Server
                                  ├─ Postgres (events)
                                  ├─ MinIO    (clips)
                                  └─ React dashboard (websocket)

প্র্যাকটিস টাস্ক

নিজের ঘরের webcam-এ virtual fence draw করে person intrusion alert বানান।
ByteTrack track ID-এর সাথে clip save logic যোগ করুন।
Loitering rule (10s ROI-তে static) implement করুন।