""" Edge production live counter — RTSP + YOLO TensorRT + line crossing. Replaces MQTT frigate-counter on Jetson with local LAN camera inference. """ from ultralytics import YOLO import cv2 import csv import numpy as np import os import signal import time from collections import deque from datetime import datetime from pathlib import Path from dotenv import load_dotenv load_dotenv() from batch_store import BatchStore # --- 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_batch.json') SOURCE = os.getenv('SOURCE', 'rtsp://user:pass@192.168.0.100:554/stream1') MODEL_PATH = os.getenv('MODEL_PATH', '/media/jetson/DATA/yolo11n.engine') CAMERA_NAME = os.getenv('CAMERA_NAME', 'CC1') OBJECT_LABEL = os.getenv('OBJECT_LABEL', 'ayam-potong') CLASS_AYAM = os.getenv('CLASS_AYAM', 'ayam') CLASS_TALENAN = os.getenv('CLASS_TALENAN', 'talenan') LINE_X = int(os.getenv('LINE_X')) if os.getenv('LINE_X') else None LINE_X_FRAC = float(os.getenv('LINE_X_FRAC', '0.5')) CROSS_DIRECTION = os.getenv('CROSS_DIRECTION', 'rtl').lower() IMGSZ = int(os.getenv('IMGSZ', '416')) HALF = os.getenv('HALF', 'true').lower() == 'true' CONF = float(os.getenv('CONF', '0.3')) DEVICE = int(os.getenv('DEVICE', '0')) TRACKER = os.getenv('TRACKER', 'bytetrack.yaml') DAILY_CUTOFF_TIME = os.getenv('DAILY_CUTOFF_TIME', '20:00') BATCH_TIMEOUT_SECONDS = float(os.getenv('BATCH_TIMEOUT_SECONDS', '300')) IGNORE_BATCH_LABEL_TIMEOUT = float(os.getenv('IGNORE_BATCH_LABEL_TIMEOUT_SECONDS', '30')) MIN_OBJECT_PER_BATCH = int(os.getenv('MIN_OBJECT_PER_BATCH', '60')) MIN_DURATION_PER_BATCH = int(os.getenv('MIN_DURATION_PER_BATCH', '60')) EXPORT_CSV = os.getenv('EXPORT_CSV', 'true').lower() == 'true' CROSS_CSV = os.getenv('CROSS_CSV', f'{OUTPUT_DIR}/batch_crossings.csv') 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')) FLUSH_EVERY_N_FRAMES = int(os.getenv('FLUSH_EVERY_N_FRAMES', '100')) 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')) OUTPUT_FPS = int(os.getenv('OUTPUT_FPS', '15')) LIVE_STREAM_ENABLED = os.getenv('LIVE_STREAM_ENABLED', 'false').lower() == 'true' LIVE_STREAM_FRAME_PATH = os.getenv('LIVE_STREAM_FRAME_PATH', '/dev/shm/jetson-counter/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')) RATE_WINDOW_SEC = int(os.getenv('RATE_WINDOW_SEC', '60')) CROSS_FLASH_FRAMES = 12 POPUP_LIFETIME = 20 LINE_PULSE_FRAMES = 12 COUNT_PULSE_FRAMES = 15 BATCH_PULSE_FRAMES = 20 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_AYAM_BOX = (0, 165, 255) C_TALENAN_BOX = (220, 120, 60) 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) def resolve_class_ids(names): name_to_id = {v: k for k, v in names.items()} missing = [n for n in (CLASS_AYAM, CLASS_TALENAN) if n not in name_to_id] if missing: raise ValueError(f'Model missing classes {missing}. Available: {list(names.values())}') return name_to_id[CLASS_AYAM], name_to_id[CLASS_TALENAN] def box_cx(box): return (int(box[0]) + int(box[2])) // 2 def resolve_line_x(frame_width): if LINE_X is not None: return LINE_X if LINE_X_FRAC != 0.5: return int(frame_width * LINE_X_FRAC) return frame_width // 2 def crossed_line(prev_cx, cx, line_x, direction=CROSS_DIRECTION): if direction == 'ltr': return prev_cx < line_x <= cx if direction == 'both': return (prev_cx > line_x >= cx) or (prev_cx < line_x <= cx) return prev_cx > line_x >= cx def now_str(): return datetime.now().strftime('%Y-%m-%d %H:%M:%S') def open_capture(source): if source.lower().startswith(('rtsp://', 'http://')): os.environ['OPENCV_FFMPEG_CAPTURE_OPTIONS'] = RTSP_FFMPEG_OPTIONS 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 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_x, h, pulse_remaining=0): 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, (line_x - offset, 0), (line_x - offset, h), color, 1, cv2.LINE_AA) cv2.line(img, (line_x + offset, 0), (line_x + offset, h), color, 1, cv2.LINE_AA) dash_len, gap = 18, 12 y = 0 while y < h: y_end = min(y + dash_len, h) cv2.line(img, (line_x, y), (line_x, y_end), C_LINE_CORE, 2, cv2.LINE_AA) y += dash_len + gap cv2.putText(img, 'COUNT LINE', (line_x - 46, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.42, C_LINE_CORE, 1, cv2.LINE_AA) def draw_hero_count(img, line_x, h, count, pulse_remaining=0): text = str(count) font = cv2.FONT_HERSHEY_SIMPLEX boost = 0.35 * (pulse_remaining / max(COUNT_PULSE_FRAMES, 1)) font_scale, thickness = 1.4 + boost, 3 (tw, th), _ = cv2.getTextSize(text, font, font_scale, thickness) pad = 14 tx, ty = line_x - tw // 2, h // 3 + th // 2 overlay_rect(img, tx - pad, ty - th - pad, tx + tw + pad, ty + pad, C_PANEL, alpha=0.78) cv2.rectangle(img, (tx - pad, ty - th - pad), (tx + tw + pad, ty + pad), C_LINE_CORE, 2) cv2.putText(img, text, (tx, ty), font, font_scale, C_GREEN, thickness, cv2.LINE_AA) def draw_hud(img, w, batch_num, batch_count, total_ayam, elapsed_sec, rate, camera_id, clock): bar_h = 52 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, 'BATCH', (16, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA) batch_label = str(batch_num) if batch_num else '—' cv2.putText(img, batch_label, (16, 44), cv2.FONT_HERSHEY_SIMPLEX, 0.9, C_ACCENT, 2, cv2.LINE_AA) cv2.putText(img, 'COUNT', (100, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA) cv2.putText(img, str(batch_count), (100, 44), cv2.FONT_HERSHEY_SIMPLEX, 0.9, C_GREEN, 2, cv2.LINE_AA) cv2.putText(img, 'TOTAL', (190, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA) cv2.putText(img, str(total_ayam), (190, 44), cv2.FONT_HERSHEY_SIMPLEX, 0.7, C_TEXT, 1, cv2.LINE_AA) cv2.putText(img, 'UPTIME', (280, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA) cv2.putText(img, f'{elapsed_sec / 3600:.1f}h', (280, 44), cv2.FONT_HERSHEY_SIMPLEX, 0.7, C_TEXT, 1, cv2.LINE_AA) cv2.putText(img, 'RATE', (380, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 1, cv2.LINE_AA) cv2.putText(img, f'{rate:.1f}/min', (380, 44), cv2.FONT_HERSHEY_SIMPLEX, 0.7, C_ACCENT, 1, cv2.LINE_AA) cv2.putText(img, clock, (w - 180, 36), cv2.FONT_HERSHEY_SIMPLEX, 0.55, C_TEXT, 1, cv2.LINE_AA) cv2.putText(img, f'CAM {camera_id}', (w - 180, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.45, C_MUTED, 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 def draw_batch_banner(img, w, batch_num, pulse_remaining): if pulse_remaining <= 0: return text = f'NEW BATCH {batch_num}' font = cv2.FONT_HERSHEY_SIMPLEX (tw, th), _ = cv2.getTextSize(text, font, 0.8, 2) x1, y1 = w // 2 - tw // 2 - 16, 62 x2, y2 = w // 2 + tw // 2 + 16, 62 + th + 20 overlay_rect(img, x1, y1, x2, y2, C_PANEL, alpha=0.7) cv2.rectangle(img, (x1, y1), (x2, y2), C_ACCENT, 2) cv2.putText(img, text, (w // 2 - tw // 2, 62 + th + 4), font, 0.8, C_ACCENT, 2, cv2.LINE_AA) 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}') print(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 return cap, w, h, fps return None, 0, 0, OUTPUT_FPS def run(): global shutdown_requested store = BatchStore( db_path=DB_PATH, state_file=STATE_FILE, camera_name=CAMERA_NAME, object_label=OBJECT_LABEL, cutoff_time=DAILY_CUTOFF_TIME, batch_timeout=BATCH_TIMEOUT_SECONDS, ignore_batch_label_timeout=IGNORE_BATCH_LABEL_TIMEOUT, min_object_per_batch=MIN_OBJECT_PER_BATCH, min_duration_per_batch=MIN_DURATION_PER_BATCH, logger=lambda msg: print(f'[{now_str()}] {msg}'), ) store.start_cutoff_watcher() cross_logger = None if EXPORT_CSV: cross_logger = CsvLogger(CROSS_CSV, ['batch', 'frame', 'timestamp', 'chicken_id']) model = YOLO(MODEL_PATH) ayam_cls, talenan_cls = resolve_class_ids(model.names) ayam_tracked = {} talenan_tracked = {} ayam_line_crossed = set() talenan_line_crossed = set() ayam_cross_flash = {} talenan_cross_flash = {} line_pulse = count_pulse = batch_pulse = 0 popups = [] session_start = time.time() frame_idx = 0 inf_ms = 0.0 video_writer = None crossing_times = deque() prev_gray = None cap, w, h, fps = connect_stream(SOURCE) if cap is None: store.shutdown() return line_x = resolve_line_x(w) print(f'Jetson counter | {w}x{h} @ {fps}fps | line x={line_x} | cross={CROSS_DIRECTION}') print(f'Model: {MODEL_PATH} | imgsz={IMGSZ} half={HALF}') print(f'DB: {DB_PATH}') print(f'State: {STATE_FILE}') if RECORD_VIDEO: video_writer = VideoSegmentWriter(OUTPUT_DIR, w, h, fps, VIDEO_SEGMENT_SEC) reconnect_count = 0 while not shutdown_requested: ret, frame = cap.read() if not ret: if not IS_LIVE: break reconnect_count += 1 print(f'Stream dropped (attempt {reconnect_count}), 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_x = resolve_line_x(w) continue now = time.time() elapsed = now - session_start mono = time.monotonic() ayam_crossed_frame = batch_closed_frame = batch_started_frame = False skip_inference = False if MOTION_DETECTION_ENABLED: gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) if prev_gray is not None: diff = cv2.absdiff(gray, prev_gray) mean_diff = cv2.mean(diff)[0] skip_inference = mean_diff < MOTION_THRESHOLD prev_gray = gray if not skip_inference: inf_start = time.time() results = model.track( frame, device=DEVICE, persist=True, conf=CONF, imgsz=IMGSZ, half=HALF, tracker=TRACKER, verbose=False, ) inf_ms = inf_ms * 0.9 + (time.time() - inf_start) * 1000 * 0.1 else: results = [None] r = results[0] if r is not None and r.boxes.id is not None: ids = r.boxes.id.int().tolist() boxes = r.boxes.xyxy.tolist() clss = r.boxes.cls.int().tolist() kpts_all = r.keypoints.xy.tolist() if r.keypoints else [] talenan_items, ayam_items = [], [] for i, (track_id, box, cls_id) in enumerate(zip(ids, boxes, clss)): cx = box_cx(box) x1, y1, x2, y2 = [int(v) for v in box] kpts = kpts_all[i] if i < len(kpts_all) else None item = (track_id, cx, x1, y1, x2, y2, kpts) if cls_id == talenan_cls: talenan_items.append(item) elif cls_id == ayam_cls: ayam_items.append(item) for track_id, cx, x1, y1, x2, y2, _ in talenan_items: if track_id in talenan_tracked: prev_cx, _ = talenan_tracked[track_id] if crossed_line(prev_cx, cx, line_x) and track_id not in talenan_line_crossed: talenan_line_crossed.add(track_id) if store.record_talenan_crossing(track_id): batch_closed_frame = True talenan_cross_flash[track_id] = CROSS_FLASH_FRAMES popups.append({'x': cx - 20, 'y': (y1 + y2) // 2, 'born': frame_idx, 'text': 'BATCH CLOSED'}) talenan_tracked[track_id] = (cx, mono) for track_id, cx, x1, y1, x2, y2, kpts in ayam_items: if track_id in ayam_tracked: prev_cx, _ = ayam_tracked[track_id] if crossed_line(prev_cx, cx, line_x) and track_id not in ayam_line_crossed: ayam_line_crossed.add(track_id) _, started_new = store.record_ayam_crossing(track_id) if cross_logger: cross_logger.write_row([ store.current_batch_number, frame_idx, datetime.now().isoformat(), track_id, ]) ayam_crossed_frame = True crossing_times.append(mono) if started_new: batch_started_frame = True ayam_cross_flash[track_id] = CROSS_FLASH_FRAMES popups.append({'x': cx - 12, 'y': (y1 + y2) // 2, 'born': frame_idx, 'text': '+1'}) ayam_tracked[track_id] = (cx, mono) if os.getenv('DEBUG_TRACKING', '').lower() == 'true': ayam_tracks_str = f' ayam_tracked: {sorted(ayam_tracked.keys())}' if ayam_tracked else '' talenan_tracks_str = f' talenan_tracked: {sorted(talenan_tracked.keys())}' if talenan_tracked else '' cx_sample = '' if ayam_items: track_id, cx, _, _, _, _, _ = ayam_items[0] prev = ayam_tracked.get(track_id, (None,))[0] if ayam_tracked.get(track_id) else None cx_sample = f' sample tid={track_id} prev_cx={prev} cx={cx}' print( f'[DEBUG F{frame_idx}] ayam_dets={len(ayam_items)} ' f'talenan_dets={len(talenan_items)} ' f'line_x={line_x}{talenan_tracks_str}{ayam_tracks_str}' f' crossed: a={sorted(ayam_line_crossed)} t={sorted(talenan_line_crossed)}' f'{cx_sample}' ) for track_id, cx, x1, y1, x2, y2, _ in talenan_items: flash = talenan_cross_flash.get(track_id, 0) color = C_GREEN if flash > 0 else C_TALENAN_BOX cv2.rectangle(frame, (x1, y1), (x2, y2), color, 3 if flash > 0 else 2) draw_pill(frame, f'TALENAN {track_id}', x1, y1 - 4, color) for track_id, cx, x1, y1, x2, y2, kpts in ayam_items: flash = ayam_cross_flash.get(track_id, 0) color = C_GREEN if flash > 0 else C_AYAM_BOX cv2.rectangle(frame, (x1, y1), (x2, y2), color, 3 if flash > 0 else 2) draw_pill(frame, f'ID {track_id}', x1, y1 - 4, color) if kpts is not None: draw_skeleton_bold(frame, kpts) if ayam_crossed_frame: line_pulse = LINE_PULSE_FRAMES count_pulse = COUNT_PULSE_FRAMES if batch_closed_frame: line_pulse = LINE_PULSE_FRAMES if batch_started_frame: batch_pulse = BATCH_PULSE_FRAMES batch_num = store.current_batch_number or 0 batch_count = store.current_batch_count display_total = store.display_total() 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_x, h, line_pulse) draw_hero_count(frame, line_x, h, batch_count, count_pulse) draw_hud(frame, w, batch_num, batch_count, display_total, elapsed, rate, CAMERA_NAME, now_str()) draw_batch_banner(frame, w, batch_num, batch_pulse) 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) for flash_store in (ayam_cross_flash, talenan_cross_flash): 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_pulse = max(0, count_pulse - 1) batch_pulse = max(0, batch_pulse - 1) if video_writer is not None: 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) _, jpeg = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, LIVE_STREAM_QUALITY]) with open(LIVE_STREAM_FRAME_PATH, 'wb') as f: f.write(jpeg.tobytes()) except Exception: pass frame_idx += 1 if frame_idx % FLUSH_EVERY_N_FRAMES == 0: print( f'[{now_str()}] Frame {frame_idx} | Batch {batch_num}: {batch_count} ' f'| Total: {display_total} | Uptime {elapsed / 3600:.2f}h' ) prune_stale_tracks(ayam_tracked, mono) prune_stale_tracks(talenan_tracked, mono) cap.release() if video_writer is not None: video_writer.release() if cross_logger: cross_logger.close() store.shutdown() print('\n=== Batch Summary (SQLite) ===') print(f'Database: {DB_PATH}') if __name__ == '__main__': run()