ON DIH CHEESE ts is just so food

pipeline/core.py

@@ -1,9 +1,12 @@
+import concurrent.futures
 import logging
+import time
 from collections import deque
+from concurrent.futures import Future
 from typing import List, Optional, Tuple
 
 from pipeline.protocols import Model, Sink, StreamSource, Verifier
-from pipeline.types import BoundingBox, Detection, Frame, Ticket
+from pipeline.types import BoundingBox, Detection, Frame, Ticket, nms_by_class
 
 logger = logging.getLogger(__name__)
 
@@ -17,73 +20,179 @@ class SmokingDetectionPipeline:
         vlm_model: Optional[Verifier] = None,
         sinks: Optional[list[Sink]] = None,
         inference_interval: float = 0.5,
+        active_interval: float = 0.0,
+        cooldown_frames: int = 60,
     ):
         self.person_model = person_model
         self.cigarette_model = cigarette_model
         self.verifier_model = verifier_model
         self.vlm_model = vlm_model
         self.sinks = sinks or []
-        self.inference_interval = inference_interval
 
+        # Adaptive Sampling Settings
+        self.idle_interval = inference_interval
+        self.active_interval = active_interval
+        self.cooldown_frames = cooldown_frames
+
+        # State
+        self.current_interval = self.idle_interval
+        self.active_cooldown = 0
         self.last_inference_time = 0.0
+
+        # Async Inference State
+        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+        self.current_future: Optional[Future] = None
+        self.latest_detections: List[
+            Detection
+        ] = []  # Cache for display to prevent flickering
+        self.latest_ticket: Optional[Ticket] = None
+
         # Buffer for temporal context (optional, for VLM or sequence logic)
         self.frame_buffer = deque(maxlen=30)
-        logger.info(f"Pipeline initialized with {len(self.sinks)} sinks.")
+        logger.info(
+            f"Pipeline initialized with {len(self.sinks)} sinks. Async Inference enabled."
+        )
 
     def process_stream(self, source: StreamSource):
         """Main loop processing the video stream."""
         logger.info("Starting stream processing...")
-        # Batch processing strategy:
-        # Instead of strict batches, we process frame-by-frame but can look back
         count = 0
+
         for frame in source.stream():
             count += 1
             if count % 100 == 0:
                 logger.info(f"Processed {count} frames...")
             self.frame_buffer.append(frame)
 
-            # Skip inference if too soon
-            if (frame.timestamp - self.last_inference_time) >= self.inference_interval:
-                self._process_frame(frame)
-                self.last_inference_time = frame.timestamp
+            # 1. Check if previous inference finished
+            if self.current_future and self.current_future.done():
+                try:
+                    # Retrieve results from the finished job
+                    has_activity, detections, ticket = self.current_future.result()
+
+                    # Update Cache
+                    self.latest_detections = detections
+                    self.latest_ticket = ticket
+                    self.current_future = None
+
+                    # Adaptive Logic Update
+                    if has_activity:
+                        if self.current_interval != self.active_interval:
+                            logger.info(
+                                "Frame %s: Activity detected! Switching to ACTIVE mode (0s interval).",
+                                frame.frame_id,
+                            )
+                        self.current_interval = self.active_interval
+                        self.active_cooldown = self.cooldown_frames
+                    elif self.active_cooldown > 0:
+                        if self.active_cooldown % 30 == 0:
+                            logger.debug(
+                                "Cooling down... %s frames left.", self.active_cooldown
+                            )
+                        self.active_cooldown -= 1
+                    else:
+                        if self.current_interval != self.idle_interval:
+                            logger.info(
+                                "Cooldown complete. Switching to IDLE mode (0.5s interval)."
+                            )
+                        self.current_interval = self.idle_interval
+
+                except Exception as e:
+                    logger.error(f"Inference error: {e}")
+                    self.current_future = None
+
+            # 2. Schedule new inference if idle and time condition met
+            now = frame.timestamp
+            time_since_last = now - self.last_inference_time
+
+            if self.current_future is None and time_since_last >= self.current_interval:
+                # Submit 'detect' job. We pass a COPY of data or rely on immutable usage.
+                # Frame.data is numpy, which is mutable but we trust models not to write to it.
+                # Since models might run on GPU/Thread, and we continue, it's safer to not rely on 'frame' object mutation.
+                # We will return the results and apply them later.
+                self.last_inference_time = now
+                self.current_future = self.executor.submit(self._run_inference, frame)
+
+            # 3. Apply cached detections to current frame for visualization
+            # This ensures smooth video output even if inference runs at 10fps
+            frame.detections.extend(self.latest_detections)
+
+            # 4. Run NMS to clean up overlapping boxes before display
+            frame.detections = nms_by_class(frame.detections, iou_threshold=0.5)
+
+            # Ticket dispatch: Only attach to ONE frame, then clear
+            # This prevents duplicate alerts in sinks like FileAlertStore
+            if self.latest_ticket:
+                frame.ticket = self.latest_ticket
+                self.latest_ticket = None  # Consume the ticket
 
             # Update sinks (Visualizers, etc)
             for sink in self.sinks:
                 sink.handle(frame)
 
         logger.info("Stream processing finished.")
+        self.executor.shutdown(wait=False)
+
+    def _run_inference(
+        self, frame: Frame
+    ) -> Tuple[bool, List[Detection], Optional[Ticket]]:
+        """
+        Runs the full detection chain.
+        Returns (ActivityDetected, ListOfDetections, TicketOrNone).
+        Does NOT modify frame in place (thread safety).
+        """
+        t0 = time.time()
+        detections = []
 
-    def _process_frame(self, frame: Frame):
         # --- Step 1: Broad Phase (Person Detection) ---
         people = self.person_model.predict(frame.data)
-        frame.detections.extend(people)
+        detections.extend(people)
 
         if not people:
-            return  # FAST EXIT
-
-        logger.debug(f"Frame {frame.frame_id}: Found {len(people)} people.")
+            # logger.debug("No people found.")
+            return False, detections, None
 
         # --- Step 2: Narrow Phase (Cigarette Candidate) ---
         candidates = self._find_candidates(frame, people)
 
         if not candidates:
-            return  # EXIT
-
-        logger.debug(f"Frame {frame.frame_id}: Found {len(candidates)} candidates.")
+            return True, detections, None
 
         # --- Step 3: Verification (Classifier) ---
         best_hit = self._verify_candidates(frame, candidates)
 
+        ticket = None
         # --- Step 4: Action ---
         if best_hit:
             person, cig, score = best_hit
+            logger.info("CONFIRMED SMOKING! Score: %.2f", score)
             ticket = Ticket(
                 frame_id=frame.frame_id,
                 confidence=score,
-                evidence=[frame.data],
-                stage_confirmed="Verifier",  # Simplified for now
+                evidence=[frame.data],  # Note: this keeps ref to image data
+                stage_confirmed="Verifier",
             )
-            frame.ticket = ticket
+
+            # Add the verified box to detections for visualization
+            verified_det = Detection(
+                box=BoundingBox(
+                    x1=person.box.x1
+                    + cig.box.x1,  # Approx logic, need exact coords from find_candidates
+                    y1=person.box.y1 + cig.box.y1,
+                    x2=person.box.x1 + cig.box.x2,
+                    y2=person.box.y1 + cig.box.y2,
+                    conf=score,  # Placeholder
+                ),
+                score=score,
+                label="verified_smoking",
+            )
+            detections.append(verified_det)
+
+        dt = time.time() - t0
+        if dt > 0.1:
+            logger.debug("Inference took %.3fs (Slow)", dt)
+
+        return True, detections, ticket
 
     def _find_candidates(
         self, frame: Frame, people: List[Detection]

pipeline/outputs.py

@@ -1,16 +1,142 @@
+import collections
 import json
 import logging
+import threading
+import time
 from pathlib import Path
-from typing import List, Optional
+from queue import Empty, Queue
+from typing import Deque, List, Optional
 
 import cv2
 import numpy as np
 
+from pipeline.protocols import Sink
 from pipeline.types import Detection, Frame
 
 logger = logging.getLogger(__name__)
 
 
+class SmartEventSink(Sink):
+    """
+    Smart Event Sink:
+    - Maintains a ring buffer of recent frames (e.g., 2 seconds context).
+    - When an alert (Ticket) occurs, it triggers a 'recording' state.
+    - Captures pre-event context + event + post-event context.
+    - Saves these 'events' as separate video clips or sequences.
+    """
+
+    def __init__(
+        self,
+        output_dir: str,
+        fps: float = 30.0,
+        pre_event_seconds=2,
+        post_event_seconds=3,
+    ):
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.fps = fps
+        self.buffer_size = int(fps * pre_event_seconds)
+        self.post_event_frames = int(fps * post_event_seconds)
+
+        self.buffer: Deque[Frame] = collections.deque(maxlen=self.buffer_size)
+
+        # State
+        self.recording = False
+        self.remaining_post_frames = 0
+        self.current_event_writer: Optional[cv2.VideoWriter] = None
+        self.current_event_id = None
+
+        logger.info(f"Initialized SmartEventSink in {output_dir}")
+
+    def handle(self, frame: Frame):
+        # Always add to buffer if not recording to keep context
+        if not self.recording:
+            self.buffer.append(frame)
+
+        # Check for trigger
+        if frame.ticket:
+            self._trigger_event(frame)
+
+        # If recording, write frame
+        if self.recording:
+            self._write_to_event(frame)
+
+            # Decrement counter
+            # If we keep getting tickets, we keep updating the counter (extending recording)
+            if frame.ticket:
+                if self.remaining_post_frames != self.post_event_frames:
+                    logger.debug(
+                        "Event %s extended due to new detection.", self.current_event_id
+                    )
+                self.remaining_post_frames = self.post_event_frames
+            else:
+                self.remaining_post_frames -= 1
+
+            if self.remaining_post_frames <= 0:
+                self._stop_event()
+
+    def _trigger_event(self, trigger_frame: Frame):
+        if not self.recording:
+            # Start new event
+            self.recording = True
+            self.current_event_id = int(time.time())
+            self.remaining_post_frames = self.post_event_frames
+
+            filename = self.output_dir / f"event_{self.current_event_id}.mp4"
+            h, w = trigger_frame.data.shape[:2]
+
+            self.current_event_writer = cv2.VideoWriter(
+                str(filename), cv2.VideoWriter.fourcc(*"mp4v"), self.fps, (w, h)
+            )
+
+            # Dump buffer (Pre-event context)
+            while self.buffer:
+                f = self.buffer.popleft()
+                self._write_to_event(f)
+
+            logger.info("Started recording event %s", self.current_event_id)
+
+    def _write_to_event(self, frame: Frame):
+        if self.current_event_writer:
+            # Draw simple visualization for the clip
+            vis = frame.data.copy()
+            if frame.ticket:
+                cv2.rectangle(vis, (0, 0), (vis.shape[1], 40), (0, 0, 255), -1)
+                cv2.putText(
+                    vis,
+                    "SMOKING DETECTED",
+                    (10, 30),
+                    cv2.FONT_HERSHEY_SIMPLEX,
+                    1,
+                    (255, 255, 255),
+                    2,
+                )
+
+            for det in frame.detections:
+                cv2.rectangle(
+                    vis,
+                    (det.box.x1, det.box.y1),
+                    (det.box.x2, det.box.y2),
+                    (255, 0, 0),
+                    2,
+                )
+
+            self.current_event_writer.write(vis)
+
+    def _stop_event(self):
+        self.recording = False
+        if self.current_event_writer:
+            self.current_event_writer.release()
+            self.current_event_writer = None
+        logger.info("Finished recording event %s", self.current_event_id)
+
+    def close(self):
+        """Releases resources and stops any active recording."""
+        if self.recording:
+            self._stop_event()
+        logger.info("SmartEventSink closed.")
+
+
 class FileAlertStore:
     """
     Persists actionable alerts (Tickets) to a JSONL file.
@@ -51,9 +177,16 @@ class VideoFileStore:
     """
     Writes processed frames to a video file with visualization overlays.
     Also optionally writes metadata sidecar files.
+    Async implementation to prevent blocking the main pipeline.
     """
 
-    def __init__(self, output_path: str, fps: int = 30, fourcc: str = "mp4v"):
+    def __init__(
+        self,
+        output_path: str,
+        fps: float = 30.0,
+        fourcc: str = "mp4v",
+        queue_size: int = 60,
+    ):
         self.output_path = output_path
         self.metadata_path = str(Path(output_path).with_suffix(".jsonl"))
         self.fps = fps
@@ -61,14 +194,42 @@ class VideoFileStore:
         self.writer: Optional[cv2.VideoWriter] = None
         self.frame_size = None
 
+        # Async setup
+        self.queue = Queue(maxsize=queue_size)
+        self.running = True
+        self.thread = threading.Thread(target=self._writer_loop, daemon=True)
+
         # Open metadata file
         self.meta_file = open(self.metadata_path, "w")
-        logger.info(
-            f"Initialized VideoFileStore writing to {output_path} and {self.metadata_path}"
-        )
+        logger.info(f"Initialized Async VideoFileStore writing to {output_path}")
+        self.thread.start()
 
     def handle(self, frame: Frame):
-        """Process a frame: Draw boxes and write to video/log."""
+        """Enqueue frame for processing."""
+        if not self.running:
+            return
+
+        # We BLOCK here if queue is full to guarantee no frames are dropped.
+        # This may cause the main loop to slow down, but output integrity > speed.
+        # The queue size (30 frames ~1 second) provides a buffer for burst writes.
+        self.queue.put(frame, block=True)
+
+    def _writer_loop(self):
+        while self.running or not self.queue.empty():
+            try:
+                frame = self.queue.get(timeout=1.0)
+            except Empty:
+                continue
+
+            try:
+                self._process_frame(frame)
+            except Exception as e:
+                logger.error(f"Error processing video output: {e}")
+            finally:
+                self.queue.task_done()
+
+    def _process_frame(self, frame: Frame):
+        # Heavy lifting happens here in the thread
         vis_image = frame.data.copy()
 
         # 1. Visualization
@@ -146,6 +307,10 @@ class VideoFileStore:
         self.meta_file.flush()
 
     def close(self):
+        self.running = False
+        if self.thread.is_alive():
+            self.thread.join(timeout=2.0)
+
         if self.writer:
             self.writer.release()
         if self.meta_file:

pipeline/protocols.py

@@ -16,6 +16,11 @@ class Verifier(Protocol):
 
 
 class StreamSource(Protocol):
+    @property
+    def fps(self) -> float:
+        """Return the frames per second of the source."""
+        ...
+
     def stream(self) -> Iterator[Frame]:
         """Yield frames from source."""
         ...
@@ -25,3 +30,7 @@ class Sink(Protocol):
     def handle(self, frame: Frame):
         """Process or persist a frame (and its ticket/detections)."""
         ...
+
+    def close(self):
+        """Clean up resources."""
+        ...

pipeline/sources.py

@@ -15,6 +15,21 @@ logger = logging.getLogger(__name__)
 class FileStreamSource(StreamSource):
     def __init__(self, path: str):
         self.path = path
+        self._fps = 30.0
+
+        # Probe for FPS if it's a single file
+        if "*" not in self.path:
+            try:
+                cap = cv2.VideoCapture(self.path)
+                if cap.isOpened():
+                    self._fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+                cap.release()
+            except Exception as e:
+                logger.warning(f"Could not probe FPS from {path}: {e}")
+
+    @property
+    def fps(self) -> float:
+        return self._fps
 
     def stream(self) -> Iterator[Frame]:
         # Handle glob patterns if path is a directory
@@ -48,11 +63,26 @@ class ThreadedRTSPSource(StreamSource):
     def __init__(self, rtsp_url: str, queue_size=5):
         self.rtsp_url = rtsp_url
         self.queue = Queue(maxsize=queue_size)
+        self._fps = 30.0
+
+        # Probe FPS (with timeout safety ideally, but simple here)
+        try:
+            cap = cv2.VideoCapture(self.rtsp_url)
+            if cap.isOpened():
+                self._fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+            cap.release()
+        except Exception:
+            logger.warning("Could not probe FPS from RTSP source.")
+
         self.stopped = False
         self.thread = threading.Thread(target=self._update, daemon=True)
         self._start()
         logger.info(f"Initialized ThreadedRTSPSource for {rtsp_url}")
 
+    @property
+    def fps(self) -> float:
+        return self._fps
+
     def _start(self):
         self.thread.start()
 

pipeline/types.py

@@ -14,12 +14,31 @@ class BoundingBox:
 
     @property
     def area(self):
-        return (self.x2 - self.x1) * (self.y2 - self.y1)
+        return max(0, self.x2 - self.x1) * max(0, self.y2 - self.y1)
 
     @property
     def xyxy(self):
         return [self.x1, self.y1, self.x2, self.y2]
 
+    def iou(self, other: "BoundingBox") -> float:
+        """Calculate Intersection over Union with another box."""
+        # Intersection
+        ix1 = max(self.x1, other.x1)
+        iy1 = max(self.y1, other.y1)
+        ix2 = min(self.x2, other.x2)
+        iy2 = min(self.y2, other.y2)
+
+        inter_w = max(0, ix2 - ix1)
+        inter_h = max(0, iy2 - iy1)
+        inter_area = inter_w * inter_h
+
+        # Union
+        union_area = self.area + other.area - inter_area
+
+        if union_area == 0:
+            return 0.0
+        return inter_area / union_area
+
     def crop(self, image: Any) -> Any:
         # Assuming image is a numpy array (H, W, C)
         # Handle cases where box is partially out of bounds
@@ -39,6 +58,62 @@ class Detection:
     metadata: dict = field(default_factory=dict)
 
 
+def nms(detections: List[Detection], iou_threshold: float = 0.5) -> List[Detection]:
+    """
+    Non-Maximum Suppression: Remove overlapping boxes, keeping the highest confidence ones.
+
+    Args:
+        detections: List of Detection objects
+        iou_threshold: Boxes with IoU > threshold are suppressed
+
+    Returns:
+        Filtered list of detections
+    """
+    if not detections:
+        return []
+
+    # Sort by confidence (highest first)
+    sorted_dets = sorted(detections, key=lambda d: d.score, reverse=True)
+
+    keep = []
+    while sorted_dets:
+        # Take the best one
+        best = sorted_dets.pop(0)
+        keep.append(best)
+
+        # Filter out boxes that overlap too much with 'best'
+        remaining = []
+        for det in sorted_dets:
+            if best.box.iou(det.box) < iou_threshold:
+                remaining.append(det)
+        sorted_dets = remaining
+
+    return keep
+
+
+def nms_by_class(
+    detections: List[Detection], iou_threshold: float = 0.5
+) -> List[Detection]:
+    """
+    Run NMS separately for each class label.
+    This prevents boxes of different classes from suppressing each other.
+    """
+    if not detections:
+        return []
+
+    # Group by label
+    by_label: dict[str, List[Detection]] = {}
+    for det in detections:
+        by_label.setdefault(det.label, []).append(det)
+
+    # Run NMS per class
+    result = []
+    for label, dets in by_label.items():
+        result.extend(nms(dets, iou_threshold))
+
+    return result
+
+
 @dataclass
 class Ticket:
     """Represents a potential alert event."""