#!/usr/bin/env python3 """Live dashboard for chicken-counter pipeline.""" from __future__ import annotations import argparse import json import mimetypes import re import sqlite3 import threading import time from datetime import datetime, timezone from http.server import HTTPServer, SimpleHTTPRequestHandler from pathlib import Path from socketserver import ThreadingMixIn from urllib.parse import parse_qs, unquote, urlparse class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True DEFAULT_SHM_DIR = "/dev/shm" DEFAULT_PORT = 8080 TEMPLATE_DIR = Path(__file__).resolve().parent / "templates" _db_conn = None _db_lock = threading.Lock() _db_path = "" _mortality_dirs: list[Path] = [] def _load_initial_cycle_start_date() -> str: """Read cycle_start_date from configs/cycle7_batch_optimized.yaml if available.""" cfg_path = Path(__file__).resolve().parent / "configs" / "cycle7_batch_optimized.yaml" if cfg_path.exists(): content = cfg_path.read_text(encoding="utf-8") match = re.search(r"^\s*cycle_start_date:\s*['\"]?([^'\"\s#]+)['\"]?", content, re.MULTILINE) if match: return match.group(1).strip() return "2026-05-22" _cycle_start_date: str = _load_initial_cycle_start_date() def _persist_cycle_start_date(new_date: str) -> bool: """Update in-memory cycle_start_date and save to config YAML files.""" global _cycle_start_date _cycle_start_date = new_date updated_any = False for cfg_name in ("cycle7_batch_optimized.yaml", "cycle7_batch.yaml"): cfg_path = Path(__file__).resolve().parent / "configs" / cfg_name if cfg_path.exists(): content = cfg_path.read_text(encoding="utf-8") pattern = r"^([ \t]*cycle_start_date:[ \t]*)(?:['\"]?)([^'\"\r\n#]+)(?:['\"]?)([ \t]*(?:#.*)?)$" def replacer(match: re.Match) -> str: prefix = match.group(1) comment = match.group(3) or "" if comment and not comment.startswith(" "): comment = f" {comment.lstrip()}" if not comment.startswith(" "): comment = f" {comment}" return f'{prefix}"{new_date}"{comment}' new_content, count = re.subn(pattern, replacer, content, count=1, flags=re.MULTILINE) if count > 0: cfg_path.write_text(new_content, encoding="utf-8") updated_any = True print(f"[dashboard] 📅 Updated cycle_start_date to: {new_date} (persisted in configs: {updated_any})") return updated_any def _calc_cycle_info(target_date_str: str) -> dict: if not _cycle_start_date or not target_date_str: return {} try: from datetime import date as date_type start_d = date_type.fromisoformat(str(_cycle_start_date)) run_d = date_type.fromisoformat(str(target_date_str)) c_day = (run_d - start_d).days stage = "early_cycle" if 0 <= c_day <= 15 else ("mid_cycle" if c_day >= 16 else "pre_cycle") return {"cycle_day": c_day, "stage": stage} except Exception: return {} def _get_db(): global _db_conn if not _db_path: return None with _db_lock: if _db_conn is None: _db_conn = sqlite3.connect(_db_path, check_same_thread=False) _db_conn.row_factory = sqlite3.Row _db_conn.execute("PRAGMA journal_mode=WAL") _db_conn.execute("PRAGMA cache_size=-8000") return _db_conn def _init_db(db_path: str) -> None: conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL") conn.execute("""CREATE TABLE IF NOT EXISTS batch_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, location TEXT NOT NULL, camera_id TEXT NOT NULL, total_entered INTEGER NOT NULL DEFAULT 0, frames_processed INTEGER NOT NULL DEFAULT 0, elapsed_seconds REAL NOT NULL DEFAULT 0.0, stopped_reason TEXT NOT NULL DEFAULT '', source_video TEXT NOT NULL DEFAULT '', generated_at TEXT NOT NULL DEFAULT '', UNIQUE(date, location, camera_id))""") conn.commit() conn.close() def _discover_cameras(shm_dir): shm = Path(shm_dir) cameras = [] if shm.is_dir(): for entry in sorted(shm.iterdir()): if entry.is_dir() and entry.name.startswith("chicken_counter_"): cameras.append(entry.name[len("chicken_counter_"):]) return cameras class DashboardHandler(SimpleHTTPRequestHandler): shm_dir = DEFAULT_SHM_DIR poll_ms = 1000 run_date = "" def log_message(self, format, *args): pass def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With") self.send_header("Content-Length", "0") self.end_headers() def do_GET(self): try: self._handle() except (BrokenPipeError, ConnectionResetError): pass def do_POST(self): try: parsed = urlparse(self.path) path = unquote(parsed.path) if path == "/api/config/cycle_start_date": self._handle_set_cycle_start_date() return self._send_error(404) except (BrokenPipeError, ConnectionResetError): pass def _handle(self): parsed = urlparse(self.path) path = unquote(parsed.path) if path == "/": self._serve_html() return if path.startswith("/stream/"): self._handle_stream(path) return if path in ("/api/status", "/api/system/status"): self._handle_status() return if path == "/api/cameras": self._send_json({"cameras": _discover_cameras(self.shm_dir)}) return if path == "/api/config/cycle_start_date": query = parse_qs(parsed.query) if "set" in query and query["set"]: new_date = query["set"][0].strip() try: from datetime import date as date_type date_type.fromisoformat(new_date) persisted = _persist_cycle_start_date(new_date) self._send_json({ "status": "ok", "message": f"Cycle start date successfully set to {new_date}", "cycle_start_date": _cycle_start_date, "persisted": persisted, }) return except ValueError as err: self._send_json({"error": f"Invalid date format (expected YYYY-MM-DD): {err}"}, status_code=400) return self._send_json({"cycle_start_date": _cycle_start_date}) return if path == "/api/config": self._send_json({ "cycle_start_date": _cycle_start_date, "shm_dir": self.shm_dir, "db_path": _db_path, }) return if path.startswith("/api/db/"): self._handle_db(path) return if path.startswith("/api/mortality"): self._handle_mortality(path) return if path.startswith("/shm/"): self._handle_shm(path) return self._send_error(404) def _handle_set_cycle_start_date(self): content_length = int(self.headers.get("Content-Length", 0)) if content_length > 0: body = self.rfile.read(content_length).decode("utf-8") try: data = json.loads(body) new_date = str(data.get("cycle_start_date", "")).strip() if new_date: from datetime import date as date_type date_type.fromisoformat(new_date) persisted = _persist_cycle_start_date(new_date) self._send_json({ "status": "ok", "message": f"Cycle start date successfully set to {new_date}", "cycle_start_date": _cycle_start_date, "persisted": persisted, }) return else: self._send_json({"error": "Missing 'cycle_start_date' in request body"}, status_code=400) return except ValueError as err: self._send_json({"error": f"Invalid date format (expected YYYY-MM-DD): {err}"}, status_code=400) return except Exception as err: self._send_json({"error": str(err)}, status_code=400) return self._send_json({"error": "Empty request body"}, status_code=400) def _handle_status(self): cams = _discover_cameras(self.shm_dir) now = time.time() active_streams = [] for cam in cams: stat_file = Path(self.shm_dir) / f"chicken_counter_{cam}" / "stats.json" if stat_file.is_file() and (now - stat_file.stat().st_mtime) < 15.0: active_streams.append(cam) is_running = len(active_streams) > 0 conn = _get_db() latest_date = None total_chickens = 0 if conn: row = conn.execute("SELECT MAX(date) AS latest_date, SUM(total_entered) AS total FROM batch_runs").fetchone() if row: latest_date = row["latest_date"] total_chickens = row["total"] or 0 self._send_json({ "status": "running" if is_running else "idle", "is_counting_active": is_running, "active_cameras": active_streams, "latest_counted_date": latest_date, "total_chickens_all_time": total_chickens, "cycle_start_date": _cycle_start_date, "timestamp": datetime.now(timezone.utc).isoformat(), }) def _handle_stream(self, path): cam_id = path[len("/stream/"):] frame_path = Path(self.shm_dir) / f"chicken_counter_{cam_id}" / "frame.jpg" if not frame_path.exists(): self._send_error(404) return self.send_response(200) self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame") self.send_header("Cache-Control", "no-cache") self.end_headers() last_mtime = 0 try: while True: try: mtime = frame_path.stat().st_mtime if mtime != last_mtime: last_mtime = mtime data = frame_path.read_bytes() self.wfile.write( b"--frame\r\n" b"Content-Type: image/jpeg\r\n" b"Content-Length: " + str(len(data)).encode() + b"\r\n\r\n" + data + b"\r\n" ) self.wfile.flush() except (FileNotFoundError, OSError): pass time.sleep(0.1) except (BrokenPipeError, ConnectionResetError): pass def _handle_shm(self, path): rel = path[len("/shm/"):] parts = rel.split("/", 1) if len(parts) < 2: self._send_error(404) return cam_id = parts[0] file = parts[1] fpath = Path(self.shm_dir) / f"chicken_counter_{cam_id}" / file if str(fpath.resolve()).startswith(str(Path(self.shm_dir).resolve())): if fpath.exists(): ct = "image/jpeg" if file.endswith(".jpg") else "application/json" self.send_response(200) self.send_header("Content-Type", ct) self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") self.end_headers() self.wfile.write(fpath.read_bytes()) return self._send_error(404) def _handle_db(self, path): conn = _get_db() if not conn: self._send_json({}) return if path == "/api/db/summary": row = conn.execute("SELECT COUNT(DISTINCT date) AS days, COUNT(DISTINCT location) AS locations, COUNT(*) AS total_runs, SUM(total_entered) AS total_chickens, ROUND(SUM(elapsed_seconds)/3600.0,1) AS total_hours FROM batch_runs").fetchone() d = dict(row) d["cycle_start_date"] = _cycle_start_date self._send_json(d) return if path == "/api/db/history": rows = conn.execute("SELECT date, location, COUNT(*) AS cams, SUM(total_entered) AS total, ROUND(SUM(elapsed_seconds)/60.0,1) AS minutes FROM batch_runs GROUP BY date, location ORDER BY date DESC, location LIMIT 50").fetchall() history = [] for r in rows: d = dict(r) d.update(_calc_cycle_info(d["date"])) history.append(d) self._send_json(history) return # /api/db/date/ prefix = "/api/db/date/" if path.startswith(prefix): date = path[len(prefix):] cameras = conn.execute("SELECT camera_id, total_entered, frames_processed, ROUND(elapsed_seconds,1) AS elapsed_seconds, stopped_reason, source_video, location FROM batch_runs WHERE date=? ORDER BY camera_id", (date,)).fetchall() total = conn.execute("SELECT SUM(total_entered) AS total, ROUND(SUM(elapsed_seconds)/60.0,1) AS minutes FROM batch_runs WHERE date=?", (date,)).fetchone() res = {"date": date, "total": dict(total), "cameras": [dict(r) for r in cameras]} res.update(_calc_cycle_info(date)) self._send_json(res) return # /api/db/camera/ prefix = "/api/db/camera/" if path.startswith(prefix): cam_id = path[len(prefix):] rows = conn.execute("SELECT date, location, total_entered, frames_processed, ROUND(elapsed_seconds,1) AS elapsed_seconds, stopped_reason FROM batch_runs WHERE camera_id=? ORDER BY date DESC LIMIT 50", (cam_id,)).fetchall() self._send_json([dict(r) for r in rows]) return # /api/db/location/ prefix = "/api/db/location/" if path.startswith(prefix): loc = path[len(prefix):] history = conn.execute("SELECT date, GROUP_CONCAT(camera_id,', ') AS cameras, SUM(total_entered) AS total, ROUND(SUM(elapsed_seconds)/60.0,1) AS minutes FROM batch_runs WHERE location=? GROUP BY date ORDER BY date DESC LIMIT 50", (loc,)).fetchall() summary = conn.execute("SELECT COUNT(DISTINCT date) AS days, SUM(total_entered) AS total, ROUND(SUM(elapsed_seconds)/3600.0,1) AS hours FROM batch_runs WHERE location=?", (loc,)).fetchone() self._send_json({"location": loc, "summary": dict(summary), "history": [dict(r) for r in history]}) return self._send_json({}) def _handle_mortality(self, path: str) -> None: """Serve mortality detection results. GET /api/mortality/latest - Most recent mortality_report.json across all dirs. GET /api/mortality/history - List of all mortality reports found (newest first). GET /api/mortality/date/ - Mortality breakdown for a specific date (YYYY-MM-DD). GET /api/mortality/image/ - Serve an output_*.jpg annotated image by filename. """ if not _mortality_dirs: self._send_json({"error": "No mortality directory configured. Start dashboard with --mortality-dir."}) return def _enrich_report(data: dict) -> dict: if "total_mortality_count" not in data and "results" in data: data["total_mortality_count"] = sum(r.get("count", 0) for r in data["results"]) return data def _find_report_paths() -> list[Path]: found = [] for mdir in _mortality_dirs: p = Path(mdir) if p.is_dir(): found.extend(list(p.rglob("mortality_report.json"))) return found # --- /api/mortality/history --- if path == "/api/mortality/history": results = [] for report_path in _find_report_paths(): try: data = json.loads(report_path.read_text(encoding="utf-8")) data["_dir"] = str(report_path.parent) data["_report_mtime"] = report_path.stat().st_mtime results.append(_enrich_report(data)) except (json.JSONDecodeError, OSError): pass results.sort(key=lambda x: x.get("_report_mtime", 0), reverse=True) self._send_json(results) return # --- /api/mortality/latest --- if path == "/api/mortality/latest": latest = None latest_mtime = 0.0 for report_path in _find_report_paths(): try: mtime = report_path.stat().st_mtime if mtime > latest_mtime: latest_mtime = mtime latest = json.loads(report_path.read_text(encoding="utf-8")) latest["_dir"] = str(report_path.parent) except (json.JSONDecodeError, OSError): pass if latest: self._send_json(_enrich_report(latest)) else: self._send_json({"error": "No mortality report found."}) return # --- /api/mortality/date/ --- prefix_date = "/api/mortality/date/" if path.startswith(prefix_date): target_date = path[len(prefix_date):] matched_reports = [] total_day_carcasses = 0 total_day_images = 0 all_results = [] for report_path in _find_report_paths(): try: data = json.loads(report_path.read_text(encoding="utf-8")) report_date = data.get("date") or time.strftime("%Y-%m-%d", time.localtime(report_path.stat().st_mtime)) if report_date == target_date or report_path.parent.name == target_date: enriched = _enrich_report(data) total_day_carcasses += enriched.get("total_mortality_count", 0) total_day_images += enriched.get("total_images", 0) all_results.extend(enriched.get("results", [])) matched_reports.append(enriched) except (json.JSONDecodeError, OSError): pass self._send_json({ "date": target_date, "total_mortality_count": total_day_carcasses, "total_images": total_day_images, "reports": matched_reports, "results": all_results, }) return # --- /api/mortality/image/ --- prefix_img = "/api/mortality/image/" if path.startswith(prefix_img): raw_name = path[len(prefix_img):] filename = Path(raw_name).name # Prevent path traversal attacks # Only allow serving output_*.jpg files for security if not (filename.startswith("output_") and filename.lower().endswith((".jpg", ".jpeg", ".png"))): self._send_error(403) return for mdir in _mortality_dirs: p = Path(mdir) if p.is_dir(): for img_path in (p.rglob(filename) if filename else []): if img_path.is_file(): try: data = img_path.read_bytes() ct = mimetypes.guess_type(filename)[0] or "image/jpeg" self.send_response(200) self.send_header("Content-Type", ct) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-cache") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) return except (FileNotFoundError, OSError): pass self._send_error(404) return self._send_error(404) def _serve_html(self): html_path = TEMPLATE_DIR / "index.html" html = html_path.read_text(encoding="utf-8") html = html.replace("{{ poll_ms }}", str(self.poll_ms)) html = html.replace("{{ shm_dir }}", self.shm_dir) html = html.replace("{{ date }}", self.run_date or "today") html = html.replace("{{ db_path }}", _db_path) data = html.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) def _send_json(self, obj, status_code=200): data = json.dumps(obj).encode("utf-8") self.send_response(status_code) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_error(self, code): self.send_response(code) self.send_header("Content-Length", "0") self.end_headers() def main(): global _db_path, _mortality_dirs parser = argparse.ArgumentParser(description="Chicken Counter live dashboard") parser.add_argument("--port", type=int, default=DEFAULT_PORT) parser.add_argument("--shm-dir", default=DEFAULT_SHM_DIR) parser.add_argument("--poll-ms", type=int, default=1000) parser.add_argument("--date", default="") parser.add_argument("--db", default="db/chicken_counts.db") parser.add_argument("--cycle-start-date", default="", help="Start date of cycle (Day 0) in YYYY-MM-DD format.") parser.add_argument( "--mortality-dir", action="append", dest="mortality_dirs", default=[], metavar="DIR", help="Directory containing mortality_report.json and output images. Repeatable for multiple dirs.", ) args = parser.parse_args() _mortality_dirs = [Path(d).resolve() for d in args.mortality_dirs] if args.cycle_start_date: global _cycle_start_date _cycle_start_date = args.cycle_start_date DashboardHandler.shm_dir = args.shm_dir DashboardHandler.poll_ms = args.poll_ms DashboardHandler.run_date = args.date _db_path = str(Path(args.db).resolve()) if args.db else "" if _db_path: _init_db(_db_path) server = ThreadingHTTPServer(("0.0.0.0", args.port), DashboardHandler) date_info = f" date={args.date}" if args.date else "" print(f"[dashboard] http://0.0.0.0:{args.port} shm={args.shm_dir} db={args.db}{date_info}") try: server.serve_forever() except KeyboardInterrupt: print("\n[dashboard] stopped") server.server_close() if __name__ == "__main__": main()