detect_video.py

"""Process a video file and annotate detected faces frame by frame."""

from __future__ import annotations

import argparse
import os
import sys

from face_scan.media import write_json
from face_scan.observability import AuditLogger, DEFAULT_CASCADE_SHA256, configure_logger, sha256_file
from face_scan.workflows import run_video_detection, validate_detector


DEFAULT_CASCADE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "haarcascade_frontalface_default.xml")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run face detection over a video file.")
    parser.add_argument("video", help="Path to the input video file.")
    parser.add_argument("-o", "--output", default=None, help="Optional path for an annotated MP4 output.")
    parser.add_argument("--summary-json", default=None, help="Optional path for a JSON run summary.")
    parser.add_argument("--snapshot-dir", default=None, help="Directory to save sampled snapshot frames with faces.")
    parser.add_argument("--snapshot-interval", type=float, default=5.0, help="Minimum seconds between saved snapshots.")
    parser.add_argument("--sample-every", type=int, default=1, help="Run detection every Nth frame.")
    parser.add_argument("--max-frames", type=int, default=0, help="Optional cap on frames to read (0 means full video).")
    parser.add_argument("--cascade", default=DEFAULT_CASCADE_PATH, help="Path to the Haar cascade XML file.")
    parser.add_argument(
        "--cascade-sha256",
        default=os.getenv("FACE_SCAN_CASCADE_SHA256") or None,
        help="Expected sha256 for the cascade XML.",
    )
    parser.add_argument("--skip-cascade-check", action="store_true", help="Skip cascade integrity check.")
    parser.add_argument("--scale-factor", type=float, default=1.1, help="Scale factor between pyramid steps.")
    parser.add_argument("--min-neighbors", type=int, default=5, help="Minimum neighbors needed for a detection.")
    parser.add_argument("--min-size", type=int, nargs=2, default=[60, 60], metavar=("MIN_WIDTH", "MIN_HEIGHT"))
    parser.add_argument("--draw-labels", action="store_true", help="Label each detected face.")
    parser.add_argument("--show-metrics", action="store_true", help="Overlay metrics on output frames.")
    parser.add_argument("--no-display", action="store_true", help="Skip showing the annotated playback window.")
    parser.add_argument(
        "--privacy",
        choices=("none", "blur", "pixelate", "black"),
        default=os.getenv("FACE_SCAN_PRIVACY", "none"),
        help="Redact detected faces for privacy.",
    )
    parser.add_argument(
        "--log-level",
        choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"),
        default=os.getenv("FACE_SCAN_LOG_LEVEL", "INFO"),
        help="Log level.",
    )
    parser.add_argument("--log-file", default=os.getenv("FACE_SCAN_LOG_FILE") or None, help="Optional log file path.")
    parser.add_argument(
        "--log-format",
        choices=("text", "json"),
        default=os.getenv("FACE_SCAN_LOG_FORMAT", "text"),
        help="Log output format.",
    )
    parser.add_argument("--audit-log", default=os.getenv("FACE_SCAN_AUDIT_LOG") or None, help="Optional audit log path.")
    return parser.parse_args()


def verify_cascade(args: argparse.Namespace, logger, audit: AuditLogger | None) -> bool:
    if not os.path.exists(args.cascade):
        logger.error("Cascade XML is missing: %s", args.cascade)
        if audit:
            audit.emit("video_error", reason="missing_cascade")
        return False

    if args.skip_cascade_check:
        return True

    expected = args.cascade_sha256
    if expected is None and os.path.abspath(args.cascade) == DEFAULT_CASCADE_PATH:
        expected = DEFAULT_CASCADE_SHA256
    if expected:
        actual = sha256_file(args.cascade)
        if actual.lower() != expected.lower():
            logger.error("Cascade integrity check failed for %s", args.cascade)
            logger.error("Expected sha256=%s got=%s", expected, actual)
            if audit:
                audit.emit(
                    "video_error",
                    reason="cascade_hash_mismatch",
                    expected_sha256=expected,
                    actual_sha256=actual,
                )
            return False
    return True


def main() -> int:
    args = parse_args()
    logger = configure_logger(args.log_level, log_file=args.log_file, log_format=args.log_format)
    audit = AuditLogger(args.audit_log) if args.audit_log else None

    if not os.path.exists(args.video):
        logger.error("Input video not found: %s", args.video)
        if audit:
            audit.emit("video_error", reason="missing_video")
        return 1

    if args.sample_every < 1:
        logger.error("--sample-every must be >= 1")
        return 1

    if audit:
        audit.emit(
            "video_start",
            video=args.video,
            cascade=args.cascade,
            output=args.output,
            summary_json=args.summary_json,
        )

    if not verify_cascade(args, logger, audit):
        return 1

    try:
        detector = validate_detector(args.cascade, logger)
        summary = run_video_detection(
            source_path=args.video,
            detector=detector,
            scale_factor=args.scale_factor,
            min_neighbors=args.min_neighbors,
            min_size=tuple(args.min_size),
            privacy=args.privacy,
            draw_labels=args.draw_labels,
            show_metrics=args.show_metrics,
            output_path=args.output,
            summary_logger=logger,
            sample_every=args.sample_every,
            max_frames=args.max_frames,
            snapshot_dir=args.snapshot_dir,
            snapshot_interval=args.snapshot_interval,
            no_display=args.no_display,
        )
    except (OSError, RuntimeError, ValueError) as exc:
        logger.error("%s", exc)
        if audit:
            audit.emit("video_error", reason="runtime_failure", message=str(exc))
        return 1

    logger.info(
        "Video finished: processed=%s frames_with_faces=%s total_faces=%s max_faces=%s avg_detect=%.4fs",
        summary.frames_processed,
        summary.frames_with_faces,
        summary.total_faces,
        summary.max_faces_in_frame,
        summary.avg_detection_seconds,
    )
    if args.summary_json:
        summary.summary_path = args.summary_json
        write_json(args.summary_json, summary.to_dict())
        logger.info("Wrote summary to %s", args.summary_json)
    if audit:
        audit.emit(
            "video_finish",
            ok=True,
            frames=summary.frames_processed,
            frames_with_faces=summary.frames_with_faces,
            total_faces=summary.total_faces,
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())