#!/usr/bin/env python3 """ Run edge detection on test video frames — visual verification. Uses a minimal 3-stage pipeline: extract_frames → filter_scenes → detect_edges. No YOLO, OCR, or downstream stages. Usage: python tests/detect/manual/run_region_analysis.py [--job JOB_ID] [--port PORT] [--local] Opens: http://mpr.local.ar/detection/?job= What to look for in the frame viewer: - "Edges" toggle appears (cyan) - Cyan boxes around horizontal line pairs (hoarding edges) - No boxes on players, ball, or sky - Boxes concentrated in the lower third of the frame """ import argparse import logging import os import sys import time as _time parser = argparse.ArgumentParser() parser.add_argument("--job", default=f"cv-{int(_time.time()) % 100000}") parser.add_argument("--port", type=int, default=6379) parser.add_argument("--local", action="store_true", help="Run CV locally (no inference server)") args = parser.parse_args() os.environ["REDIS_URL"] = f"redis://localhost:{args.port}/0" if args.local: os.environ.pop("INFERENCE_URL", None) logging.basicConfig(level=logging.DEBUG, format="%(levelname)-7s %(name)s — %(message)s") sys.path.insert(0, ".") from langgraph.graph import END, StateGraph from detect import emit from detect.models import PipelineStats from detect.profiles.soccer import SoccerBroadcastProfile from detect.stages.frame_extractor import extract_frames from detect.stages.scene_filter import scene_filter from detect.stages.edge_detector import detect_edge_regions from detect.state import DetectState logger = logging.getLogger(__name__) VIDEO = "media/mpr/out/chunks/95043d50-4df6-4ac8-bbd5-2ba873117c6e/chunk_0000.mp4" INFERENCE_URL = os.environ.get("INFERENCE_URL") # --- 3-stage pipeline --- NODES = ["extract_frames", "filter_scenes", "detect_edges"] def _emit_transition(job_id: str, node: str, status: str, node_states: dict): node_states[node] = status nodes = [{"id": n, "status": node_states.get(n, "pending")} for n in NODES] emit.graph_update(job_id, nodes) def node_extract(state: DetectState) -> dict: job_id = state.get("job_id", "") ns = state.get("_node_states", {n: "pending" for n in NODES}) _emit_transition(job_id, "extract_frames", "running", ns) profile = SoccerBroadcastProfile() config = profile.frame_extraction_config() frames = extract_frames(state["video_path"], config, job_id=job_id) _emit_transition(job_id, "extract_frames", "done", ns) return {"frames": frames, "stats": PipelineStats(frames_extracted=len(frames)), "_node_states": ns} def node_filter(state: DetectState) -> dict: job_id = state.get("job_id", "") ns = state.get("_node_states", {}) _emit_transition(job_id, "filter_scenes", "running", ns) profile = SoccerBroadcastProfile() config = profile.scene_filter_config() kept = scene_filter(state.get("frames", []), config, job_id=job_id) stats = state.get("stats", PipelineStats()) stats.frames_after_scene_filter = len(kept) _emit_transition(job_id, "filter_scenes", "done", ns) return {"filtered_frames": kept, "stats": stats, "_node_states": ns} def node_edges(state: DetectState) -> dict: job_id = state.get("job_id", "") ns = state.get("_node_states", {}) _emit_transition(job_id, "detect_edges", "running", ns) profile = SoccerBroadcastProfile() config = profile.region_analysis_config() regions = detect_edge_regions( state.get("filtered_frames", []), config, inference_url=INFERENCE_URL, job_id=job_id, ) total = sum(len(r) for r in regions.values()) stats = state.get("stats", PipelineStats()) stats.cv_regions_detected = total _emit_transition(job_id, "detect_edges", "done", ns) return {"edge_regions_by_frame": regions, "stats": stats, "_node_states": ns} def build_3stage_graph() -> StateGraph: graph = StateGraph(DetectState) graph.add_node("extract_frames", node_extract) graph.add_node("filter_scenes", node_filter) graph.add_node("detect_edges", node_edges) graph.set_entry_point("extract_frames") graph.add_edge("extract_frames", "filter_scenes") graph.add_edge("filter_scenes", "detect_edges") graph.add_edge("detect_edges", END) return graph def main(): logger.info("Job: %s", args.job) logger.info("Mode: %s", "remote" if INFERENCE_URL else "local") logger.info("Pipeline: extract_frames → filter_scenes → detect_edges") logger.info("Open: http://mpr.local.ar/detection/?job=%s", args.job) input("\nPress Enter to start...") emit.set_run_context(run_id=args.job, parent_job_id=args.job, run_type="initial", log_level="DEBUG") graph = build_3stage_graph() pipeline = graph.compile() initial_state = { "video_path": VIDEO, "job_id": args.job, "profile_name": "soccer_broadcast", } result = pipeline.invoke(initial_state) # Print results regions = result.get("edge_regions_by_frame", {}) total = sum(len(boxes) for boxes in regions.values()) frames_with_regions = sum(1 for boxes in regions.values() if boxes) logger.info("Results:") logger.info(" Total edge regions: %d", total) logger.info(" Frames with regions: %d / %d", frames_with_regions, len(result.get("filtered_frames", []))) for seq, boxes in sorted(regions.items()): if boxes: labels = [f"{b.label}({b.confidence:.2f})" for b in boxes] logger.info(" Frame %d: %s", seq, ", ".join(labels)) logger.info("Done. Check the frame viewer for cyan boxes.") logger.info("") # --- Parameter sensitivity --- logger.info("=== Parameter sensitivity (local debug) ===") from detect.stages.edge_detector import _load_cv_edges edges_mod = _load_cv_edges() filtered = result.get("filtered_frames", []) if filtered: sample = filtered[0] for canny_low in [20, 50, 80, 120]: dbg = edges_mod.detect_edges_debug(sample.image, canny_low=canny_low) logger.info( " canny_low=%d → %d horizontals, %d pairs, %d regions", canny_low, dbg["horizontal_count"], dbg["pair_count"], len(dbg["regions"]), ) logger.info("") logger.info("=== Editor test ===") logger.info(" Dashboard: http://mpr.local.ar/detection/?job=%s", args.job) logger.info(" Editor: http://mpr.local.ar/detection/?job=%s#/editor/detect_edges", args.job) if __name__ == "__main__": main()