"""Command-line entry point for pedestrian detection and tracking.""" from __future__ import annotations import argparse import logging import sys from dataclasses import replace from pathlib import Path from typing import Any def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="YOLO + tracking + head-pose + turn + voice-prompt research system", epilog="Desktop GUI: python main.py gui | Offline analysis: python main.py analyze", ) parser.add_argument("--config", type=Path, default=Path("config/experiment.yaml")) parser.add_argument("--input-type", choices=("camera", "video")) parser.add_argument("--camera-id", type=int) parser.add_argument("--video", type=str) parser.add_argument("--model", type=str) parser.add_argument("--device", type=str) parser.add_argument("--confidence", type=float) parser.add_argument("--iou", type=float) parser.add_argument("--image-size", type=int) parser.add_argument("--save-video", action="store_true", default=None) parser.add_argument("--no-display", action="store_true", default=None) parser.add_argument("--debug", action="store_true", default=None) parser.add_argument("--tracker", choices=("bytetrack",)) parser.add_argument("--disable-tracker", action="store_true", default=None) parser.add_argument("--track-buffer", type=int) parser.add_argument("--match-threshold", type=float) trajectory_group = parser.add_mutually_exclusive_group() trajectory_group.add_argument("--show-trajectories", action="store_true", default=None) trajectory_group.add_argument("--no-trajectories", action="store_true", default=None) parser.add_argument("--disable-face", action="store_true", default=None) parser.add_argument("--face-every", type=int) parser.add_argument("--max-face-persons", type=int) parser.add_argument("--no-head-pose", action="store_true", default=None) parser.add_argument("--show-face-landmarks", action="store_true", default=None) parser.add_argument("--no-face-roi", action="store_true", default=None) parser.add_argument("--disable-turn", action="store_true", default=None) parser.add_argument("--signage-yaw-direction", choices=("positive", "negative")) parser.add_argument("--turn-weak-threshold", type=float) parser.add_argument("--turn-medium-threshold", type=float) parser.add_argument("--turn-strong-threshold", type=float) parser.add_argument("--turn-min-duration", type=float) parser.add_argument("--no-turn-frame-log", action="store_true", default=None) parser.add_argument("--disable-voice", action="store_true", default=None) parser.add_argument("--voice-mode", choices=("prompt", "control", "disabled")) parser.add_argument("--audio-file", type=str) parser.add_argument("--voice-trigger-strategy", choices=("trigger_line_crossing", "region_entry", "fixed_x_position", "first_confirmed_track")) parser.add_argument("--voice-response-window", type=float) parser.add_argument("--voice-cooldown", type=float) parser.add_argument("--voice-volume", type=float) parser.add_argument("--no-voice-decision-log", action="store_true", default=None) parser.add_argument("--disable-person-state", action="store_true", default=None) parser.add_argument("--no-person-state-frame-log", action="store_true", default=None) parser.add_argument("--person-state-min-duration", type=float) parser.add_argument("--disable-unified-logging", action="store_true", default=None) parser.add_argument("--no-unified-events", action="store_true", default=None) parser.add_argument("--no-analysis-table", action="store_true", default=None) parser.add_argument("--no-session-summary", action="store_true", default=None) parser.add_argument("--camera-position-note", type=str) return parser def build_analysis_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="main.py analyze", description="Offline prompt/control statistical analysis") parser.add_argument("--config", type=Path, default=Path("config/experiment.yaml")) parser.add_argument("--input", type=Path, default=Path("data/logs/unified")) parser.add_argument("--output", type=Path, default=Path("data/analysis")) parser.add_argument("--no-analysis-plots", action="store_true") parser.add_argument("--analysis-recursive", action="store_true", default=None) parser.add_argument("--analysis-response-test", choices=("auto", "chi_square", "fisher")) parser.add_argument("--analysis-numeric-test", choices=("mannwhitney", "ttest")) return parser def build_gui_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="main.py gui", description="Phase 12 desktop experiment and final-results GUI") parser.add_argument("--config", type=Path, default=Path("config/experiment.yaml")) return parser def build_operation_parser(command:str)->argparse.ArgumentParser: descriptions={"protocol":"Generate the Phase 10 experiment protocol","quality-check":"Check Phase 7/8 log quality","pilot-report":"Generate a pilot experiment report","presentation-summary":"Generate a midterm presentation summary"} parser=argparse.ArgumentParser(prog=f"main.py {command}",description=descriptions[command]);parser.add_argument("--config",type=Path,default=Path("config/experiment.yaml")) if command=="protocol":parser.add_argument("--output",type=Path,default=Path("docs/experiment_protocol.md"));parser.add_argument("--yaml-output",type=Path,default=Path("data/experiment_notes/experiment_protocol.yaml")) elif command=="quality-check":parser.add_argument("--logs",type=Path,default=Path("data/logs/unified"));parser.add_argument("--analysis",type=Path,default=Path("data/analysis"));parser.add_argument("--output",type=Path,default=Path("data/analysis")) else:parser.add_argument("--analysis",type=Path,default=Path("data/analysis"));parser.add_argument("--logs",type=Path,default=Path("data/logs/unified"));parser.add_argument("--output",type=Path,required=False) return parser def build_phase11_parser(command:str)->argparse.ArgumentParser: p=argparse.ArgumentParser(prog=f"main.py {command}",description="Phase 11 pilot diagnostics and calibration history");p.add_argument("--config",type=Path,default=Path("config/experiment.yaml")) if command=="pilot-diagnostics":p.add_argument("--analysis",type=Path,default=Path("data/analysis"));p.add_argument("--logs",type=Path,default=Path("data/logs/unified"));p.add_argument("--output",type=Path,default=Path("data/analysis")) elif command=="calibration-add":p.add_argument("--path",type=Path);p.add_argument("--operator",default="unknown");p.add_argument("--category",required=True);p.add_argument("--before",required=True);p.add_argument("--after",required=True);p.add_argument("--reason",required=True);p.add_argument("--report-id");p.add_argument("--notes",default="") else:p.add_argument("--path",type=Path) return p def build_phase12_parser(command:str)->argparse.ArgumentParser: p=argparse.ArgumentParser(prog=f"main.py {command}",description="Phase 12 final experiment workflow");p.add_argument("--config",type=Path,default=Path("config/experiment.yaml"));p.add_argument("--output",type=Path,default=Path("data/final_results")) if command in {"main-progress","final-analysis"}:p.add_argument("--logs",type=Path,default=Path("data/logs/unified"));p.add_argument("--analysis",type=Path,default=Path("data/analysis")) elif command in {"final-report","presentation-assets"}:p.add_argument("--analysis",type=Path,default=Path("data/analysis")) return p def cli_overrides(args: argparse.Namespace) -> dict[str, Any]: sections: dict[str, dict[str, Any]] = { "input": {}, "detector": {}, "tracker": {}, "face": {}, "head_pose": {}, "turn_detection": {}, "turn_logging": {}, "audio": {}, "voice_prompt": {}, "voice_logging": {}, "output": {}, "display": {}, "debug": {}, "person_state": {}, "person_state_logging": {}, "unified_logging": {}, } mapping = { "input_type": ("input", "type"), "camera_id": ("input", "camera_id"), "video": ("input", "video_path"), "model": ("detector", "model_path"), "device": ("detector", "device"), "confidence": ("detector", "confidence_threshold"), "iou": ("detector", "iou_threshold"), "image_size": ("detector", "image_size"), "save_video": ("output", "save_video"), "debug": ("debug", "enabled"), "tracker": ("tracker", "type"), "track_buffer": ("tracker", "track_buffer_frames"), "match_threshold": ("tracker", "match_thresh"), "face_every": ("face", "process_every_n_frames"), "max_face_persons": ("face", "max_persons_per_frame"), "signage_yaw_direction": ("turn_detection", "signage_yaw_direction"), "turn_weak_threshold": ("turn_detection", "weak_threshold_deg"), "turn_medium_threshold": ("turn_detection", "medium_threshold_deg"), "turn_strong_threshold": ("turn_detection", "strong_threshold_deg"), "turn_min_duration": ("turn_detection", "min_turn_duration_sec"), "voice_mode": ("voice_prompt", "mode"), "audio_file": ("audio", "file_path"), "voice_trigger_strategy": ("voice_prompt", "trigger_strategy"), "voice_response_window": ("voice_prompt", "response_window_sec"), "voice_cooldown": ("voice_prompt", "global_cooldown_sec"), "voice_volume": ("audio", "volume"), "person_state_min_duration": ("person_state", "min_track_duration_sec"), "camera_position_note": ("unified_logging", "camera_position_note"), } for argument, (section, key) in mapping.items(): value = getattr(args, argument) if value is not None: sections[section][key] = value if args.no_display is not None: sections["display"]["enabled"] = not args.no_display if args.disable_tracker is not None: sections["tracker"]["enabled"] = not args.disable_tracker if args.show_trajectories is not None: sections["display"]["show_trajectories"] = True if args.no_trajectories is not None: sections["display"]["show_trajectories"] = False if args.disable_face is not None: sections["face"]["enabled"] = False if args.no_head_pose is not None: sections["head_pose"]["enabled"] = False if args.show_face_landmarks is not None: sections["display"]["show_face_landmarks"] = True if args.no_face_roi is not None: sections["display"]["show_face_roi"] = False if args.disable_turn is not None: sections["turn_detection"]["enabled"] = False if args.no_turn_frame_log is not None: sections["turn_logging"]["log_frame_turns"] = False if args.disable_voice is not None: sections["voice_prompt"].update({"enabled": False, "mode": "disabled"}) if args.no_voice_decision_log is not None: sections["voice_logging"]["log_decisions"] = False if args.disable_person_state is not None: sections["person_state"]["enabled"] = False if args.no_person_state_frame_log is not None: sections["person_state_logging"]["log_frame_states"] = False if args.disable_unified_logging is not None: sections["unified_logging"]["enabled"] = False if args.no_unified_events is not None: sections["unified_logging"]["log_unified_events"] = False if args.no_analysis_table is not None: sections["unified_logging"]["log_person_analysis_table"] = False if args.no_session_summary is not None: sections["unified_logging"]["log_session_summary"] = False return {key: value for key, value in sections.items() if value} def main(argv: list[str] | None = None) -> int: actual = list(sys.argv[1:] if argv is None else argv) if actual and actual[0] in {"main-progress","snapshot","final-analysis","final-report","presentation-assets"}: command=actual[0];args=build_phase12_parser(command).parse_args(actual[1:]);from src.config_loader import load_config;from src.phase12_cli import assets,final_report,progress,snapshot,workflow try: c=load_config("config/default.yaml",args.config) if command=="main-progress":outputs=progress(c,args.logs,args.analysis,args.output) elif command=="snapshot":outputs=snapshot(c,args.config,args.output) elif command=="final-report":outputs=final_report(c,args.analysis,args.output) elif command=="presentation-assets":outputs=assets(c,args.analysis,args.output) else:r=workflow(c,args.logs,args.analysis,args.output);outputs=[Path(x) for x in r.output_files]+[args.output/"final_workflow_result.json"] [print(x) for x in outputs];return 0 except Exception as exc:print(f"{command} error: {exc}",file=sys.stderr);return 1 if actual and actual[0] in {"pilot-diagnostics","calibration-add","calibration-list"}: from datetime import datetime command=actual[0];args=build_phase11_parser(command).parse_args(actual[1:]);from src.config_loader import load_config;from src.models import CalibrationChangeRecord;from src.phase11_cli import add_calibration,list_calibrations,run_diagnostics try: c=load_config("config/default.yaml",args.config) if command=="pilot-diagnostics":outputs=run_diagnostics(c,args.analysis,args.logs,args.output);[print(x) for x in outputs] elif command=="calibration-add": record=CalibrationChangeRecord(f"change_{datetime.now():%Y%m%d_%H%M%S_%f}",datetime.now().astimezone().isoformat(timespec="seconds"),args.operator,args.category,args.before,args.after,args.reason,args.report_id,args.notes);print(add_calibration(c,record,args.path)) else: for r in list_calibrations(c,args.path):print(f"{r.created_at}\t{r.category}\t{r.before_value} -> {r.after_value}\t{r.reason}") return 0 except Exception as exc:print(f"{command} error: {exc}",file=sys.stderr);return 1 if actual and actual[0] in {"protocol","quality-check","pilot-report","presentation-summary"}: command=actual[0];args=build_operation_parser(command).parse_args(actual[1:]) from src.config_loader import load_config from src.phase10_cli import generate_protocol,run_pilot_report,run_presentation,run_quality try: config=load_config(Path("config/default.yaml"),args.config) if command=="protocol":outputs=generate_protocol(config,args.output,args.yaml_output) elif command=="quality-check":outputs=run_quality(config,args.logs,args.analysis,args.output) elif command=="pilot-report":outputs=[run_pilot_report(config,args.analysis,args.output or Path(config.pilot_report.output_markdown),args.logs)] else:outputs=[run_presentation(config,args.analysis,args.output or Path(config.presentation_summary.output_markdown),args.logs)] for output in outputs:print(output) return 0 except Exception as exc:print(f"{command} error: {exc}",file=sys.stderr);return 1 if actual and actual[0] == "gui": args = build_gui_parser().parse_args(actual[1:]) try: from src.gui.gui_app import run_gui return run_gui(args.config) except (RuntimeError, ImportError) as exc: print(f"GUI error: {exc}", file=sys.stderr) return 1 if actual and actual[0] == "analyze": logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s") args = build_analysis_parser().parse_args(actual[1:]) from src.analysis_cli import run_analysis_cli from src.config_loader import load_config from src.exceptions import ConfigurationError try: config = load_config(Path("config/default.yaml"), args.config) settings = replace(config.analysis, input_directory=str(args.input), output_directory=str(args.output), generate_plots=False if args.no_analysis_plots else config.analysis.generate_plots, recursive=args.analysis_recursive if args.analysis_recursive is not None else config.analysis.recursive, response_rate_test=args.analysis_response_test or config.analysis.response_rate_test, numeric_test=args.analysis_numeric_test or config.analysis.numeric_test) run_analysis_cli(settings, args.input, args.output) return 0 except (ConfigurationError, FileNotFoundError, RuntimeError) as exc: print(f"Analysis error: {exc}", file=sys.stderr) return 1 args = build_parser().parse_args(actual) from src.application import Application from src.config_loader import load_config from src.exceptions import CameraOpenError, ConfigurationError, FrameReadError, ModelLoadError try: config = load_config(Path("config/default.yaml"), args.config, cli_overrides(args)) Application(config).run() return 0 except ConfigurationError as exc: print(f"Configuration error: {exc}", file=sys.stderr) return 1 except (CameraOpenError, FrameReadError) as exc: print(f"Input error: {exc}", file=sys.stderr) return 2 except ModelLoadError as exc: print(f"Model error: {exc}", file=sys.stderr) return 3 except Exception as exc: print(f"Unexpected error: {exc}", file=sys.stderr) return 10 if __name__ == "__main__": raise SystemExit(main())