""" app.py TrackIQ Backend """ import os os.environ.setdefault("YOLO_CONFIG_DIR", "/tmp") import cv2, csv, json, threading, queue, uuid, mimetypes, subprocess, io, time from pathlib import Path from datetime import datetime from collections import defaultdict from typing import Dict import numpy as np from flask import (Flask, request, jsonify, Response, render_template, send_file, send_from_directory, abort, stream_with_context) try: from flask_cors import CORS except ImportError: def CORS(_app): return _app from ultralytics import YOLO app = Flask(__name__, static_folder="static", template_folder="templates") CORS(app) BASE = Path(__file__).parent UPLOAD_DIR = BASE / "uploads"; UPLOAD_DIR.mkdir(exist_ok=True) OUTPUT_DIR = BASE / "outputs"; OUTPUT_DIR.mkdir(exist_ok=True) LOG_DIR = BASE / "logs"; LOG_DIR.mkdir(exist_ok=True) MODELS_DIR = BASE / "models"; MODELS_DIR.mkdir(exist_ok=True) JOB_STATE_PATH = LOG_DIR / "jobs_state.json" # COCO_TO_LABEL = { 0: "Human", 1: "Bicycle", 2: "Vehicle", 3: "Motorcycle", 5: "Bus", 7: "Truck", 9: "Traffic light", 11: "Road sign", } DEFAULT_CLASSES = list(COCO_TO_LABEL.values()) CSV_FIELDS = [ "frame", "timestamp_sec", "scene_name", "group_id", "video_name", "track_id", "class_name", "confidence", "bbox_x1", "bbox_y1", "bbox_x2", "bbox_y2", "cx", "cy", "frame_width", "frame_height", "crossed_line", "direction", "speed_px_s" ] LOG_GROUP_ID = "Group_07" CLASS_COLORS = { "Vehicle": (255, 120, 80), "Motorcycle": (80, 200, 80), "Truck": (0, 180, 255), "Bus": (220, 80, 255), "Human": (236, 72, 153), "Bicycle": (6, 182, 212), "Traffic light": (239, 68, 68), "Road sign": (249, 115, 22), } # ── CORRECTION 2 : seuil de confiance abaissé + résolution augmentée ───────── CONF = 0.25 # WAS 0.40 (trop élevé pour webcam intérieure) IOU = 0.45 SKIP = 1 INFER_SZ = 640 _jobs: Dict[str, dict] = {} _sse_queues: Dict[str, queue.Queue] = {} _stats_lock = threading.Lock() _global_stats = { "total_frames": 0, "total_detections": 0, "detections_by_class": defaultdict(int), "scenes": [] } # ── Webcam state ────────────────────────────────────────────────────────────── _webcam_active = False _webcam_classes = [] _webcam_frame_count = 0 _webcam_detections = 0 _webcam_detections_by_class = defaultdict(int) _webcam_current_counts = defaultdict(int) _webcam_seen_track_ids = set() _webcam_lock = threading.Lock() # ── Chargement du modèle en arrière-plan ────────────────────────────────────── _model = None _model_ready = False _model_loading_started = False def _load_model_background(): global _model, _model_ready print("⏳ Chargement du modèle YOLO...") _model = _get_model() _model_ready = True print("✅ Modèle YOLO chargé !") def _ensure_model_loading(): global _model_loading_started if _model_ready or _model_loading_started: return _model_loading_started = True threading.Thread(target=_load_model_background, daemon=True).start() # ── Utiliser le modèle YOLO inclus dans le dépôt ────────────────────────────── def _get_model(key="yolo11n"): p = MODELS_DIR / f"{key}.pt" if not p.exists(): m = YOLO(f"{key}.pt") import shutil dl = Path(f"{key}.pt") if dl.exists(): shutil.move(str(dl), str(p)) return m return YOLO(str(p)) # Lancer le chargement dès le démarrage, sauf pendant les tests rapides du dashboard. if os.environ.get("TRACKIQ_DISABLE_AUTO_MODEL_LOAD") != "1": _ensure_model_loading() # ── Fonction utilitaire pour dessiner les détections ────────────────────────── def _draw_detections(frame, results, classes_filter=None): """Dessine les boîtes de détection sur le frame""" detection_count = 0 if results and results[0].boxes: boxes = results[0].boxes for box in boxes: cls = int(box.cls[0]) conf = float(box.conf[0]) class_name = COCO_TO_LABEL.get(cls) if class_name is None: continue if classes_filter and class_name not in classes_filter: continue detection_count += 1 x1, y1, x2, y2 = map(int, box.xyxy[0]) color = CLASS_COLORS.get(class_name, (255, 255, 255)) label = f"{class_name} {conf:.2f}" cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2) cv2.putText(frame, label, (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2) return frame, detection_count def _job_public(job, jid): return { "job_id": jid, "status": job.get("status", "unknown"), "name": job.get("name", ""), "path": job.get("path", ""), "output_path": job.get("output_path", ""), "log_path": job.get("log_path", ""), "frames": job.get("frames", 0), "processed_frames": job.get("processed_frames", 0), "progress": job.get("progress", 0), "detections": dict(job.get("detections", {})), "stats": job.get("stats", {}), "classes": job.get("classes", DEFAULT_CLASSES), "created_at": job.get("created_at"), "updated_at": job.get("updated_at"), } def _persist_jobs(): payload = {"jobs": [_job_public(job, jid) for jid, job in _jobs.items()]} try: JOB_STATE_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8") except Exception as e: print(f"[STATE] Could not persist jobs: {e}") def _restore_jobs(): if not JOB_STATE_PATH.exists(): return try: payload = json.loads(JOB_STATE_PATH.read_text(encoding="utf-8")) except Exception as e: print(f"[STATE] Could not restore jobs: {e}") return for item in payload.get("jobs", []): jid = item.get("job_id") path = item.get("path") if not jid or not path: continue status = item.get("status", "uploaded") if status == "processing": status = "uploaded" output_path = item.get("output_path") or str(OUTPUT_DIR / f"{jid}_output.mp4") if Path(output_path).exists() and item.get("stats"): status = "done" elif not str(path).startswith(("http://", "https://", "rtsp://")) and not Path(path).exists(): continue detections = defaultdict(int, item.get("detections") or item.get("stats", {}).get("unique_counts", {})) _jobs[jid] = { "status": status, "path": path, "name": item.get("name", ""), "output_path": output_path, "log_path": item.get("log_path") or str(LOG_DIR / f"{jid}.csv"), "detections": detections, "frames": item.get("frames", 0), "processed_frames": item.get("processed_frames", 0), "progress": 100 if status == "done" else item.get("progress", 0), "classes": item.get("classes", DEFAULT_CLASSES), "latest_frame": None, "latest_frame_no": 0, "frame_detections": [], "timeline": [], "stats": item.get("stats", {}), "created_at": item.get("created_at"), "updated_at": item.get("updated_at"), "lock": threading.Lock() } def _float_value(value, default=0.0): try: return float(value) except (TypeError, ValueError): return default def _int_value(value, default=0): try: return int(float(value)) except (TypeError, ValueError): return default def _clean_direction(value): direction = (value or "").strip() return "" if direction.lower() == "unknown" else direction def _build_scene_from_csv(csv_path): rows = [] with csv_path.open(newline="", encoding="utf-8") as fh: for row in csv.DictReader(fh): frame = _int_value(row.get("frame")) timestamp_sec = _float_value(row.get("timestamp_sec")) class_name = row.get("class_name") or "Unknown" scene_name = row.get("scene_name") or csv_path.stem.replace("trackiq_", "") video_name = row.get("video_name") or csv_path.name track_id = row.get("track_id") or f"row-{len(rows)}" rows.append({ "frame": frame, "timestamp_sec": timestamp_sec, "scene_name": scene_name, "group_id": LOG_GROUP_ID, "video_name": video_name, "track_id": track_id, "class_name": class_name, "confidence": _float_value(row.get("confidence")), "bbox_x1": _int_value(row.get("bbox_x1")), "bbox_y1": _int_value(row.get("bbox_y1")), "bbox_x2": _int_value(row.get("bbox_x2")), "bbox_y2": _int_value(row.get("bbox_y2")), "cx": _int_value(row.get("cx")), "cy": _int_value(row.get("cy")), "frame_width": _int_value(row.get("frame_width")), "frame_height": _int_value(row.get("frame_height")), "crossed_line": row.get("crossed_line") or "false", "direction": _clean_direction(row.get("direction")), "speed_px_s": _float_value(row.get("speed_px_s")), }) if not rows: return None rows.sort(key=lambda item: (item["frame"], item["class_name"], str(item["track_id"]))) scene_id = rows[0]["scene_name"] counted_track_ids = set() unique_counts = defaultdict(int) timeline = [] last_timeline_frame = None for det in rows: unique_key = (det["class_name"], str(det["track_id"])) if unique_key not in counted_track_ids: counted_track_ids.add(unique_key) unique_counts[det["class_name"]] += 1 if last_timeline_frame is None or det["frame"] != last_timeline_frame: last_timeline_frame = det["frame"] if len(timeline) < 60: timeline.append({ "frame": det["frame"], "ts": det["timestamp_sec"], **dict(unique_counts) }) total_frames = max(det["frame"] for det in rows) duration_s = max(det["timestamp_sec"] for det in rows) fps = round(total_frames / duration_s, 2) if duration_s > 0 else 0 generated_at = datetime.fromtimestamp(csv_path.stat().st_mtime).isoformat() stats = { "scene_id": scene_id, "video_name": rows[0]["video_name"], "total_frames": total_frames, "processed_frames": total_frames, "total_unique": sum(unique_counts.values()), "total_detections": sum(unique_counts.values()), "unique_counts": dict(unique_counts), "duration_s": round(duration_s, 2), "fps": fps, "generated_at": generated_at, "timeline": timeline, } return scene_id, rows, stats def _restore_data_csv_jobs(): data_dir = BASE / "data" if not data_dir.exists(): return for csv_path in sorted(data_dir.glob("*.csv")): try: restored = _build_scene_from_csv(csv_path) except Exception as e: print(f"[STATE] Could not restore CSV {csv_path}: {e}") continue if not restored: continue jid, frame_detections, stats = restored if jid in _jobs: job = _jobs[jid] if not job.get("frame_detections"): job["frame_detections"] = frame_detections if not job.get("timeline"): job["timeline"] = stats.get("timeline", []) if not job.get("stats"): job["stats"] = stats continue _jobs[jid] = { "status": "done", "path": str(csv_path), "name": stats["video_name"], "output_path": str(OUTPUT_DIR / f"{jid}_output.mp4"), "log_path": str(csv_path), "detections": defaultdict(int, stats["unique_counts"]), "frames": stats["total_frames"], "processed_frames": stats["processed_frames"], "progress": 100, "classes": DEFAULT_CLASSES, "latest_frame": None, "latest_frame_no": 0, "frame_detections": frame_detections, "timeline": stats["timeline"], "stats": stats, "created_at": stats["generated_at"], "updated_at": stats["generated_at"], "lock": threading.Lock() } def _rebuild_global_stats(): with _stats_lock: _global_stats["total_frames"] = 0 _global_stats["total_detections"] = 0 _global_stats["detections_by_class"] = defaultdict(int) _global_stats["scenes"] = [] for jid, job in _jobs.items(): if job.get("status") != "done" or not job.get("stats"): continue stats = job["stats"] counts = stats.get("unique_counts", {}) total = stats.get("total_unique", sum(counts.values())) _global_stats["total_frames"] += stats.get("total_frames", job.get("frames", 0)) _global_stats["total_detections"] += total for cls, cnt in counts.items(): _global_stats["detections_by_class"][cls] += cnt _global_stats["scenes"].append({ "scene_id": jid, "video_name": job.get("name", ""), "frames": stats.get("total_frames", job.get("frames", 0)), "total": total, "unique_counts": counts, "generated_at": stats.get("generated_at") or job.get("updated_at"), "duration_s": stats.get("duration_s", 0), "fps": stats.get("fps", 0), "timeline": stats.get("timeline", []) }) def _csv_text_for_job(job): output = io.StringIO() writer = csv.DictWriter(output, fieldnames=CSV_FIELDS, lineterminator="\n") writer.writeheader() for det in job.get("frame_detections", []): writer.writerow({ "frame": det["frame"], "timestamp_sec": f"{det['timestamp_sec']:.3f}", "scene_name": det["scene_name"], "group_id": LOG_GROUP_ID, "video_name": det["video_name"], "track_id": det["track_id"], "class_name": det["class_name"], "confidence": f"{det['confidence']:.3f}", "bbox_x1": det["bbox_x1"], "bbox_y1": det["bbox_y1"], "bbox_x2": det["bbox_x2"], "bbox_y2": det["bbox_y2"], "cx": det["cx"], "cy": det["cy"], "frame_width": det["frame_width"], "frame_height": det["frame_height"], "crossed_line": det["crossed_line"], "direction": _clean_direction(det.get("direction")), "speed_px_s": f"{det['speed_px_s']:.1f}" }) return output.getvalue() def _write_job_csv(jid, job): log_path = LOG_DIR / f"{jid}.csv" log_path.write_text(_csv_text_for_job(job), encoding="utf-8") job["log_path"] = str(log_path) return log_path _restore_jobs() _restore_data_csv_jobs() _rebuild_global_stats() # ── Routes HTML ─────────────────────────────────────────────────────────────── @app.route("/") def index(): return render_template("index.html") @app.route("/home") def home(): return render_template("home.html") @app.route("/dashboard") def dashboard(): return render_template("dashboard.html") @app.route("/logs") def logs(): return render_template("logs.html") @app.route("/static/") def serve_static(filename): return send_from_directory("static", filename) # ── API Routes ──────────────────────────────────────────────────────────────── @app.route("/health") def health(): return jsonify({"status": "ok", "model_ready": _model_ready}), 200 @app.route("/api/upload", methods=["POST"]) def api_upload(): jid = uuid.uuid4().hex[:10] f = request.files.get("video") source_url = (request.form.get("url") or "").strip() if not f and not source_url: return jsonify({"error": "No file or URL"}), 400 if f: source_name = f.filename dest = UPLOAD_DIR / f"{jid}_{f.filename}" source_path = str(dest) else: source_name = source_url.rsplit("/", 1)[-1] or source_url dest = source_url source_path = source_url print(f"\n{'='*60}") print(f"[UPLOAD] Job ID: {jid}") print(f"[UPLOAD] File: {source_name}") print(f"[UPLOAD] Destination: {dest}") print(f"[UPLOAD] Saving file to disk..." if f else "[UPLOAD] Registering stream URL...") print(f"{'='*60}") if f: f.save(str(dest)) file_size = dest.stat().st_size / (1024*1024) print(f"[UPLOAD] ✅ File saved successfully") print(f"[UPLOAD] File size: {file_size:.2f} MB") else: print(f"[UPLOAD] ✅ URL registered successfully") _jobs[jid] = { "status": "uploaded", "path": source_path, "name": source_name, "output_path": str(OUTPUT_DIR / f"{jid}_output.mp4"), "log_path": str(LOG_DIR / f"{jid}.csv"), "detections": defaultdict(int), "frames": 0, "processed_frames": 0, "progress": 0, "classes": DEFAULT_CLASSES, "latest_frame": None, "latest_frame_no": 0, "frame_detections": [], "timeline": [], "stats": {}, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "lock": threading.Lock() } _persist_jobs() print(f"[UPLOAD] Job created and ready for processing\n") return jsonify({"job_id": jid}) @app.route("/api/run", methods=["POST"]) def api_run(): data = request.json or {} jid = data.get("job_id") if not jid or jid not in _jobs: return jsonify({"error": "Unknown job_id"}), 404 _ensure_model_loading() if not _model_ready: return jsonify({"error": "Model not ready yet, please wait"}), 503 print("\n" + "="*60) print(f"[RUN] Starting analysis for job: {jid}") classes = data.get("classes", DEFAULT_CLASSES) _jobs[jid]["classes"] = classes _jobs[jid]["updated_at"] = datetime.now().isoformat() print(f"[RUN] Classes: {classes}") print("="*60) if _jobs[jid].get("status") == "processing": return jsonify({"status": "already_running"}) threading.Thread(target=_worker, args=(jid,), daemon=True).start() return jsonify({"status": "started"}) def _worker(jid): global _global_stats job = _jobs[jid] job["status"] = "processing" job["detections"] = defaultdict(int) job["frame_detections"] = [] job["progress"] = 0 job["stats"] = {} job["timeline"] = [] job["updated_at"] = datetime.now().isoformat() _persist_jobs() classes_filter = set(job.get("classes") or DEFAULT_CLASSES) counted_track_ids = set() print(f"\n[PROCESS] Opening video: {job['path']}") cap = cv2.VideoCapture(job["path"]) if not cap.isOpened(): print("[PROCESS] ERROR: Could not open video file") job["status"] = "error" job["error"] = "Could not open video file" return # Get video properties fps = cap.get(cv2.CAP_PROP_FPS) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) job["frames"] = total_frames print(f"[PROCESS] Video properties:") print(f" - FPS: {fps}") print(f" - Resolution: {width}x{height}") print(f" - Total frames: {total_frames}") print(f"[PROCESS] Initializing video writer...") # Initialize video writer for output fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(job["output_path"], fourcc, fps, (width, height)) if not out.isOpened(): print("[PROCESS] ERROR: Could not initialize video writer") cap.release() job["status"] = "error" job["error"] = "Could not initialize video writer" return print(f"[PROCESS] Output video: {job['output_path']}") print(f"[PROCESS] Starting frame processing...\n") frame_count = 0 frame_detections = [] # Store all detections for CSV export while True: ret, frame = cap.read() if not ret: break frame_count += 1 timestamp_sec = (frame_count - 1) / fps if fps > 0 else 0 # Run tracking so each physical object gets a stable track_id. results = _model.track( frame, conf=CONF, iou=IOU, imgsz=INFER_SZ, persist=True, verbose=False ) # Draw detections on frame annotated_frame = frame.copy() if results and results[0].boxes: for box_index, box in enumerate(results[0].boxes): cls = int(box.cls[0]) conf = float(box.conf[0]) class_name = COCO_TO_LABEL.get(cls) if class_name and class_name in classes_filter: # Get bounding box coordinates x1, y1, x2, y2 = map(int, box.xyxy[0]) cx = (x1 + x2) // 2 cy = (y1 + y2) // 2 raw_track_id = box.id[0] if box.id is not None else None track_id = str(int(raw_track_id)) if raw_track_id is not None else f"untracked-{class_name}-{box_index}" unique_key = (class_name, track_id) if unique_key not in counted_track_ids: counted_track_ids.add(unique_key) job["detections"][class_name] = job["detections"].get(class_name, 0) + 1 # Store detection data for CSV frame_detections.append({ "frame": frame_count, "timestamp_sec": timestamp_sec, "scene_name": jid, "group_id": LOG_GROUP_ID, "video_name": job["name"], "track_id": track_id, "class_name": class_name, "confidence": conf, "bbox_x1": x1, "bbox_y1": y1, "bbox_x2": x2, "bbox_y2": y2, "cx": cx, "cy": cy, "frame_width": width, "frame_height": height, "crossed_line": "false", "direction": "", "speed_px_s": 0.0 }) # Draw bounding box color = CLASS_COLORS.get(class_name, (255, 255, 255)) label = f"{class_name} {conf:.2f}" cv2.rectangle(annotated_frame, (x1, y1), (x2, y2), color, 2) cv2.putText(annotated_frame, label, (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2) # Write annotated frame to output video out.write(annotated_frame) ok, buffer = cv2.imencode(".jpg", annotated_frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) if ok: progress = (frame_count / total_frames * 100) if total_frames > 0 else 0 with job["lock"]: job["latest_frame"] = buffer.tobytes() job["latest_frame_no"] = frame_count job["processed_frames"] = frame_count job["progress"] = progress job["updated_at"] = datetime.now().isoformat() if frame_count == 1 or frame_count % 15 == 0: timeline_point = { "frame": frame_count, "ts": timestamp_sec, **dict(job["detections"]) } job["timeline"].append(timeline_point) # Log progress every 30 frames if frame_count % 30 == 0: progress = (frame_count / total_frames * 100) if total_frames > 0 else 0 detections = sum(job['detections'].values()) print(f"[PROCESS] Frame {frame_count}/{total_frames} ({progress:.1f}%) - Detections: {detections}") # Save frame detections for later CSV export job["frame_detections"] = frame_detections # Release resources cap.release() out.release() job["status"] = "done" job["frames"] = frame_count job["processed_frames"] = frame_count job["progress"] = 100 job["updated_at"] = datetime.now().isoformat() output_size = Path(job["output_path"]).stat().st_size / (1024*1024) duration_s = frame_count / fps if fps > 0 else 0 total_detections = sum(job['detections'].values()) stats = { "scene_id": jid, "video_name": job["name"], "total_frames": frame_count, "processed_frames": frame_count, "total_unique": total_detections, "total_detections": total_detections, "unique_counts": dict(job["detections"]), "duration_s": round(duration_s, 2), "fps": fps, "generated_at": datetime.now().isoformat(), "timeline": job.get("timeline", []), } job["stats"] = stats log_path = _write_job_csv(jid, job) _persist_jobs() print(f"\n[PROCESS] Processing complete!") print(f"[PROCESS] Total frames: {frame_count}") print(f"[PROCESS] Duration: {duration_s:.2f}s") print(f"[PROCESS] Total detections: {total_detections}") print(f"[PROCESS] Detections: {dict(job['detections'])}") print(f"[PROCESS] Output size: {output_size:.2f} MB") print(f"[PROCESS] CSV log: {log_path}") print("="*60 + "\n") with _stats_lock: _global_stats["total_frames"] += frame_count _global_stats["total_detections"] += total_detections for cls, cnt in job["detections"].items(): _global_stats["detections_by_class"][cls] += cnt _global_stats["scenes"].append({ "scene_id": jid, "video_name": job["name"], "frames": frame_count, "total": total_detections, "unique_counts": dict(job["detections"]), "generated_at": datetime.now().isoformat(), "duration_s": duration_s, "fps": fps, "timeline": job.get("timeline", []) }) _persist_jobs() @app.route("/api/status/") def api_status(jid): job = _jobs.get(jid) if not job: return jsonify({"error": "not found"}), 404 return jsonify({ "job_id": jid, "status": job["status"], "name": job.get("name", ""), "progress": job.get("progress", 0), "processed_frames": job.get("processed_frames", 0), "frames": job.get("frames", 0), "stats": job.get("stats", {}), "video_url": f"/api/video/{jid}" if Path(job.get("output_path", "")).exists() else None, "csv_url": f"/api/logs/{jid}/csv" if Path(job.get("log_path", "")).exists() else None }) # ── Dashboard ───────────────────────────────────────────────────────────────── @app.route("/api/dashboard/stats") def api_dashboard_stats(): with _stats_lock: live_jobs = [] active_counts = defaultdict(int) active_frames = 0 active_processed = 0 for jid, job in _jobs.items(): if job.get("status") not in ["uploaded", "processing", "running"]: continue counts = dict(job.get("detections", {})) for cls, cnt in counts.items(): active_counts[cls] += cnt active_frames += job.get("frames", 0) or 0 active_processed += job.get("processed_frames", 0) or 0 live_jobs.append({ "scene_id": jid, "video_name": job.get("name", ""), "status": job.get("status"), "frames": job.get("frames", 0), "processed_frames": job.get("processed_frames", 0), "progress": job.get("progress", 0), "total": sum(counts.values()), "unique_counts": counts, "fps": job.get("stats", {}).get("fps", 0), "duration_s": 0, "timeline": job.get("timeline", []), }) combined_counts = defaultdict(int, _global_stats["detections_by_class"]) for cls, cnt in active_counts.items(): combined_counts[cls] += cnt return jsonify({ "global_unique_counts": dict(combined_counts), "completed_unique_counts": dict(_global_stats["detections_by_class"]), "active_unique_counts": dict(active_counts), "scenes": _global_stats["scenes"], "active_jobs": live_jobs, "total_frames": _global_stats["total_frames"], "active_frames": active_frames, "active_processed_frames": active_processed, "total_detections": _global_stats["total_detections"] + sum(active_counts.values()) }), 200 # ── Logs ────────────────────────────────────────────────────────────────────── @app.route("/api/logs") def api_logs(): with _stats_lock: return jsonify(_global_stats["scenes"]), 200 @app.route("/api/logs/rows") def api_logs_rows(): rows = [] for jid, job in _jobs.items(): for det in job.get("frame_detections", []): rows.append({ "frame": det["frame"], "frame_id": det["frame"], "timestamp_sec": det["timestamp_sec"], "timestamp_s": det["timestamp_sec"], "scene_name": det["scene_name"], "scene_id": jid, "group_id": LOG_GROUP_ID, "video_name": det["video_name"], "track_id": det["track_id"], "class_name": det["class_name"], "confidence": det["confidence"], "bbox_x1": det["bbox_x1"], "bbox_y1": det["bbox_y1"], "bbox_x2": det["bbox_x2"], "bbox_y2": det["bbox_y2"], "x1": det["bbox_x1"], "y1": det["bbox_y1"], "x2": det["bbox_x2"], "y2": det["bbox_y2"], "cx": det["cx"], "cy": det["cy"], "frame_width": det["frame_width"], "frame_height": det["frame_height"], "crossed_line": det["crossed_line"], "direction": _clean_direction(det.get("direction")), "speed_px_s": det["speed_px_s"] }) rows.sort(key=lambda r: (r["scene_id"], r["frame"], str(r["track_id"]))) return jsonify(rows), 200 @app.route("/api/logs/clear", methods=["DELETE"]) def api_logs_clear(): with _stats_lock: _global_stats["scenes"].clear() _global_stats["total_frames"] = 0 _global_stats["total_detections"] = 0 _global_stats["detections_by_class"].clear() keys = [k for k in _jobs if _jobs[k].get("status") == "done"] for k in keys: del _jobs[k] _persist_jobs() return jsonify({"status": "cleared"}), 200 @app.route("/api/logs//csv") def api_logs_csv(scene_id): job = _jobs.get(scene_id) if not job: return jsonify({"error": "not found"}), 404 log_path = Path(job.get("log_path", "")) if not job.get("frame_detections") and log_path.exists(): return send_file( str(log_path), mimetype="text/csv", as_attachment=True, download_name=f"{scene_id}_logs.csv" ) output = _csv_text_for_job(job) return Response( output, mimetype="text/csv", headers={"Content-Disposition": f"attachment;filename={scene_id}_logs.csv"} ), 200 # ── Webcam ──────────────────────────────────────────────────────────────────── @app.route("/api/webcam/start", methods=["POST"]) def api_webcam_start(): global _webcam_active, _webcam_classes, _webcam_frame_count global _webcam_detections, _webcam_detections_by_class global _webcam_current_counts, _webcam_seen_track_ids data = request.json or {} classes = data.get("classes", DEFAULT_CLASSES) _ensure_model_loading() if not _model_ready: return jsonify({"error": "Model not ready"}), 503 with _webcam_lock: _webcam_active = True _webcam_classes = classes _webcam_frame_count = 0 _webcam_detections = 0 _webcam_detections_by_class = defaultdict(int) _webcam_current_counts = defaultdict(int) _webcam_seen_track_ids = set() print(f" Webcam started with classes: {classes}") return jsonify({"status": "started"}), 200 @app.route("/api/webcam/frame", methods=["POST"]) def api_webcam_frame(): global _webcam_frame_count, _webcam_detections, _webcam_detections_by_class global _webcam_current_counts, _webcam_seen_track_ids if not _webcam_active or not _model_ready: return jsonify({"error": "Webcam not active"}), 503 try: img_data = request.data if not img_data: return jsonify({"error": "No image data"}), 400 nparr = np.frombuffer(img_data, np.uint8) frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR) if frame is None: return jsonify({"error": "Could not decode image"}), 400 # Tracking YOLO: count unique objects by persistent track_id, not by frame. results = _model.track( frame, conf=CONF, iou=IOU, imgsz=INFER_SZ, persist=True, verbose=False ) annotated, _ = _draw_detections(frame, results, _webcam_classes) current_counts = defaultdict(int) new_unique_counts = defaultdict(int) with _webcam_lock: _webcam_frame_count += 1 if results and results[0].boxes: for box in results[0].boxes: cls = int(box.cls[0]) class_name = COCO_TO_LABEL.get(cls) if class_name and class_name in _webcam_classes: current_counts[class_name] += 1 if box.id is None: continue track_id = int(box.id[0]) unique_key = (class_name, track_id) if unique_key not in _webcam_seen_track_ids: _webcam_seen_track_ids.add(unique_key) new_unique_counts[class_name] += 1 for class_name, count in new_unique_counts.items(): _webcam_detections_by_class[class_name] += count _webcam_current_counts = current_counts _webcam_detections = sum(_webcam_detections_by_class.values()) ret, buffer = cv2.imencode('.jpg', annotated) return Response(buffer.tobytes(), mimetype='image/jpeg'), 200 except Exception as e: print(f"❌ Webcam frame error: {e}") return jsonify({"error": str(e)}), 500 @app.route("/api/webcam/stop", methods=["POST"]) def api_webcam_stop(): global _webcam_active with _webcam_lock: _webcam_active = False if _webcam_frame_count > 0 and _webcam_detections_by_class: jid = "webcam_" + uuid.uuid4().hex[:8] unique_counts = dict(_webcam_detections_by_class) total = sum(unique_counts.values()) now = datetime.now().isoformat() _jobs[jid] = { "status": "done", "path": "webcam", "name": f"Webcam_{now[:10]}", "output_path": "", "log_path": "", "detections": defaultdict(int, unique_counts), "frames": _webcam_frame_count, "processed_frames": _webcam_frame_count, "progress": 100, "classes": list(_webcam_classes), "latest_frame": None, "latest_frame_no": 0, "frame_detections": [], "timeline": [], "stats": { "scene_id": jid, "video_name": f"Webcam_{now[:10]}", "total_frames": _webcam_frame_count, "processed_frames": _webcam_frame_count, "total_unique": total, "total_detections": total, "unique_counts": unique_counts, "duration_s": 0, "fps": 0, "generated_at": now, "timeline": [], }, "created_at": now, "updated_at": now, "lock": threading.Lock() } with _stats_lock: _global_stats["total_detections"] += total for cls, cnt in unique_counts.items(): _global_stats["detections_by_class"][cls] += cnt _global_stats["scenes"].append({ "scene_id": jid, "video_name": f"Webcam_{now[:10]}", "frames": _webcam_frame_count, "total": total, "unique_counts": unique_counts, "generated_at": now, "duration_s": 0, "fps": 0, "timeline": [] }) print("⏹️ Webcam stopped") return jsonify({"status": "stopped"}), 200 @app.route("/api/webcam/stats") def api_webcam_stats(): with _webcam_lock: # unique_counts = objets uniques vus depuis le début unique_only = dict(_webcam_detections_by_class) total_unique = sum(unique_only.values()) # frame_counts = ce qui est visible sur la frame courante seulement frame_counts = dict(_webcam_current_counts) return jsonify({ "active": _webcam_active, "frame": _webcam_frame_count, # detections = total uniques, pas cumul frames "detections": total_unique, "detections_by_class": unique_only, "frame_counts": frame_counts, "unique_counts": unique_only, }), 200 # ── Video streaming endpoints ───────────────────────────────────────────────── @app.route("/api/mjpeg/") def api_mjpeg(jid): """Stream MJPEG frames while processing""" job = _jobs.get(jid) if not job: return jsonify({"error": "not found"}), 404 def generate(): last_frame_no = -1 while job.get("status") in ["uploaded", "running", "processing", "done"]: try: with job["lock"]: frame = job.get("latest_frame") frame_no = job.get("latest_frame_no", 0) if frame and frame_no != last_frame_no: last_frame_no = frame_no yield ( b"--boundary\r\n" b"Content-Type: image/jpeg\r\n" + f"Content-Length: {len(frame)}\r\n\r\n".encode("ascii") + frame + b"\r\n" ) elif job.get("status") == "done": break time.sleep(0.05) except GeneratorExit: break return Response( stream_with_context(generate()), mimetype='multipart/x-mixed-replace; boundary=boundary' ) @app.route("/api/stream/") def api_stream(jid): """Server-Sent Events for progress updates""" job = _jobs.get(jid) if not job: return jsonify({"error": "not found"}), 404 def generate(): while True: status = job.get("status") if status == "done": yield f"data: {json.dumps({'event': 'done', 'stats': job.get('stats', {})})}\n\n" break if status == "error": yield f"data: {json.dumps({'event': 'error', 'msg': job.get('error', 'processing failed')})}\n\n" break counts = dict(job.get("detections", {})) processed = job.get("processed_frames", 0) total = job.get("frames") or 0 frame_total = total if total > 0 else None payload = { "event": "progress", "status": status, "pct": job.get("progress", 0), "processed_frames": processed, "total_frames": frame_total, "unique_counts": counts, "no_objects": not bool(counts), "hud": { "frame_str": f"F:{processed}" + (f"/{frame_total}" if frame_total else ""), "counts": counts } } yield f"data: {json.dumps(payload)}\n\n" time.sleep(0.5) return Response( stream_with_context(generate()), mimetype='text/event-stream', headers={'Cache-Control': 'no-cache'} ) @app.route("/api/video/") def api_video(jid): """Serve the processed video file with annotations""" job = _jobs.get(jid) if not job: return jsonify({"error": "not found"}), 404 video_path = Path(job.get("output_path")) if not video_path.exists(): return jsonify({"error": "video not found"}), 404 print(f"[API] Serving video: {video_path}") return send_file(str(video_path), mimetype='video/mp4') # ── Main ────────────────────────────────────────────────────────────────────── if __name__ == "__main__": import os port = int(os.environ.get("PORT", 7860)) app.run(host="0.0.0.0", port=port, debug=False)