Surveillance system with computer

Intro

Have you ever thought about turning your computer and webcam into a surveillance system? All you really need is the desire to bring your ideas to life.

Idea

The need for surveillance around my house and yard brought up a question: 'How can I keep an eye on my place when no one’s home?' I saw it as a fun challenge—I love these kinds of puzzles, especially when they involve programming and engineering.

Once I organized my thoughts on how the project should look, function, and work overall, I came up with a clear set of requirements for my program:

• Works with cheap cameras: USB cams, built-in webcams, IP cameras, etc.
• Cross-platform: Portable and works on Windows, Linux, and Single Board Computers (SBCs).
• Lightweight: Uses as few system resources as possible.
• Object detection: Specifically detects cars and people.
• Instant alerts: Sends me a message with a photo whenever an object is detected.
• Remote access: Accessible over the internet so I can easily check the live stream or saved photos from work.

Since I already had experience with Python and OpenCV for object detection, I knew which direction to go in.

The most challenging part was figuring out how to build a mobile app where a user could view camera feeds, check photos, trigger things remotely, or tweak settings.

So, the setup needed to look like this:

At first, I considered diving into the Qt framework, but that promised to be a huge challenge. Since I prefer finding simpler ways to bring my ideas to life, I thought: what if, instead of building a custom app for every mobile platform, we just use the built-in web browser as the user interface? We could put buttons, video streams, and controls right on a webpage, host it on the home computer as a web server, and access it remotely.

When I asked LLMs for advice on how to build this, their top recommendations were Python frameworks like Django and Flask. The choice was obvious: Flask. It’s simple, easy to grasp, and has straightforward documentation

As usual, a lot of trial and error eventually led to my first success: a Python script that reads the video stream, processes it with OpenCV, and recognizes objects using YOLO models from Ultralytics.

It's really cool stuff. With minimal machine learning knowledge, you can either use their base models or train your own to get the exact results you need.

Here’s a quick video demonstration of my custom-trained model for detecting drones in the sky:

The overall project architecture was designed as shown in the diagram below:

Under the hood

• A camera class continually captures frames from the video input and feeds them directly into two places: the recognition pipeline and the web server for live viewing

def run(self):
    if not self._open_capture():
        self._is_running = False
        return

    while self._is_running:
        self.start_stream.wait()
        if not self._is_running:
            break
        success, image = self.cap.read()

        if not success:
            self.read_failures += 1
            print(
                f"Camera {self.cameraID}: frame read failed from source {self.url}. Retrying...",
                flush=True,
            )
            if self.read_failures >= 5:
                print(
                    f"Camera {self.cameraID}: reopening source after {self.read_failures} consecutive read failures.",
                    flush=True,
                )
                self._reopen_capture()
            time.sleep(1)
            continue

        self.read_failures = 0
        h, w = image.shape[:2]

        if self.counter % fps_avg_frame_count == 0:
            end_time = time.time()
            self.fps = fps_avg_frame_count / (end_time - self.start_time)
            self.start_time = time.time()

        fps_text = "FPS = {:.1f}".format(self.fps)
        text_location = (left_margin, row_size)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
        (text_w, text_h), _ = cv2.getTextSize(timestamp, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)

        cvz.putTextRect(image, fps_text, text_location, 1, 1, (255, 255, 255), (0, 0, 0))
        cvz.putTextRect(image, timestamp, (w - text_w, text_h), 1, 1, (255, 255, 255), (0, 0, 0))

        with self.frame_lock:
            self.last_frame = image.copy()

        payload = (self.cameraID, time.time(), image)

        for q in (self.queue_to_detection, self.queue_to_stream):
            if q.full():
                try:
                    q.get_nowait()
                except queue.Empty:
                    pass
            try:
                q.put_nowait(payload)
            except queue.Full:
                pass

        self.counter += 1

• The Recognition class handles the heavy lifting: it receives the frames, runs YOLO object detection, and outputs the bounding box coordinates

def run(self):
    while self._isRunning:
        if not self.start_event.is_set():
            time.sleep(0.1)
            continue
        try:
            camera_id, time_id, image = self.queue_to_detection.get(timeout=1)
            if self.queue_to_detection.full():
                self.queue_to_detection.get_nowait()
        except queue.Empty:
            continue

        self.frame_counters[camera_id] = self.frame_counters.get(camera_id, 0) + 1
        if self.frame_counters[camera_id] % self.frame_skip != 0:
            continue

        detections = []
        predict_kwargs = {
            "conf": 0.5,
            "verbose": False,
            "imgsz": 320,
            "stream": True,
        }
        if not self.is_tensorrt_engine:
            predict_kwargs["device"] = self.device
            predict_kwargs["half"] = self.use_half

        results = self.model.predict(image, **predict_kwargs)

        for result in results:
            bbox = result.boxes
            for box in bbox:
                x1, y1, x2, y2 = box.xyxy[0]
                x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
                cls = int(box.cls[0])
                current_cls = self.model.names[cls]
                conf = round(box.conf[0].item(), 2)
                detections.append({"class": current_cls, "conf": conf, "bbox": (x1, y1, x2, y2)})
            with self.result_lock:
                self.output_results[camera_id] = {
                    "time": time.time(),
                    "timeID": time_id,
                    "detections": detections,
                }

• start_recognition is the launch point of the application—this is where the processing pipeline gets initialized

def main():
    while True:
        # Queues with size 1 intentionally favor live frames over backlog processing.
        _queue = Queue(maxsize=1)
        pic_queue = Queue(maxsize=1)
        tg_queue = Queue(maxsize=1)
        stream_queue = Queue(maxsize=1)
        inference_queue = Queue(maxsize=1)
        load_data_signal = threading.Event()
        start_event = threading.Event()
        start_stream = threading.Event()
        start_recognition = threading.Event()
        photo_detection_event = threading.Event()
        restart_requested = threading.Event()
        shutdown_requested = threading.Event()

        cp = CameraPipeline(
            start_stream,
            start_recognition,
            inference_queue,
            photo_detection_event,
            pic_queue,
            tg_queue,
            start_event,
            stream_queue,
            load_data_signal,
            _queue,
            restart_requested,
            shutdown_requested,
        )

        try:
            cp.start()
            cp.run()
        finally:
            cp.stop()

        if shutdown_requested.is_set():
            break

        if restart_requested.is_set():
            print("Restarting surveillance application...", flush=True)
            continue

        break

• The Pipeline class acts as the central orchestrator. It initializes all threads, events, and routes while building the Camera, Server, Messenger and Recorder objects. It also monitors conditions to trigger photo saves and send out alert notifications.

def start(self):
    self.server = Server(
        start_event=self.start_event,
        stream_flow=self.stream_queue,
        start_stream=self.start_stream,
        start_recognition=self.start_recognition,
        photo_detection_event=self.photo_detection_event,
        load_data_signal=self.load_data_signal,
        _queue=self._queue,
        route=self.init_route(),
        restart_requested=self.restart_requested,
        shutdown_requested=self.shutdown_requested,
    )

    self.detection_worker = ProcessFrame(
        queue_to_detection=self.inference_queue,
        start_event=self.start_event,
        settings_path=self.route / "settings.json",
    )

    self.photo_record = Photo_Record(
        get_frames=self.pic_queue,
        video_event=self.video_event,
        done_video_event=self.done_video_event,
        set_dir_path=str(DEFAULT_PHOTO_DIR),
        _queue=self._queue,
    )

    self.tg = TG(
        detection_event=self.start_event,
        photo_detection_event=self.photo_detection_event,
        frame_queue=self.tg_queue,
        bot_data=self.route / "bot.json",
    )

    self.threads = [
        threading.Thread(target=self.server.run, daemon=True),
        threading.Thread(target=self.photo_record.photo, daemon=True),
        threading.Thread(target=self.tg.message_menager, daemon=True),
    ]

    for t in self.threads:
        t.start()

    self.detection_worker.start()
    self.build_cameras()

    for cam_id, cam in self.cameras.items():
        print(f"Starting camera {cam_id}")
        cam.start()

def stop(self):
    if self.server:
        self.server.stop()
    if self.detection_worker:
        self.detection_worker.stop()
    for cam in self.cameras.values():
        cam.stop()

def run(self):
    while True:
        try:
            if self.restart_requested.is_set() or self.shutdown_requested.is_set():
                return

            self.start_recognition.wait(timeout=1)

            if not self.start_recognition.is_set():
                continue

            camera_id, time_id, frame = self.inference_queue.get(timeout=1)

            if frame is None:
                raise RuntimeError(f"Empty frame from camera {camera_id}")

            detections = self.detection_worker.get_detections(camera_id)

            if detections is None:
                continue

            gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            gray = cv2.GaussianBlur(gray_frame, (21, 21), 0)

            if camera_id not in self.previous_frame:
                self.previous_frame[camera_id] = gray.copy()
                continue

            diff = cv2.absdiff(self.previous_frame[camera_id], gray)
            _, thresh = cv2.threshold(diff, 50, 255, cv2.THRESH_BINARY)
            kernel = np.ones((3, 3), dtype=np.uint8)
            thresh = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)
            thresh = cv2.dilate(thresh, kernel, iterations=2)
            changed_pixels = cv2.countNonZero(thresh)
            #total_pixels = thresh.shape[0] * thresh.shape[1]
            change_ratio = changed_pixels / thresh.size
            alert_needed = False

            for detec in detections:

                x1, y1, x2, y2 = detec["bbox"]
                label = f'{detec["class"]} {detec["conf"]:.2f}'

                if detec["class"] in ("person", "car"):
                    cvz.cornerRect(
                        frame,
                        (x1, y1, x2 - x1, y2 - y1),
                        30,
                        1,
                        1,
                        (0, 255, 0),
                        (0, 0, 255),
                    )

                    cvz.putTextRect(
                        frame,
                        label,
                        (max(0, x1), max(40, y1)),
                        scale=0.6,
                        offset=3,
                        thickness=1,
                    )

                    if change_ratio > self.pixel_ratio_treshhold:
                        alert_needed = True

            if alert_needed:
                self.last_alert_time = time.time()

                for target_queue in (self.pic_queue, self.tg_queue):
                    if target_queue.full():
                        try:
                            target_queue.get_nowait()
                        except queue.Empty:
                            pass

                try:
                    self.pic_queue.put_nowait(frame)
                except queue.Full:
                    pass

                try:
                    self.tg_queue.put_nowait(frame)
                    self.photo_detection_event.set()
                except queue.Full:
                    pass

            self.previous_frame[camera_id] = gray.copy()

        except queue.Empty:
            raise RuntimeError("No frames received from camera queue")
        except Exception as e:
            print(f"Runtime error inside app_run: {e}", flush=True)
            break

• The Messenger class takes the snapshots containing detected objects and forwards them to Telegram alerts

def message_menager(self):
    while True:
        try:
            self.photo_detection_event.wait()
            while True:
                try:
                    frame = self.frame_queue.get(timeout=1)
                except Empty:
                    self.photo_detection_event.clear()
                    break
                time_point = datetime.datetime.now().strftime("%d_%m_%Y_%H-%M-%S")
                caption = f"_{time_point}"
                self.send_photo(frame, caption)
        except Exception as e:
            print(f"error with tg server responce {e}")

def send_photo(self, frame, caption):
    print("send photo initialised")
    self.data = self._load_bot_data()
    bot_api = self.data.get("token")
    chatId = self.data.get("chat_id")
    if not bot_api or not chatId:
        print("Telegram bot token or chat_id is missing.", flush=True)
        return None

    url = f"https://api.telegram.org/bot{bot_api}/sendPhoto"
    flag, buff = cv2.imencode(".jpg", frame)
    if not flag:
        print("Failed to encode frame for Telegram.", flush=True)
        return None

    files = {"photo": (f"{caption or 'snapshot'}.jpg", buff.tobytes(), "image/jpeg")}
    data = {"chat_id": chatId}
    if caption:
        data["caption"] = caption

    response = requests.post(url, data=data, files=files, timeout=15)
    if response.status_code == 200:
        print("✅ Photo sent successfully!")
        return response.json()

    print(f"❌ Error: {response.status_code}")
    print(response.text)
    return None

• The Server class acts as the bridge between the frontend web UI and the backend processing.

class Server:
    """Flask dashboard, configuration API, and MJPEG stream server."""

    def __init__(
        self,
        start_event,
        start_stream,
        start_recognition,
        stream_flow,
        photo_detection_event,
        load_data_signal,
        _queue,
        route,
        restart_requested,
        shutdown_requested,
    ):
        self.app = Flask(__name__)
        self.app.secret_key = os.environ.get("SURVEILLANCE_SECRET_KEY", "super-secret")
        self.start_event = start_event
        self.start_stream = start_stream
        self.start_recognition = start_recognition
        self.stream_flow = stream_flow
        self.photo_detection_event = photo_detection_event
        self.load_data_signal = load_data_signal
        self.restart_requested = restart_requested
        self.shutdown_requested = shutdown_requested
        self.config = ConfigStore(route)
        self.auth = AuthStore()
        self.scheduler = SurveillanceScheduler(
            start_stream=self.start_stream,
            start_recognition=self.start_recognition,
            photo_dir_queue=_queue,
        )
        self.scheduler.apply_settings(self.config.read_settings())
        self.stream_max_fps = self._read_stream_max_fps()
        self.cam_stream = {}
        self.camera_count = len(self.config.read_cameras().get("cameras", []))
        self.bot_configured = self.config.read_bot().get("configured", False)
        self.detection_started = False
        self.connection_started = False
        self.restart_in_progress = False
        self.shutdown_in_progress = False
        self.http_server = None
        self.register_routes()
        self.stream_worker = threading.Thread(target=self.frame_reciever, daemon=True)
        self.stream_worker.start()

To fix some performance lag, I reorganized the project. I moved things like system initialization, routing, and task scheduling out of the main script and into separate modules.

Here’s what the final architecture looks like:

On startup, the app sets up its routes and auto-creates a system_settings directory if missing. This holds configuration files for your cameras, bot, and app settings so everything persists between runs.

Jetson orin Nano

if you do not know what jetson orin nano is, here is a brief answer from google ai search assistant:

I was really curious to see how the SBC would perform. Surprisingly, it held its own against an RTX 3050 laptop, easily handling a single 30 FPS stream. However, performance takes a hit as you stack on more cameras—more camera objects mean way higher GPU load. So solution was to reduce FPS reading for sake to cut down cpu and gpu load without visible difference for the user.

Since NVIDIA JetPack supports TensorRT (which is tailored for blazing-fast edge computing), I decided to convert the standard YOLO .pt model into a TensorRT .engine file.

from ultralytics import YOLO

model = YOLO("/<path to dir>/yolov8n.pt")
model.export(format="engine", half=True, device = 0, dynamic=True)

I designed the app to be fully cross-platform out of the box—no OS-specific tweaks required for Windows or Linux.

Check out some of the detected snapshot examples below:

Full overview is here on video bellow:

For a project advisor, I used OpenAI's Codex. It’s an excellent tool for correcting architecture, adding annotations, and things like that. On the flip side, you have to be careful when having it modify your code—it can sometimes make changes you didn't even ask for!

Here are my next steps for expanding the project:

• Add video recording.
• Add individual camera settings so I can tweak them based on their specific hardware features.
• Improve the recognition logic to better handle false detections (false positives).