Files
chicken-counting-sukawarna-det/dashboard.py
T

591 lines
23 KiB
Python

#!/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/<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/<id>
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/<name>
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/<date> - Mortality breakdown for a specific date (YYYY-MM-DD).
GET /api/mortality/image/<name> - 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/<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/<filename> ---
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()