You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
263 lines
17 KiB
263 lines
17 KiB
"""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())
|
|
|