""" Edge production live counter — RTSP + YOLO RKNN + ByteTrack + line crossing. Runs on RK3588 hardware with RKNN model (320×320 input). Uses ByteTrack (Kalman filter + two-stage IoU association) for tracking. """ import contextlib import numpy as np import cv2 import csv import json import os import shutil import signal import socket import sys import threading import time from collections import deque from datetime import datetime from pathlib import Path from dotenv import load_dotenv load_dotenv() from rknnlite.api import RKNNLite from counter_store import CounterStore from video_session_writer import VideoSessionWriter # --- config (override via env / .env) --- OUTPUT_DIR = os.getenv("OUTPUT_DIR", "/opt/jetson-counter") DB_PATH = os.getenv("DB_PATH", f"{OUTPUT_DIR}/jetson_counter.db") STATE_FILE = os.getenv("STATE_FILE", f"{OUTPUT_DIR}/current_counter.json") SOURCE = os.getenv("SOURCE", "rtsp://user:pass@192.168.0.100:554/stream1") MODEL_PATH = os.getenv("MODEL_PATH", "/opt/jetson-counter/yolo11n.rknn") CAMERA_NAME = os.getenv("CAMERA_NAME", "CC1") OBJECT_LABEL = os.getenv("OBJECT_LABEL", "object") CLASS_OBJECT = os.getenv("CLASS_OBJECT", "object") LINE_Y1 = int(os.getenv("LINE_Y1")) if os.getenv("LINE_Y1") else None LINE_Y1_FRAC = float(os.getenv("LINE_Y1_FRAC", "0.33")) LINE_Y2 = int(os.getenv("LINE_Y2")) if os.getenv("LINE_Y2") else None LINE_Y2_FRAC = float(os.getenv("LINE_Y2_FRAC", "0.66")) IMGSZ = int(os.getenv("IMGSZ", "320")) HALF = os.getenv("HALF", "false").lower() == "true" CONF = float(os.getenv("CONF", "0.3")) NMS_IOU = float(os.getenv("NMS_IOU", "0.45")) DEVICE = int(os.getenv("DEVICE", "0")) # RKNN NPU core mask CORE_MASK = int(os.getenv("CORE_MASK", "1")) # YOLO decoder config NUM_CLASSES = int(os.getenv("NUM_CLASSES", "2")) SCORE_SIGMOID = os.getenv("SCORE_SIGMOID", "false").lower() == "true" # ByteTrack settings TRACK_HIGH_THRESH = float(os.getenv("TRACK_HIGH_THRESH", "0.5")) TRACK_LOW_THRESH = float(os.getenv("TRACK_LOW_THRESH", "0.1")) TRACK_MATCH_THRESH = float(os.getenv("TRACK_MATCH_THRESH", "0.8")) TRACK_BUFFER = int(os.getenv("TRACK_BUFFER", "30")) TRACK_MIN_HITS = int(os.getenv("TRACK_MIN_HITS", "3")) # Dedup guard against ID-switch double counts: ignore a second crossing in the # same direction that happens within DEDUP_FRAMES and DEDUP_PX (x-distance) of a # recent crossing. DEDUP_FRAMES = int(os.getenv("DEDUP_FRAMES", "15")) DEDUP_PX = float(os.getenv("DEDUP_PX", "60")) # Trajectory inheritance across ID switches: when a new track appears, inherit the # last position of a recently-seen nearby track so crossings are not missed when # the ID changes right at the line. INHERIT_SEC = float(os.getenv("INHERIT_SEC", "1.0")) INHERIT_PX = float(os.getenv("INHERIT_PX", "60")) DAILY_CUTOFF_TIME = os.getenv("DAILY_CUTOFF_TIME", "20:00") EXPORT_CSV = os.getenv("EXPORT_CSV", "true").lower() == "true" CROSS_CSV = os.getenv("CROSS_CSV", f"{OUTPUT_DIR}/crossings.csv") # Save an annotated frame snapshot each time an object crosses a line and the # counter increases. SAVE_CROSS_SNAPSHOT = os.getenv("SAVE_CROSS_SNAPSHOT", "false").lower() == "true" # Also save one snapshot the first time each object is detected (before it crosses), # named with the same track id so it can be correlated with the crossing snapshot. SAVE_DETECT_SNAPSHOT = os.getenv("SAVE_DETECT_SNAPSHOT", "false").lower() == "true" CROSS_SNAPSHOT_DIR = os.getenv("CROSS_SNAPSHOT_DIR", f"{OUTPUT_DIR}/snapshots") CROSS_SNAPSHOT_QUALITY = int(os.getenv("CROSS_SNAPSHOT_QUALITY", "85")) # Retention: delete oldest snapshots when either limit is exceeded (0 = disabled). CROSS_SNAPSHOT_MAX_FILES = int(os.getenv("CROSS_SNAPSHOT_MAX_FILES", "1000")) CROSS_SNAPSHOT_MAX_AGE_DAYS = float(os.getenv("CROSS_SNAPSHOT_MAX_AGE_DAYS", "7")) # Run the cleanup sweep at most every N seconds to limit filesystem scans. CROSS_SNAPSHOT_CLEANUP_SEC = int(os.getenv("CROSS_SNAPSHOT_CLEANUP_SEC", "3600")) RATE_WINDOW_SEC = int(os.getenv("RATE_WINDOW_SEC", "60")) WARMUP_FRAMES = int(os.getenv("WARMUP_FRAMES", "30")) RECONNECT_DELAY_SEC = int(os.getenv("RECONNECT_DELAY_SEC", "3")) MAX_RECONNECT_ATTEMPTS = int(os.getenv("MAX_RECONNECT_ATTEMPTS", "0")) # Max stream-error prints per outage (Cannot open / Stream dropped). After this, # further errors are silent until the stream recovers, then the budget resets. # 0 = unlimited (old behavior). STREAM_ERROR_LOG_MAX = int(os.getenv("STREAM_ERROR_LOG_MAX", "3")) TRACKED_PRUNE_SEC = int(os.getenv("TRACKED_PRUNE_SEC", "300")) RECORD_VIDEO = os.getenv("RECORD_VIDEO", "false").lower() == "true" VIDEO_SEGMENT_SEC = int(os.getenv("VIDEO_SEGMENT_SEC", "3600")) # legacy segment mode only OUTPUT_FPS = int(os.getenv("OUTPUT_FPS", "15")) # ByteTrack-style session recording: encode in SHM, move to VIDEO_OUTPUT_DIR on session end. SHM_DIR = os.getenv("SHM_DIR", "/dev/shm/zenai-kpc-counter") VIDEO_OUTPUT_DIR = os.getenv("VIDEO_OUTPUT_DIR", OUTPUT_DIR) RECORD_END_DELAY = float(os.getenv("RECORD_END_DELAY", "1")) RECORD_DISCARD_EMPTY = os.getenv("RECORD_DISCARD_EMPTY", "true").lower() == "true" RECORD_RETENTION_DAYS = int(os.getenv("RECORD_RETENTION_DAYS", "3")) _crf_raw = os.getenv("RECORD_CRF", "").strip() RECORD_CRF = int(_crf_raw) if _crf_raw else -1 RECORD_PRESET = os.getenv("RECORD_PRESET", "").strip() # segment = old hourly dump to OUTPUT_DIR; session = SHM + webhook/control lifecycle RECORD_MODE = os.getenv("RECORD_MODE", "session").strip().lower() LIVE_STREAM_ENABLED = os.getenv("LIVE_STREAM_ENABLED", "false").lower() == "true" LIVE_STREAM_FRAME_PATH = os.getenv( "LIVE_STREAM_FRAME_PATH", f"{SHM_DIR}/live_frame.jpg" ) LIVE_STREAM_QUALITY = int(os.getenv("LIVE_STREAM_QUALITY", "75")) LIVE_STREAM_EVERY_N = int(os.getenv("LIVE_STREAM_EVERY_N", "2")) RTSP_FFMPEG_OPTIONS = os.getenv( "OPENCV_FFMPEG_CAPTURE_OPTIONS", "rtsp_transport;tcp|fflags;nobuffer|flags;low_delay", ) IS_LIVE = SOURCE.lower().startswith(("rtsp://", "http://")) MOTION_DETECTION_ENABLED = os.getenv("MOTION_DETECTION_ENABLED", "false").lower() == "true" MOTION_THRESHOLD = float(os.getenv("MOTION_THRESHOLD", "5.0")) # Per-pixel intensity change (0-255) for a pixel to count as "moved". MOTION_PIXEL_DELTA = int(os.getenv("MOTION_PIXEL_DELTA", "25")) # Fraction of frame pixels that must change (0-1) to trigger inference. Small, # so an object entering the edge of the frame is detected immediately. MOTION_MIN_AREA_FRAC = float(os.getenv("MOTION_MIN_AREA_FRAC", "0.002")) # Always run inference at least every N frames even if no motion (heartbeat), so a # slow/stationary object is never missed for long. MOTION_HEARTBEAT_FRAMES = int(os.getenv("MOTION_HEARTBEAT_FRAMES", "15")) # --- Runtime control (toggle counting on/off on the fly) --- # When enabled, the process watches a small JSON control file and honors its # "counting" flag. Set false to always count (ignore the control file). CONTROL_ENABLED = os.getenv("CONTROL_ENABLED", "false").lower() == "true" CONTROL_FILE = os.getenv("CONTROL_FILE", f"{OUTPUT_DIR}/control.json") # Whether counting is active on startup when no control file exists yet. CONTROL_DEFAULT_COUNTING = os.getenv("CONTROL_DEFAULT_COUNTING", "true").lower() == "true" # Re-read the control file at most every N seconds. CONTROL_POLL_SEC = float(os.getenv("CONTROL_POLL_SEC", "1.0")) # Optional TCP control socket. When enabled, the counter listens for line-based # commands so counting can be toggled over the network (in addition to the file). CONTROL_SOCKET_ENABLED = os.getenv("CONTROL_SOCKET_ENABLED", "false").lower() == "true" CONTROL_SOCKET_HOST = os.getenv("CONTROL_SOCKET_HOST", "127.0.0.1") CONTROL_SOCKET_PORT = int(os.getenv("CONTROL_SOCKET_PORT", "5090")) # --- Status webhook gate (direction-locked when IN or OUT) --- # When enabled, counting runs only while status is IN or OUT, and only that # direction is counted (IN → in crossings, OUT → out crossings). OFF pauses all. STATUS_WEBHOOK_ENABLED = os.getenv("STATUS_WEBHOOK_ENABLED", "false").lower() == "true" STATUS_WEBHOOK_FILE = os.getenv( "STATUS_WEBHOOK_FILE", "/opt/zenai-kpc-python/status_webhook_state.json", ) _active_raw = os.getenv("STATUS_WEBHOOK_ACTIVE_VALUES", "IN,OUT") STATUS_WEBHOOK_ACTIVE_VALUES = frozenset( part.strip().upper() for part in _active_raw.split(",") if part.strip() ) STATUS_WEBHOOK_INACTIVE_VALUE = os.getenv("STATUS_WEBHOOK_INACTIVE_VALUE", "OFF").strip().upper() CROSS_FLASH_FRAMES = 12 POPUP_LIFETIME = 20 LINE_PULSE_FRAMES = 12 COUNT_PULSE_FRAMES = 15 SKELETON = [(0, 1), (4, 3), (1, 2), (3, 2), (2, 6), (2, 5), (2, 7), (7, 8)] SK_COLORS = [ (0, 255, 255), (0, 255, 255), (255, 0, 255), (255, 0, 255), (0, 255, 0), (255, 255, 0), (0, 0, 255), (200, 200, 0), ] C_PANEL = (28, 24, 18) C_BORDER = (90, 85, 75) C_ACCENT = (255, 200, 60) C_GREEN = (80, 220, 100) C_TEXT = (235, 235, 235) C_MUTED = (150, 150, 150) C_OBJECT_BOX = (0, 165, 255) C_LINE_CORE = (180, 220, 255) C_LINE_GLOW = (100, 160, 220) shutdown_requested = False def request_shutdown(signum, frame): global shutdown_requested shutdown_requested = True print("\nShutdown requested — finishing current frame...") signal.signal(signal.SIGINT, request_shutdown) signal.signal(signal.SIGTERM, request_shutdown) # ============================================================================= # YOLO output decoder (NMS only — boxes are pre-decoded by the model) # ============================================================================= def _nms(boxes, scores, iou_thr=0.45): order = np.argsort(scores)[::-1] keep = [] while len(order) > 0: idx = order[0] keep.append(idx) if len(order) == 1: break xx1 = np.maximum(boxes[idx, 0], boxes[order[1:], 0]) yy1 = np.maximum(boxes[idx, 1], boxes[order[1:], 1]) xx2 = np.minimum(boxes[idx, 2], boxes[order[1:], 2]) yy2 = np.minimum(boxes[idx, 3], boxes[order[1:], 3]) w = np.maximum(0.0, xx2 - xx1) h = np.maximum(0.0, yy2 - yy1) inter = w * h area_i = (boxes[idx, 2] - boxes[idx, 0]) * (boxes[idx, 3] - boxes[idx, 1]) area_o = (boxes[order[1:], 2] - boxes[order[1:], 0]) * ( boxes[order[1:], 3] - boxes[order[1:], 1] ) iou = inter / (area_i + area_o - inter + 1e-16) order = order[1:][iou < iou_thr] return np.array(keep) # ============================================================================= # IoU helpers (xyxy format) # ============================================================================= def _ious_xyxy(boxes_a, boxes_b): """Pairwise IoU: (N,4) vs (M,4) → (N,M) matrix.""" n, m = len(boxes_a), len(boxes_b) if n == 0 or m == 0: return np.zeros((n, m), dtype=np.float32) xx1 = np.maximum(boxes_a[:, None, 0], boxes_b[None, :, 0]) yy1 = np.maximum(boxes_a[:, None, 1], boxes_b[None, :, 1]) xx2 = np.minimum(boxes_a[:, None, 2], boxes_b[None, :, 2]) yy2 = np.minimum(boxes_a[:, None, 3], boxes_b[None, :, 3]) iw = np.maximum(0.0, xx2 - xx1) ih = np.maximum(0.0, yy2 - yy1) inter = iw * ih area_a = (boxes_a[:, 2] - boxes_a[:, 0]) * (boxes_a[:, 3] - boxes_a[:, 1]) area_b = (boxes_b[:, 2] - boxes_b[:, 0]) * (boxes_b[:, 3] - boxes_b[:, 1]) return inter / (area_a[:, None] + area_b[None, :] - inter + 1e-16) def _greedy_match(cost_matrix, threshold=0.3): """Greedy linear assignment. Returns pairs (row_idx, col_idx).""" if cost_matrix.size == 0: return [] n, m = cost_matrix.shape flat = [(cost_matrix[i, j], i, j) for i in range(n) for j in range(m)] flat.sort() row_used = set() col_used = set() pairs = [] for cost, i, j in flat: if cost >= threshold: break if i in row_used or j in col_used: continue row_used.add(i) col_used.add(j) pairs.append((i, j)) return pairs # ============================================================================= # Kalman filter box tracker (state: x, y, w, h, vx, vy, vw, vh) # ============================================================================= class KalmanBoxTracker: count = 0 def __init__(self, bbox_xyxy): KalmanBoxTracker.count += 1 self.track_id = KalmanBoxTracker.count x1, y1, x2, y2 = bbox_xyxy w, h = x2 - x1, y2 - y1 x, y = x1 + w / 2, y1 + h / 2 self.kf = _KalmanFilter() self.kf.x[:4, 0] = np.array([x, y, w, h], dtype=np.float32) self.time_since_update = 0 self.hits = 1 self.hit_streak = 1 self.age = 1 def predict(self): if self.kf.x[6] + self.kf.x[2] <= 0: self.kf.x[6] *= 0.0 self.kf.predict() self.age += 1 self.time_since_update += 1 def update(self, bbox_xyxy): self.time_since_update = 0 self.hits += 1 self.hit_streak += 1 x1, y1, x2, y2 = bbox_xyxy w, h = x2 - x1, y2 - y1 x, y = x1 + w / 2, y1 + h / 2 self.kf.update(np.array([x, y, w, h], dtype=np.float32)) def get_state(self): """Returns xyxy bbox from Kalman state.""" xx = self.kf.x[:4, 0] x, y, w, h = xx[0], xx[1], xx[2], xx[3] x1 = x - w / 2 y1 = y - h / 2 x2 = x + w / 2 y2 = y + h / 2 return np.array([x1, y1, x2, y2], dtype=np.float32) def get_cx(self): return float(self.kf.x[0, 0]) def get_cy(self): return float(self.kf.x[1, 0]) class _KalmanFilter: """8-state constant-velocity Kalman filter for bounding box tracking.""" def __init__(self): ndim, dt = 4, 1.0 self.motion_mat = np.eye(2 * ndim, 2 * ndim, dtype=np.float32) for i in range(ndim): self.motion_mat[i, ndim + i] = dt self.update_mat = np.eye(ndim, 2 * ndim, dtype=np.float32) self._std_weight_position = 1.0 / 20 self._std_weight_velocity = 1.0 / 160 self.x = np.zeros((8, 1), dtype=np.float32) self.P = np.eye(8, dtype=np.float32) * 10.0 def predict(self): std_pos = [ self._std_weight_position * self.x[2], self._std_weight_position * self.x[3], self._std_weight_position * self.x[2], self._std_weight_position * self.x[3], ] std_vel = [ self._std_weight_velocity * self.x[2], self._std_weight_velocity * self.x[3], self._std_weight_velocity * self.x[2], self._std_weight_velocity * self.x[3], ] Q = np.diag(np.square(np.concatenate([std_pos, std_vel]))) self.x = self.motion_mat @ self.x self.P = self.motion_mat @ self.P @ self.motion_mat.T + Q def update(self, z): R = np.diag( np.square( [ self._std_weight_position * z[2], self._std_weight_position * z[3], self._std_weight_position * z[2], self._std_weight_position * z[3], ] ) ) H = self.update_mat S = H @ self.P @ H.T + R K = self.P @ H.T @ np.linalg.inv(S) y = z.reshape(4, 1) - H @ self.x self.x = self.x + K @ y I_KH = np.eye(8) - K @ H self.P = I_KH @ self.P @ I_KH.T + K @ R @ K.T # ============================================================================= # ByteTrack multi-object tracker # ============================================================================= class ByteTracker: """ByteTrack: two-stage association with Kalman filter prediction.""" def __init__( self, track_high_thresh=0.5, track_low_thresh=0.1, match_thresh=0.8, track_buffer=30, min_hits=3, ): self.high_thresh = track_high_thresh self.low_thresh = track_low_thresh self.match_thresh = match_thresh self.track_buffer = track_buffer self.min_hits = min_hits self.tracked_tracks = [] self.lost_tracks = [] self.removed_tracks = [] self.frame_id = 0 def update(self, boxes_xyxy, scores): self.frame_id += 1 # --- separate detections by score --- if len(boxes_xyxy) > 0: remain = scores > self.low_thresh remain_orig_idx = np.where(remain)[0] dets = boxes_xyxy[remain] det_scores = scores[remain] is_high = det_scores > self.high_thresh is_low = ~is_high else: remain_orig_idx = np.zeros(0, dtype=np.int64) dets = np.zeros((0, 4), dtype=np.float32) det_scores = np.zeros(0, dtype=np.float32) is_high = np.zeros(0, dtype=bool) is_low = np.zeros(0, dtype=bool) # --- Kalman predict all existing tracks --- track_pool = self.tracked_tracks + self.lost_tracks num_tracks = len(track_pool) # Per-frame tracking results matched_track_idx = set() det_to_track = {} tracked_map = {} lost_map = {} # Pre-allocate these for scoping high_idx = np.array([], dtype=np.int64) low_idx = np.array([], dtype=np.int64) match_pairs_high = [] if num_tracks > 0: track_boxes = np.zeros((num_tracks, 4), dtype=np.float32) for ti, trk in enumerate(track_pool): trk.predict() track_boxes[ti] = trk.get_state() # --- first association: high-score ↔ all tracks --- high_idx = np.where(is_high)[0] high_dets = dets[is_high] unmatched_tracks = list(range(num_tracks)) if len(high_dets) > 0: iou_mat = _ious_xyxy(high_dets, track_boxes) cost_mat = 1.0 - iou_mat matches = _greedy_match(cost_mat, threshold=1.0 - self.match_thresh) for di, ti in matches: det_global = int(high_idx[di]) orig_idx = int(remain_orig_idx[det_global]) track_pool[ti].update(dets[det_global]) track_pool[ti].hit_streak = max(1, track_pool[ti].hit_streak) matched_track_idx.add(ti) det_to_track[orig_idx] = track_pool[ti].track_id tracked_map[track_pool[ti].track_id] = (track_pool[ti].get_cx(), track_pool[ti].get_cy()) match_pairs_high.append((det_global, ti)) unmatched_tracks = [ t for t in range(num_tracks) if t not in matched_track_idx ] # --- second association: low-score ↔ unmatched tracks --- low_idx = np.where(is_low)[0] low_dets = dets[is_low] if len(low_dets) > 0 and len(unmatched_tracks) > 0: unmatched_boxes = track_boxes[unmatched_tracks] iou_mat = _ious_xyxy(low_dets, unmatched_boxes) cost_mat = 1.0 - iou_mat matches2 = _greedy_match( cost_mat, threshold=1.0 - self.match_thresh ) for di, uti in matches2: det_global = int(low_idx[di]) pool_idx = unmatched_tracks[uti] orig_idx = int(remain_orig_idx[det_global]) track_pool[pool_idx].update(dets[det_global]) track_pool[pool_idx].hit_streak = max( 1, track_pool[pool_idx].hit_streak ) matched_track_idx.add(pool_idx) det_to_track[orig_idx] = track_pool[pool_idx].track_id tracked_map[track_pool[pool_idx].track_id] = ( track_pool[pool_idx].get_cx(), track_pool[pool_idx].get_cy(), ) # --- reset hit_streak for unmatched tracks --- for ti, trk in enumerate(track_pool): if ti not in matched_track_idx: trk.hit_streak = 0 # --- lifecycle management --- new_tracked = [] new_lost = [] for trk in track_pool: if trk.time_since_update > self.track_buffer: self.removed_tracks.append(trk) elif trk.time_since_update > 0: new_lost.append(trk) else: new_tracked.append(trk) self.tracked_tracks = new_tracked self.lost_tracks = new_lost # --- confirmed tracks (both tracked and lost) --- for trk in self.tracked_tracks + self.lost_tracks: if trk.hit_streak >= self.min_hits or trk.hits >= self.min_hits: tracked_map.setdefault(trk.track_id, (trk.get_cx(), trk.get_cy())) for trk in self.lost_tracks: if trk.hit_streak >= self.min_hits or trk.hits >= self.min_hits: lost_map[trk.track_id] = (trk.get_cx(), trk.get_cy()) # --- new tracks from unmatched high-score dets --- high_all = np.where(is_high)[0] matched_det_ids = set(det_to_track.keys()) for dg in high_all: orig_idx = int(remain_orig_idx[int(dg)]) if orig_idx not in matched_det_ids: trk = KalmanBoxTracker(dets[int(dg)]) self.tracked_tracks.append(trk) det_to_track[orig_idx] = trk.track_id tracked_map[trk.track_id] = (trk.get_cx(), trk.get_cy()) return tracked_map, det_to_track, lost_map # ============================================================================= # RKNN YOLO wrapper (detect output format: (1, 4+num_classes, N)) # ============================================================================= class RKNNYOLO: def __init__( self, model_path, core_mask=1, imgsz=320, conf=0.3, iou=0.45, num_classes=2, num_keypoints=0, score_sigmoid=False, ): self.imgsz = imgsz self.conf = conf self.iou = iou self.num_classes = num_classes self.num_keypoints = num_keypoints self.score_sigmoid = score_sigmoid self.rknn = RKNNLite(verbose=False) ret = self.rknn.load_rknn(model_path) if ret != 0: raise RuntimeError(f"Failed to load RKNN model: {model_path}") ret = self.rknn.init_runtime(core_mask=core_mask) if ret != 0: raise RuntimeError(f"Failed to init RKNN runtime (core_mask={core_mask})") try: sdk_ver = self.rknn.get_sdk_version() print(f"RKNN SDK version: {sdk_ver}") except Exception: pass print(f"RKNN model loaded: {model_path} imgsz={imgsz} core_mask={core_mask}") def _preprocess(self, frame): h0, w0 = frame.shape[:2] scale = min(self.imgsz / h0, self.imgsz / w0) nh, nw = int(h0 * scale), int(w0 * scale) resized = cv2.resize(frame, (nw, nh), interpolation=cv2.INTER_LINEAR) letterbox = np.full((self.imgsz, self.imgsz, 3), 114, dtype=np.uint8) dy = (self.imgsz - nh) // 2 dx = (self.imgsz - nw) // 2 letterbox[dy : dy + nh, dx : dx + nw] = resized rgb = cv2.cvtColor(letterbox, cv2.COLOR_BGR2RGB) gains = np.array([scale, scale, dy, dx], dtype=np.float32) return rgb, gains def __call__(self, frame): h0, w0 = frame.shape[:2] rgb, gains = self._preprocess(frame) scale, _, pad_y, pad_x = gains inp = np.expand_dims(rgb, axis=0) inp = np.ascontiguousarray(inp.astype(np.uint8)) outputs = self.rknn.inference(inputs=[inp]) if len(outputs) == 0: return [] out = outputs[0] out = np.squeeze(out, axis=0) if out.shape[0] == self.num_classes + 4: out = out.T boxes_cxcywh = out[:, :4].copy() cls_raw = out[:, 4:].copy() if self.score_sigmoid: cls_scores = 1.0 / (1.0 + np.exp(-np.clip(cls_raw, -10, 10))) else: cls_scores = cls_raw boxes_xyxy = np.stack( [ boxes_cxcywh[:, 0] - boxes_cxcywh[:, 2] / 2, boxes_cxcywh[:, 1] - boxes_cxcywh[:, 3] / 2, boxes_cxcywh[:, 0] + boxes_cxcywh[:, 2] / 2, boxes_cxcywh[:, 1] + boxes_cxcywh[:, 3] / 2, ], axis=1, ) max_scores = cls_scores.max(axis=1) class_ids = cls_scores.argmax(axis=1) mask = max_scores > self.conf if mask.sum() == 0: return [] bboxes = boxes_xyxy[mask].astype(np.float32) scores = max_scores[mask].astype(np.float32) clses = class_ids[mask] bboxes[:, 0] = (bboxes[:, 0] - pad_x) / scale bboxes[:, 1] = (bboxes[:, 1] - pad_y) / scale bboxes[:, 2] = (bboxes[:, 2] - pad_x) / scale bboxes[:, 3] = (bboxes[:, 3] - pad_y) / scale bboxes[:, 0] = np.clip(bboxes[:, 0], 0, w0) bboxes[:, 1] = np.clip(bboxes[:, 1], 0, h0) bboxes[:, 2] = np.clip(bboxes[:, 2], 0, w0) bboxes[:, 3] = np.clip(bboxes[:, 3], 0, h0) detections = [] for cls_id in range(self.num_classes): idx = np.where(clses == cls_id)[0] if len(idx) == 0: continue keep = _nms(bboxes[idx], scores[idx], iou_thr=self.iou) for k in keep: j = idx[k] detections.append( { "bbox": bboxes[j].tolist(), "score": float(scores[j]), "cls": int(clses[j]), "keypoints": None, } ) return detections def release(self): self.rknn.release() # ============================================================================= # Drawing helpers # ============================================================================= def resolve_line_y1(frame_height): if LINE_Y1 is not None: return LINE_Y1 return int(frame_height * LINE_Y1_FRAC) def resolve_line_y2(frame_height): if LINE_Y2 is not None: return LINE_Y2 return int(frame_height * LINE_Y2_FRAC) def crossed_top_down(prev_y, y, line_y): return prev_y < line_y <= y def crossed_bottom_up(prev_y, y, line_y): return prev_y > line_y >= y def is_duplicate_cross(recent, cx, frame_idx): """True if a crossing near cx happened within the dedup window (ID-switch guard).""" while recent and frame_idx - recent[0][0] > DEDUP_FRAMES: recent.popleft() for _, prev_cx in recent: if abs(prev_cx - cx) <= DEDUP_PX: return True return False def _inherit_prev(tracked, new_tid, cx, cy, mono, max_age, max_px): """Find a recently-seen nearby track for ID-switch continuation. Returns (cx, cy, mono, source_tid) or None. The predecessor must be close in BOTH x and y (same physical object at the same spot). Matching on x only would let a new track inherit a far-away y, seeding a huge prev->cur jump that fabricates a line crossing in the wrong direction.""" best = None best_dist = max_px for tid, (tcx, tcy, ts) in tracked.items(): if tid == new_tid: continue if mono - ts > max_age: continue if abs(tcy - cy) > max_px: continue dist = abs(tcx - cx) if dist <= best_dist: best_dist = dist best = (tcx, tcy, ts, tid) return best def _default_cross_state(): """Per-track line-crossing state for sequence counting. seen_out / seen_in: ever registered a pass of that line (arm or count). counted: already contributed one IN or OUT (at most one per physical object). """ return {"seen_out": False, "seen_in": False, "counted": False} def _copy_cross_state(src): return { "seen_out": bool(src.get("seen_out", False)), "seen_in": bool(src.get("seen_in", False)), "counted": bool(src.get("counted", False)), } def now_str(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def read_counting_flag(default=True): """Read the 'counting' flag from the control file. Returns default on any error.""" try: with open(CONTROL_FILE, "r", encoding="utf-8") as f: data = json.load(f) return bool(data.get("counting", default)) except FileNotFoundError: return default except Exception: return default def read_status_webhook(): """Return normalized status from webhook file, or inactive value on error.""" try: with open(STATUS_WEBHOOK_FILE, "r", encoding="utf-8") as f: data = json.load(f) status = str(data.get("status", "")).strip().upper() return status or STATUS_WEBHOOK_INACTIVE_VALUE except Exception: return STATUS_WEBHOOK_INACTIVE_VALUE def status_webhook_allows_counting(status): return status in STATUS_WEBHOOK_ACTIVE_VALUES def write_control_file(counting): """Create/update the control file atomically (used to seed defaults).""" try: Path(CONTROL_FILE).parent.mkdir(parents=True, exist_ok=True) tmp = f"{CONTROL_FILE}.tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump({"counting": bool(counting)}, f) os.replace(tmp, CONTROL_FILE) except Exception as exc: print(f"[{now_str()}] Failed to write control file: {exc}") def start_control_socket(): """Start a TCP server for runtime control. Commands (newline-terminated): START | RESUME | ON -> counting on STOP | PAUSE | OFF -> counting off TOGGLE -> flip STATUS | GET -> report current state It writes the shared control file, so the main loop's file-poll applies it. Returns the server socket (call .close() to stop).""" srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((CONTROL_SOCKET_HOST, CONTROL_SOCKET_PORT)) srv.listen(5) def handle(conn, addr): with conn: conn.settimeout(30) try: buf = b"" while not shutdown_requested: try: chunk = conn.recv(256) except socket.timeout: break if not chunk: break buf += chunk while b"\n" in buf: line, buf = buf.split(b"\n", 1) cmd = line.decode("utf-8", "ignore").strip().upper() if not cmd: continue current = read_counting_flag(CONTROL_DEFAULT_COUNTING) if cmd in ("START", "RESUME", "ON"): write_control_file(True) resp = "OK counting=on" elif cmd in ("STOP", "PAUSE", "OFF"): write_control_file(False) resp = "OK counting=off" elif cmd == "TOGGLE": write_control_file(not current) resp = f"OK counting={'off' if current else 'on'}" elif cmd in ("STATUS", "GET"): resp = f"OK counting={'on' if current else 'off'}" else: resp = "ERR unknown command" conn.sendall((resp + "\n").encode("utf-8")) except Exception: pass def loop(): print(f"Control socket listening on {CONTROL_SOCKET_HOST}:{CONTROL_SOCKET_PORT}") while not shutdown_requested: try: conn, addr = srv.accept() except OSError: break t = threading.Thread(target=handle, args=(conn, addr), daemon=True) t.start() threading.Thread(target=loop, daemon=True).start() return srv @contextlib.contextmanager def _quiet_opencv_stderr(): """Silence OpenCV/FFmpeg stderr (DESCRIBE failed, cap.cpp WARN, etc.).""" saved_fd = None devnull_fd = None try: stderr_fd = sys.stderr.fileno() saved_fd = os.dup(stderr_fd) devnull_fd = os.open(os.devnull, os.O_WRONLY) os.dup2(devnull_fd, stderr_fd) except (AttributeError, OSError, ValueError): yield return try: yield finally: os.dup2(saved_fd, stderr_fd) os.close(saved_fd) os.close(devnull_fd) def open_capture(source): if source.lower().startswith(("rtsp://", "http://")): os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = RTSP_FFMPEG_OPTIONS # When STREAM_ERROR_LOG_MAX > 0, native OpenCV/FFmpeg stderr is suppressed; # stream_error_log emits at most that many user-facing messages per outage. if STREAM_ERROR_LOG_MAX > 0: with _quiet_opencv_stderr(): cap = cv2.VideoCapture(source, cv2.CAP_FFMPEG) else: cap = cv2.VideoCapture(source, cv2.CAP_FFMPEG) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) return cap def warmup_stream(cap, n=WARMUP_FRAMES): print("Warming up stream...") for _ in range(n): cap.read() print("Stream ready!") def open_video_writer(path, w, h, fps): return cv2.VideoWriter(path, cv2.VideoWriter_fourcc(*"avc1"), fps, (w, h)) class CsvLogger: def __init__(self, path, header): Path(path).parent.mkdir(parents=True, exist_ok=True) new_file = not Path(path).exists() or Path(path).stat().st_size == 0 self.file = open(path, "a", newline="", buffering=1) self.writer = csv.writer(self.file) if new_file: self.writer.writerow(header) self.file.flush() def write_row(self, row): self.writer.writerow(row) self.file.flush() def close(self): self.file.close() class VideoSegmentWriter: def __init__(self, output_dir, w, h, fps, segment_sec): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) self.w, self.h, self.fps = w, h, fps self.segment_sec = segment_sec self.segment_start = time.monotonic() self.writer = None self._open_next() def _segment_path(self): ts = datetime.now().strftime("%Y%m%d_%H%M%S") return str(self.output_dir / f"live_{ts}.mp4") def _open_next(self): if self.writer is not None: self.writer.release() path = self._segment_path() self.writer = open_video_writer(path, self.w, self.h, self.fps) self.segment_start = time.monotonic() print(f"Recording segment: {path}") def write(self, frame): if time.monotonic() - self.segment_start >= self.segment_sec: self._open_next() self.writer.write(frame) def release(self): if self.writer is not None: self.writer.release() def prune_stale_tracks(tracked, now_mono): stale = [ tid for tid, (_, _, ts) in tracked.items() if now_mono - ts > TRACKED_PRUNE_SEC ] for tid in stale: del tracked[tid] def cleanup_snapshots(snapshot_dir, max_files, max_age_days): """Delete oldest / expired crossing snapshots to bound disk usage.""" d = Path(snapshot_dir) if not d.is_dir(): return files = sorted(d.rglob("*.jpg"), key=lambda p: p.stat().st_mtime) if max_age_days > 0: cutoff = time.time() - max_age_days * 86400 for p in list(files): if p.stat().st_mtime < cutoff: p.unlink(missing_ok=True) files.remove(p) if max_files > 0 and len(files) > max_files: for p in files[: len(files) - max_files]: p.unlink(missing_ok=True) def overlay_rect(img, x1, y1, x2, y2, color, alpha=0.65): x1, y1 = max(0, x1), max(0, y1) x2, y2 = min(img.shape[1], x2), min(img.shape[0], y2) if x2 <= x1 or y2 <= y1: return roi = img[y1:y2, x1:x2] patch = np.full_like(roi, color, dtype=np.uint8) cv2.addWeighted(patch, alpha, roi, 1 - alpha, 0, roi) def draw_pill(img, text, x, y, bg, fg=C_TEXT, font_scale=0.45, pad_x=6, pad_y=4): font = cv2.FONT_HERSHEY_SIMPLEX (tw, th), baseline = cv2.getTextSize(text, font, font_scale, 1) x1, y1 = x, y - th - pad_y x2, y2 = x + tw + pad_x * 2, y + baseline + pad_y cv2.rectangle(img, (x1, y1), (x2, y2), bg, -1) cv2.rectangle(img, (x1, y1), (x2, y2), C_BORDER, 1) cv2.putText(img, text, (x + pad_x, y), font, font_scale, fg, 1, cv2.LINE_AA) def draw_elegant_counting_line(img, line_y, w, pulse_remaining=0, label="LINE"): strength = pulse_remaining / max(LINE_PULSE_FRAMES, 1) glow_alpha = 0.12 + 0.18 * strength for offset in (14, 9, 5): color = tuple(int(c * glow_alpha) for c in C_LINE_GLOW) cv2.line(img, (0, line_y - offset), (w, line_y - offset), color, 1, cv2.LINE_AA) cv2.line(img, (0, line_y + offset), (w, line_y + offset), color, 1, cv2.LINE_AA) dash_len, gap = 18, 12 x = 0 while x < w: x_end = min(x + dash_len, w) cv2.line(img, (x, line_y), (x_end, line_y), C_LINE_CORE, 2, cv2.LINE_AA) x += dash_len + gap cv2.putText( img, label, (14, line_y - 8), cv2.FONT_HERSHEY_SIMPLEX, 0.42, C_LINE_CORE, 1, cv2.LINE_AA, ) def draw_line_count(img, w, line_y, count, label, color, pulse_remaining=0, above=True): text = str(count) font = cv2.FONT_HERSHEY_SIMPLEX boost = 0.35 * (pulse_remaining / max(COUNT_PULSE_FRAMES, 1)) font_scale, thickness = 1.2 + boost, 3 (tw, th), _ = cv2.getTextSize(text, font, font_scale, thickness) (lw, lh), _ = cv2.getTextSize(label, font, 0.45, 1) pad = 12 box_w = max(tw, lw) + pad * 2 box_h = th + lh + pad * 2 + 6 bx2 = w - 16 bx1 = bx2 - box_w if above: by2 = line_y - 10 by1 = by2 - box_h else: by1 = line_y + 10 by2 = by1 + box_h overlay_rect(img, bx1, by1, bx2, by2, C_PANEL, alpha=0.78) cv2.rectangle(img, (bx1, by1), (bx2, by2), color, 2) tx = bx1 + (box_w - tw) // 2 ty = by1 + pad + th cv2.putText(img, text, (tx, ty), font, font_scale, color, thickness, cv2.LINE_AA) slx = bx1 + (box_w - lw) // 2 sly = ty + lh + 6 cv2.putText(img, label, (slx, sly), font, 0.45, C_MUTED, 1, cv2.LINE_AA) def draw_hud(img, w, total, total_in, total_out, elapsed_sec, rate): bar_h = 40 overlay_rect(img, 0, 0, w, bar_h, C_PANEL, alpha=0.72) cv2.line(img, (0, bar_h), (w, bar_h), C_BORDER, 1) cv2.putText( img, "TOTAL", (14, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.32, C_MUTED, 1, cv2.LINE_AA ) cv2.putText( img, str(total), (14, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.55, C_GREEN, 1, cv2.LINE_AA, ) cv2.putText( img, "IN", (100, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.32, C_MUTED, 1, cv2.LINE_AA ) cv2.putText( img, str(total_in), (100, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.55, C_GREEN, 1, cv2.LINE_AA, ) cv2.putText( img, "OUT", (170, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.32, C_MUTED, 1, cv2.LINE_AA ) cv2.putText( img, str(total_out), (170, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.55, C_TEXT, 1, cv2.LINE_AA, ) cv2.putText( img, "UPTIME", (250, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.32, C_MUTED, 1, cv2.LINE_AA, ) cv2.putText( img, f"{elapsed_sec / 3600:.1f}h", (250, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_TEXT, 1, cv2.LINE_AA, ) cv2.putText( img, "RATE", (340, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.32, C_MUTED, 1, cv2.LINE_AA ) cv2.putText( img, f"{rate:.1f}/min", (340, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_ACCENT, 1, cv2.LINE_AA, ) def draw_footer(img, w, h, frame_idx, live_tag, inf_ms=0.0, model_name=""): bar_h = 28 overlay_rect(img, 0, h - bar_h, w, h, C_PANEL, alpha=0.55) cv2.putText( img, f"{live_tag} | {model_name} | Frame {frame_idx} | Inf {inf_ms:.1f}ms", (12, h - 9), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA, ) def draw_skeleton_bold(img, kpts): for (a, b), color in zip(SKELETON, SK_COLORS): if a < len(kpts) and b < len(kpts): xa, ya = int(kpts[a][0]), int(kpts[a][1]) xb, yb = int(kpts[b][0]), int(kpts[b][1]) if xa > 0 and ya > 0 and xb > 0 and yb > 0: cv2.line(img, (xa, ya), (xb, yb), color, 3, cv2.LINE_AA) for kp in kpts: x, y = int(kp[0]), int(kp[1]) if x > 0 and y > 0: cv2.circle(img, (x, y), 6, (255, 255, 255), -1, cv2.LINE_AA) cv2.circle(img, (x, y), 6, (40, 40, 40), 2, cv2.LINE_AA) def draw_popups(img, popups, frame_idx): alive = [] for pop in popups: age = frame_idx - pop["born"] if age > POPUP_LIFETIME: continue alive.append(pop) fade = 1.0 - age / POPUP_LIFETIME y = pop["y"] - int(age * 1.8) color = (int(C_GREEN[0] * fade), int(C_GREEN[1] * fade), int(C_GREEN[2] * fade)) cv2.putText( img, pop["text"], (pop["x"], y), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2, cv2.LINE_AA, ) return alive class StreamErrorLog: """Rate-limit stream error prints to STREAM_ERROR_LOG_MAX per outage. After the cap, messages are suppressed until note_up() when the stream is healthy again; the next drop starts a fresh budget. """ def __init__(self, max_logs=STREAM_ERROR_LOG_MAX): self.max_logs = int(max_logs) self._emitted = 0 self._suppressed = 0 def emit(self, msg: str) -> None: if self.max_logs <= 0: print(msg) return if self._emitted < self.max_logs: self._emitted += 1 print(msg) if self._emitted >= self.max_logs: print( f"[{now_str()}] Stream errors capped at {self.max_logs} " f"this outage — further messages suppressed until stream recovers" ) else: self._suppressed += 1 def note_up(self) -> None: if self._suppressed > 0: print( f"[{now_str()}] Stream recovered " f"(suppressed {self._suppressed} error log(s) during outage)" ) self._emitted = 0 self._suppressed = 0 stream_error_log = StreamErrorLog() def connect_stream(source, warmup=WARMUP_FRAMES): attempts = 0 while not shutdown_requested: cap = open_capture(source) if not cap.isOpened(): attempts += 1 if MAX_RECONNECT_ATTEMPTS and attempts >= MAX_RECONNECT_ATTEMPTS: raise RuntimeError( f"Cannot open source after {attempts} attempts: {source}" ) stream_error_log.emit( f"Cannot open source, retry in {RECONNECT_DELAY_SEC}s..." ) time.sleep(RECONNECT_DELAY_SEC) continue if warmup > 0 and source.lower().startswith(("rtsp://", "http://")): warmup_stream(cap, warmup) w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) if not fps or fps <= 1: fps = OUTPUT_FPS stream_error_log.note_up() return cap, w, h, fps return None, 0, 0, OUTPUT_FPS # ============================================================================= # Main loop # ============================================================================= def run(): global shutdown_requested store = CounterStore( db_path=DB_PATH, state_file=STATE_FILE, camera_name=CAMERA_NAME, object_label=OBJECT_LABEL, cutoff_time=DAILY_CUTOFF_TIME, logger=lambda msg: print(f"[{now_str()}] {msg}"), ) store.start_cutoff_watcher() cross_logger = None if EXPORT_CSV: cross_logger = CsvLogger( CROSS_CSV, ["counting_date", "frame", "direction", "object_id"] ) model = RKNNYOLO( model_path=MODEL_PATH, core_mask=CORE_MASK, imgsz=IMGSZ, conf=CONF, iou=NMS_IOU, num_classes=NUM_CLASSES, score_sigmoid=SCORE_SIGMOID, ) object_cls = int(os.getenv("OBJECT_CLASS_ID", "0")) object_tracker = ByteTracker( track_high_thresh=TRACK_HIGH_THRESH, track_low_thresh=TRACK_LOW_THRESH, match_thresh=TRACK_MATCH_THRESH, track_buffer=TRACK_BUFFER, min_hits=TRACK_MIN_HITS, ) # Per-track sequence state: OUT→IN / IN-only → IN; IN→OUT / OUT-only → OUT. # Keys survive ID switches via inheritance (see _inherit_prev). object_cross_state = {} recent_cross_in = deque() recent_cross_out = deque() detect_snapshot_ids = set() object_cross_flash1 = {} object_cross_flash2 = {} line_pulse = count_in_pulse = count_out_pulse = 0 popups = [] session_start = time.time() frame_idx = 0 inf_ms = 0.0 prev_gray = None frames_since_infer = 0 video_writer = None crossing_times = deque() # Overlay shows daily store totals (resume-safe, resets on counting-day change). counter_in = store.current_count_in counter_out = store.current_count_out last_snapshot_cleanup = 0.0 counting_active = True status_nyala_active = True status_webhook_value = STATUS_WEBHOOK_INACTIVE_VALUE last_control_poll = 0.0 control_socket = None if CONTROL_ENABLED: if not Path(CONTROL_FILE).exists(): write_control_file(CONTROL_DEFAULT_COUNTING) counting_active = read_counting_flag(CONTROL_DEFAULT_COUNTING) print( f"Runtime control enabled | file={CONTROL_FILE} | " f"counting={'ON' if counting_active else 'OFF'}" ) if CONTROL_SOCKET_ENABLED: try: control_socket = start_control_socket() except Exception as exc: print(f"[{now_str()}] Failed to start control socket: {exc}") if STATUS_WEBHOOK_ENABLED: status_webhook_value = read_status_webhook() status_nyala_active = status_webhook_allows_counting(status_webhook_value) print( f"Status webhook enabled | file={STATUS_WEBHOOK_FILE} | " f"status={status_webhook_value}" ) cap, w, h, fps = connect_stream(SOURCE) if cap is None: store.shutdown() model.release() return line_y1 = resolve_line_y1(h) line_y2 = resolve_line_y2(h) print( f"RKNN+ByteTrack counter | {w}x{h} @ {fps}fps | " f"line1 y={line_y1} (IN ↓) line2 y={line_y2} (OUT ↑) | " f"seq: OUT→IN/IN-only→in, IN→OUT/OUT-only→out" ) print(f"Model: {MODEL_PATH} | imgsz={IMGSZ} | core_mask={CORE_MASK}") print( f"ByteTrack: high_thresh={TRACK_HIGH_THRESH} low_thresh={TRACK_LOW_THRESH} " f"match_thresh={TRACK_MATCH_THRESH} buffer={TRACK_BUFFER}" ) print(f"DB: {DB_PATH}") print(f"State: {STATE_FILE}") if RECORD_VIDEO: if RECORD_MODE == "segment": video_writer = VideoSegmentWriter(OUTPUT_DIR, w, h, fps, VIDEO_SEGMENT_SEC) print( f"Recording (segment): {OUTPUT_DIR} every {VIDEO_SEGMENT_SEC}s" ) else: video_writer = VideoSessionWriter( shm_dir=SHM_DIR, video_output_dir=VIDEO_OUTPUT_DIR, w=w, h=h, fps=fps if fps and fps > 1 else OUTPUT_FPS, end_delay_sec=RECORD_END_DELAY, retention_days=RECORD_RETENTION_DAYS, crf=RECORD_CRF, preset=RECORD_PRESET, discard_empty=RECORD_DISCARD_EMPTY, ) print( f"Recording (session): SHM={SHM_DIR} → out={VIDEO_OUTPUT_DIR} | " f"end_delay={RECORD_END_DELAY}s discard_empty={RECORD_DISCARD_EMPTY} | " f"ffmpeg={'yes' if shutil.which('ffmpeg') else 'NO'}" ) # If webhook already active at startup, begin a session immediately. if STATUS_WEBHOOK_ENABLED and status_nyala_active: video_writer.note_session_start( status_webhook_value, store.get_counting_date() ) elif not STATUS_WEBHOOK_ENABLED and counting_active: video_writer.note_session_start("SESSION", store.get_counting_date()) reconnect_count = 0 while not shutdown_requested: ret, frame = cap.read() if not ret: if not IS_LIVE: break reconnect_count += 1 stream_error_log.emit( f"Stream dropped (attempt {reconnect_count}), " f"reconnecting in {RECONNECT_DELAY_SEC}s..." ) cap.release() time.sleep(RECONNECT_DELAY_SEC) cap, w, h, fps = connect_stream(SOURCE) if cap is None: break line_y1 = resolve_line_y1(h) line_y2 = resolve_line_y2(h) continue now = time.time() elapsed = now - session_start mono = time.monotonic() object_crossed_frame = False cross_events_frame = [] detect_events_frame = [] if (CONTROL_ENABLED or STATUS_WEBHOOK_ENABLED) and ( now - last_control_poll ) >= CONTROL_POLL_SEC: last_control_poll = now if CONTROL_ENABLED: new_flag = read_counting_flag(CONTROL_DEFAULT_COUNTING) if new_flag != counting_active: counting_active = new_flag print( f"[{now_str()}] Counting " f"{'RESUMED' if counting_active else 'PAUSED'} via control file" ) if ( isinstance(video_writer, VideoSessionWriter) and not STATUS_WEBHOOK_ENABLED ): if counting_active: video_writer.note_session_start( "SESSION", store.get_counting_date() ) else: video_writer.note_session_end() if STATUS_WEBHOOK_ENABLED: new_status = read_status_webhook() if new_status != status_webhook_value: prev_status = status_webhook_value status_webhook_value = new_status status_nyala_active = status_webhook_allows_counting( status_webhook_value ) print(f"[{now_str()}] Status webhook {status_webhook_value}") if isinstance(video_writer, VideoSessionWriter): was_active = status_webhook_allows_counting(prev_status) if status_nyala_active and not was_active: video_writer.note_session_start( status_webhook_value, store.get_counting_date() ) elif was_active and not status_nyala_active: video_writer.note_session_end() # Raw frames only (before overlay). Session writer encodes here every frame. if isinstance(video_writer, VideoSessionWriter): video_writer.feed_frame(frame) counting_allowed = counting_active and status_nyala_active skip_inference = False if not counting_allowed: skip_inference = True elif MOTION_DETECTION_ENABLED: gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) if prev_gray is not None: diff = cv2.absdiff(gray, prev_gray) moved = int(np.count_nonzero(diff > MOTION_PIXEL_DELTA)) moved_frac = moved / diff.size skip_inference = moved_frac < MOTION_MIN_AREA_FRAC if frames_since_infer >= MOTION_HEARTBEAT_FRAMES: skip_inference = False prev_gray = gray detections = [] if not skip_inference: frames_since_infer = 0 inf_start = time.time() detections = model(frame) inf_ms = inf_ms * 0.9 + (time.time() - inf_start) * 1000 * 0.1 else: frames_since_infer += 1 object_boxes_xyxy = [] object_scores = [] object_kpts_list = [] object_cx_list = [] object_cy_list = [] for det in detections: bbox = det["bbox"] score = det["score"] cls_id = det["cls"] kpts = det["keypoints"] cx = (bbox[0] + bbox[2]) / 2.0 cy = (bbox[1] + bbox[3]) / 2.0 if cls_id == object_cls: object_boxes_xyxy.append(bbox) object_scores.append(score) object_kpts_list.append(kpts) object_cx_list.append(cx) object_cy_list.append(cy) object_boxes_xyxy = np.array(object_boxes_xyxy, dtype=np.float32).reshape(-1, 4) object_scores = np.array(object_scores, dtype=np.float32) object_track_map, object_det_to_track, object_lost_map = object_tracker.update( object_boxes_xyxy, object_scores ) if os.getenv("DEBUG_TRACKING", "").lower() == "true": if len(object_boxes_xyxy) > 0: scores_str = ( f" scores: {object_scores.round(3).tolist()}" if len(object_scores) > 0 else "" ) tracks_str = ( f" det->track: {dict(object_det_to_track)}" if object_det_to_track else "" ) crossing_str = "" if object_cross_state: line1_ids = sorted( tid for tid, st in object_cross_state.items() if st.get("seen_in") ) line2_ids = sorted( tid for tid, st in object_cross_state.items() if st.get("seen_out") ) if line1_ids: crossing_str += f" line1_crossed: {line1_ids}" if line2_ids: crossing_str += f" line2_crossed: {line2_ids}" print( f"[DEBUG F{frame_idx}] dets={len(object_boxes_xyxy)} " f"tracks={len(object_track_map)} " f"line_y1={line_y1} line_y2={line_y2}{scores_str}" f"{tracks_str}{crossing_str}" ) # --- crossing detection only on DETECTED objects this frame --- # (tracker still updates every frame to preserve IDs, but we do NOT # count on Kalman-predicted/coasting tracks to avoid double counts) for di in range(len(object_boxes_xyxy)): tid = object_det_to_track.get(di) if tid is None: continue cx = object_cx_list[di] cy = object_cy_list[di] if tid not in detect_snapshot_ids: detect_snapshot_ids.add(tid) detect_events_frame.append(tid) inherited_this_frame = False if tid not in object_tracked: inherited = _inherit_prev( object_tracked, tid, cx, cy, mono, INHERIT_SEC, INHERIT_PX ) if inherited is not None: object_tracked[tid] = inherited[:3] src_tid = inherited[3] if src_tid in object_cross_state: object_cross_state[tid] = _copy_cross_state( object_cross_state[src_tid] ) inherited_this_frame = True if os.getenv("DEBUG_TRACKING", "").lower() == "true": src_st = object_cross_state.get(tid, {}) print( f"[DEBUG F{frame_idx}] INHERIT prev for new tid={tid} " f"from tid={src_tid} ({inherited[0]:.1f},{inherited[1]:.1f})" f" seen_in={src_st.get('seen_in', False)}" f" seen_out={src_st.get('seen_out', False)}" f" counted={src_st.get('counted', False)}" ) if tid not in object_cross_state: object_cross_state[tid] = _default_cross_state() st = object_cross_state[tid] if tid in object_tracked: prev_cy = object_tracked[tid][1] # Line 1 = IN, line 2 = OUT. # Arm (no count): OUT top→down, IN bottom→up. # Count: IN top→down → IN (OUT→IN or IN-only) # OUT bottom→up → OUT (IN→OUT or OUT-only) out_down = ( crossed_top_down(prev_cy, cy, line_y2) and not st["seen_out"] ) in_up = ( crossed_bottom_up(prev_cy, cy, line_y1) and not st["seen_in"] ) in_down = ( crossed_top_down(prev_cy, cy, line_y1) and not st["seen_in"] ) out_up = ( crossed_bottom_up(prev_cy, cy, line_y2) and not st["seen_out"] ) if out_down or in_up or in_down or out_up: if out_down: st["seen_out"] = True if in_up: st["seen_in"] = True if in_down: st["seen_in"] = True if out_up: st["seen_out"] = True # Prefer decisive motion if both count triggers fire in one jump. count_in = in_down and not st["counted"] count_out = out_up and not st["counted"] if STATUS_WEBHOOK_ENABLED: # Webhook IN → only IN counts; OUT → only OUT counts. if status_webhook_value == "IN": count_out = False elif status_webhook_value == "OUT": count_in = False if count_in and count_out: if cy >= prev_cy: count_out = False else: count_in = False direction = "in" if count_in else ("out" if count_out else None) if direction is None: # Arm-only pass (OUT↓ or IN↑). No count yet. if os.getenv("DEBUG_TRACKING", "").lower() == "true": arm = [] if out_down: arm.append("out_down") if in_up: arm.append("in_up") print( f"[DEBUG F{frame_idx}] CROSS ARM: tid={tid} " f"prev_cy={prev_cy:.1f} -> cy={cy:.1f} " f"arm={'+'.join(arm)} " f"seen_in={st['seen_in']} seen_out={st['seen_out']}" ) object_tracked[tid] = (cx, cy, mono) continue recent = recent_cross_in if direction == "in" else recent_cross_out if is_duplicate_cross(recent, cx, frame_idx): st["counted"] = True if os.getenv("DEBUG_TRACKING", "").lower() == "true": print( f"[DEBUG F{frame_idx}] DUP CROSS IGNORED: tid={tid} " f"cx={cx:.1f} dir={direction}" ) object_tracked[tid] = (cx, cy, mono) continue if os.getenv("DEBUG_TRACKING", "").lower() == "true": line_n = 1 if direction == "in" else 2 line_y = line_y1 if direction == "in" else line_y2 print( f"[DEBUG F{frame_idx}] CROSS DETECTED: tid={tid} " f"prev_cy={prev_cy:.1f} -> cy={cy:.1f} " f"line={line_n} y={line_y} dir={direction}" ) recent.append((frame_idx, cx)) st["counted"] = True _, _, counted = store.record_object_crossing(tid, direction) counter_in = store.current_count_in counter_out = store.current_count_out if not counted: # Dedup in the store rejected this event; skip overlay/CSV. object_tracked[tid] = (cx, cy, mono) continue if direction == "in": count_in_pulse = COUNT_PULSE_FRAMES else: count_out_pulse = COUNT_PULSE_FRAMES if cross_logger: cross_logger.write_row( [ store.get_counting_date(), frame_idx, direction, tid, ] ) object_crossed_frame = True cross_events_frame.append((tid, direction)) crossing_times.append(mono) if isinstance(video_writer, VideoSessionWriter): video_writer.note_crossing() object_cross_flash1[tid] = CROSS_FLASH_FRAMES object_cross_flash2[tid] = CROSS_FLASH_FRAMES popups.append( { "x": int(cx) - 12, "y": int(cy), "born": frame_idx, "text": "+1", } ) object_tracked[tid] = (cx, cy, mono) for tid, (cx, cy) in object_lost_map.items(): if tid not in object_tracked: object_tracked[tid] = (cx, cy, mono) for di in range(len(object_boxes_xyxy)): tid = object_det_to_track.get(di) if tid is None: continue bbox = object_boxes_xyxy[di] x1, y1, x2, y2 = int(bbox[0]), int(bbox[1]), int(bbox[2]), int(bbox[3]) flash = max( object_cross_flash1.get(tid, 0), object_cross_flash2.get(tid, 0), ) color = C_GREEN if flash > 0 else C_OBJECT_BOX cv2.rectangle(frame, (x1, y1), (x2, y2), color, 3 if flash > 0 else 2) draw_pill(frame, f"ID {tid}", x1, y1 - 4, color) kpts = object_kpts_list[di] if di < len(object_kpts_list) else None if kpts is not None: draw_skeleton_bold(frame, kpts) if object_crossed_frame: line_pulse = LINE_PULSE_FRAMES # Keep overlay aligned with daily store (also resets after cutoff). counter_in = store.current_count_in counter_out = store.current_count_out while crossing_times and mono - crossing_times[0] > RATE_WINDOW_SEC: crossing_times.popleft() rate = (len(crossing_times) / RATE_WINDOW_SEC * 60) if crossing_times else 0.0 draw_elegant_counting_line(frame, line_y1, w, line_pulse, label="LINE IN") draw_elegant_counting_line(frame, line_y2, w, line_pulse, label="LINE OUT") draw_line_count(frame, w, line_y1, counter_in, "IN", C_GREEN, count_in_pulse, above=True) draw_line_count(frame, w, line_y2, counter_out, "OUT", C_OBJECT_BOX, count_out_pulse, above=False) draw_hud( frame, w, counter_in + counter_out, counter_in, counter_out, elapsed, rate, ) draw_footer( frame, w, h, frame_idx, "LIVE" if IS_LIVE else "FILE", inf_ms, Path(MODEL_PATH).name, ) popups = draw_popups(frame, popups, frame_idx) if (CONTROL_ENABLED or STATUS_WEBHOOK_ENABLED) and not counting_allowed: if STATUS_WEBHOOK_ENABLED and not status_nyala_active: badge = status_webhook_value else: badge = "COUNTING PAUSED" (bw, bh), _ = cv2.getTextSize(badge, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2) bx = w // 2 - bw // 2 overlay_rect(frame, bx - 14, 48, bx + bw + 14, 48 + bh + 18, C_PANEL, alpha=0.75) cv2.rectangle(frame, (bx - 14, 48), (bx + bw + 14, 48 + bh + 18), C_ACCENT, 2) cv2.putText(frame, badge, (bx, 48 + bh + 6), cv2.FONT_HERSHEY_SIMPLEX, 0.6, C_ACCENT, 2, cv2.LINE_AA) for flash_store in ( object_cross_flash1, object_cross_flash2, ): for tid in list(flash_store): flash_store[tid] -= 1 if flash_store[tid] <= 0: del flash_store[tid] line_pulse = max(0, line_pulse - 1) count_in_pulse = max(0, count_in_pulse - 1) count_out_pulse = max(0, count_out_pulse - 1) if isinstance(video_writer, VideoSegmentWriter): video_writer.write(frame) if LIVE_STREAM_ENABLED and frame_idx % LIVE_STREAM_EVERY_N == 0: try: Path(LIVE_STREAM_FRAME_PATH).parent.mkdir(parents=True, exist_ok=True) ok, jpeg = cv2.imencode( ".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, LIVE_STREAM_QUALITY] ) if ok: tmp_path = f"{LIVE_STREAM_FRAME_PATH}.tmp" with open(tmp_path, "wb") as f: f.write(jpeg.tobytes()) os.replace(tmp_path, LIVE_STREAM_FRAME_PATH) except Exception: pass if (SAVE_DETECT_SNAPSHOT and detect_events_frame) or ( SAVE_CROSS_SNAPSHOT and cross_events_frame ): try: ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] if SAVE_DETECT_SNAPSHOT and detect_events_frame: detect_dir = Path(CROSS_SNAPSHOT_DIR) / "detect" detect_dir.mkdir(parents=True, exist_ok=True) for tid in detect_events_frame: fname = f"{ts}_detect_id{tid}_f{frame_idx}.jpg" cv2.imwrite( str(detect_dir / fname), frame, [cv2.IMWRITE_JPEG_QUALITY, CROSS_SNAPSHOT_QUALITY], ) if SAVE_CROSS_SNAPSHOT and cross_events_frame: cross_dir = Path(CROSS_SNAPSHOT_DIR) / "cross" cross_dir.mkdir(parents=True, exist_ok=True) for tid, direction in cross_events_frame: fname = f"{ts}_{direction}_id{tid}_f{frame_idx}.jpg" cv2.imwrite( str(cross_dir / fname), frame, [cv2.IMWRITE_JPEG_QUALITY, CROSS_SNAPSHOT_QUALITY], ) if now - last_snapshot_cleanup >= CROSS_SNAPSHOT_CLEANUP_SEC: cleanup_snapshots( CROSS_SNAPSHOT_DIR, CROSS_SNAPSHOT_MAX_FILES, CROSS_SNAPSHOT_MAX_AGE_DAYS, ) last_snapshot_cleanup = now except Exception as exc: print(f"[{now_str()}] Failed to save snapshot: {exc}") frame_idx += 1 prune_stale_tracks(object_tracked, mono) cap.release() if isinstance(video_writer, VideoSessionWriter): video_writer.shutdown() elif video_writer is not None: video_writer.release() if cross_logger: cross_logger.close() if control_socket is not None: try: control_socket.close() except Exception: pass model.release() store.shutdown() print("\n=== Daily Counter Summary (SQLite) ===") print(f"Database: {DB_PATH}") object_tracked = {} if __name__ == "__main__": run()