#!/usr/bin/env python3 """Live dashboard for chicken-counter pipeline using Flask. Serve from project root: python3 dashboard.py [--port 8080] [--date 2026-06-10] [--db chicken_counts.db] """ from __future__ import annotations import argparse import json import logging import sqlite3 import threading from pathlib import Path from flask import Flask, Response, jsonify, render_template, send_file # Suppress Flask request logs log = logging.getLogger("werkzeug") log.setLevel(logging.ERROR) DEFAULT_SHM_DIR = "/dev/shm" DEFAULT_PORT = 8080 app = Flask(__name__) app.config["shm_dir"] = DEFAULT_SHM_DIR app.config["poll_ms"] = 1000 app.config["run_date"] = "" app.config["db_path"] = "" _db_conn = None _db_lock = threading.Lock() def _get_db(): global _db_conn db = app.config["db_path"] if not db or not Path(db).exists(): return None with _db_lock: if _db_conn is None: _db_conn = sqlite3.connect(db, 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 _discover_cameras(): shm = Path(app.config["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 @app.route("/") def index(): return render_template( "index.html", poll_ms=app.config["poll_ms"], shm_dir=app.config["shm_dir"], date=app.config["run_date"] or "today", db_path=app.config["db_path"], ) @app.route("/api/cameras") def api_cameras(): return jsonify({"cameras": _discover_cameras()}) @app.route("/shm//stats.json") def shm_stats(camera_id): stats_path = Path(app.config["shm_dir"]) / f"chicken_counter_{camera_id}" / "stats.json" if not stats_path.exists(): return jsonify({"error": "not found"}), 404 return jsonify(json.loads(stats_path.read_text())) @app.route("/shm//frame.jpg") def shm_frame(camera_id): frame_path = Path(app.config["shm_dir"]) / f"chicken_counter_{camera_id}" / "frame.jpg" if not frame_path.exists(): return jsonify({"error": "not found"}), 404 return send_file(frame_path, mimetype="image/jpeg", max_age=0, download_name=None) def _get_db(): db = app.config["db_path"] if not db or not Path(db).exists(): return None conn = sqlite3.connect(db) conn.row_factory = sqlite3.Row return conn @app.route("/api/db/summary") def db_summary(): conn = _get_db() if not conn: return jsonify({}) 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() return jsonify(dict(row)) @app.route("/api/db/history") def db_history(): conn = _get_db() if not conn: return jsonify([]) 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() return jsonify([dict(r) for r in rows]) @app.route("/api/db/date/") def db_date(date): conn = _get_db() if not conn: return jsonify({}) 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() return jsonify({"date": date, "total": dict(total), "cameras": [dict(r) for r in cameras]}) @app.route("/api/db/camera/") def db_camera(camera_id): conn = _get_db() if not conn: return jsonify([]) 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 """, (camera_id,)).fetchall() return jsonify([dict(r) for r in rows]) @app.route("/api/db/location/") def db_location(location): conn = _get_db() if not conn: return jsonify({}) 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 """, (location,)).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=? """, (location,)).fetchone() return jsonify({"location": location, "summary": dict(summary), "history": [dict(r) for r in history]}) def main(): 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=500) parser.add_argument("--date", default="", help="Processing date") parser.add_argument("--db", default="chicken_counts.db", help="SQLite database path") parser.add_argument("--debug", action="store_true") args = parser.parse_args() app.config["shm_dir"] = args.shm_dir app.config["poll_ms"] = args.poll_ms app.config["run_date"] = args.date app.config["db_path"] = str(Path(args.db).resolve()) if args.db else "" print(f"[dashboard] http://0.0.0.0:{args.port} shm={args.shm_dir} db={args.db}" + (f" date={args.date}" if args.date else "")) app.run(host="0.0.0.0", port=args.port, debug=args.debug, threaded=True) if __name__ == "__main__": main()