app.py

"""
Forensic Touch Tracker - Gradio Space
=====================================
Tracks humans in video, estimates body/hand keypoints, detects surface proximity,
and logs potential fingerprint touch events for forensic evidence collection.

Optimized design:
- Single SAM3 pass: tracks persons + segments surfaces simultaneously (saves ~50% compute)
- Temporal contact grouping: groups consecutive touch frames into discrete events
- Persistent person IDs across entire video via SAM3 object IDs
- Precise fingertip localization via MediaPipe 21-landmark hands

Components:
- SAM3 Video (facebook/sam3) – promptable concept segmentation + tracking
- RT-DETR (PekingU/rtdetr_r50vd_coco_o365) – person detection box pre-filter
- ViTPose (usyd-community/vitpose-base-simple) – 17 COCO body keypoints
- MediaPipe Hands – 21 hand landmarks per hand (5 fingertips)
"""

import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import cv2
import numpy as np
import torch
from PIL import Image
import gradio as gr

# ---------------------------------------------------------------------------
# Optional dependencies
# ---------------------------------------------------------------------------
try:
    from transformers import (
        Sam3VideoModel,
        Sam3VideoProcessor,
        RTDetrForObjectDetection,
        AutoProcessor,
        VitPoseForPoseEstimation,
    )
    HAS_TRANSFORMERS = True
except Exception as e:
    HAS_TRANSFORMERS = False
    TRANSFORMERS_ERR = str(e)
    print("WARNING: transformers not available:", TRANSFORMERS_ERR)

try:
    import mediapipe as mp
    HAS_MEDIAPIPE = True
except Exception as e:
    HAS_MEDIAPIPE = False
    MEDIAPIPE_ERR = str(e)
    print("WARNING: mediapipe not available:", MEDIAPIPE_ERR)


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
@dataclass
class ForensicConfig:
    sam3_model_id: str = "facebook/sam3"
    rtdetr_model_id: str = "PekingU/rtdetr_r50vd_coco_o365"
    vitpose_model_id: str = "usyd-community/vitpose-base-simple"
    device: str = "cuda" if torch.cuda.is_available() else "cpu"
    max_frames: int = 0
    contact_threshold_px: int = 25
    min_contact_frames: int = 3
    confidence_threshold: float = 0.5
    save_debug_video: bool = True
    output_dir: str = "./forensic_output"
    surface_prompts: List[str] = field(default_factory=lambda: [
        "door handle", "countertop", "table", "desk", "wall", "railing",
        "keyboard", "mouse", "phone", "cup", "bottle", "drawer handle"
    ])


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class TouchEvent:
    timestamp_seconds_start: float
    timestamp_seconds_end: float
    frame_index_start: int
    frame_index_end: int
    person_id: int
    body_part: str
    touch_point: Tuple[int, int]          # median point across contact window
    surface: Optional[str]
    confidence: float
    bbox: List[int]
    video_resolution: Tuple[int, int]
    num_frames: int

    def to_dict(self) -> dict:
        return {
            "timestamp_seconds_start": round(self.timestamp_seconds_start, 3),
            "timestamp_seconds_end": round(self.timestamp_seconds_end, 3),
            "frame_index_start": self.frame_index_start,
            "frame_index_end": self.frame_index_end,
            "person_id": self.person_id,
            "body_part": self.body_part,
            "touch_point": {"x": self.touch_point[0], "y": self.touch_point[1]},
            "surface": self.surface,
            "confidence": round(self.confidence, 4),
            "bbox": self.bbox,
            "video_resolution": list(self.video_resolution),
            "num_frames": self.num_frames,
        }


# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------
def extract_frames(video_path: str, max_frames: int = 0):
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    frames = []
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        if max_frames and len(frames) >= max_frames:
            break
    cap.release()
    return frames, fps, (w, h)


def distance_point_to_mask(point: Tuple[int, int], mask: np.ndarray) -> float:
    ys, xs = np.where(mask)
    if len(xs) == 0:
        return float("inf")
    dx = xs - point[0]
    dy = ys - point[1]
    return float(np.sqrt((dx ** 2 + dy ** 2)).min())


def mask_center(mask: np.ndarray) -> Tuple[int, int]:
    ys, xs = np.where(mask)
    return int(xs.mean()), int(ys.mean())


def box_iou_xyxy(a, b):
    x1 = max(a[0], b[0]); y1 = max(a[1], b[1])
    x2 = min(a[2], b[2]); y2 = min(a[3], b[3])
    inter = max(0, x2 - x1) * max(0, y2 - y1)
    area_a = (a[2] - a[0]) * (a[3] - a[1])
    area_b = (b[2] - b[0]) * (b[3] - b[1])
    union = area_a + area_b - inter
    return inter / union if union > 0 else 0.0


# ---------------------------------------------------------------------------
# Hand landmarks
# ---------------------------------------------------------------------------
FINGERTIP_INDICES = [4, 8, 12, 16, 20]
FINGERTIP_NAMES = ["thumb_tip", "index_tip", "middle_tip", "ring_tip", "pinky_tip"]


def get_mediapipe_hands():
    if not HAS_MEDIAPIPE:
        return None
    mp_hands = mp.solutions.hands
    return mp_hands.Hands(
        static_image_mode=False,
        max_num_hands=2,
        min_detection_confidence=0.5,
        min_tracking_confidence=0.5,
    )


def extract_hand_fingertips(image: np.ndarray, hands_detector) -> List[dict]:
    results = []
    if hands_detector is None:
        return results
    h, w = image.shape[:2]
    mp_results = hands_detector.process(image)
    if not mp_results or not mp_results.multi_hand_landmarks:
        return results
    for hand_idx, hand_landmarks in enumerate(mp_results.multi_hand_landmarks):
        fingertips = {}
        for idx, name in zip(FINGERTIP_INDICES, FINGERTIP_NAMES):
            lm = hand_landmarks.landmark[idx]
            fingertips[name] = (int(lm.x * w), int(lm.y * h))
        results.append({"hand_index": hand_idx, "fingertips": fingertips})
    return results


# ---------------------------------------------------------------------------
# Core tracker
# ---------------------------------------------------------------------------
class ForensicTouchTracker:
    def __init__(self, config: Optional[ForensicConfig] = None):
        self.cfg = config or ForensicConfig()
        self.device = self.cfg.device
        self.output_dir = Path(self.cfg.output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

        self._sam3_model = None
        self._sam3_processor = None
        self._rtdetr_model = None
        self._rtdetr_processor = None
        self._vitpose_model = None
        self._vitpose_processor = None
        self._hands_detector = None

        self.events: List[TouchEvent] = []

    def _load_sam3(self):
        if self._sam3_model is not None:
            return
        if not HAS_TRANSFORMERS:
            raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}")
        print("Loading SAM3 Video...")
        self._sam3_model = Sam3VideoModel.from_pretrained(
            self.cfg.sam3_model_id, device_map=self.device
        )
        self._sam3_processor = Sam3VideoProcessor.from_pretrained(self.cfg.sam3_model_id)

    def _load_rtdetr(self):
        if self._rtdetr_model is not None:
            return
        if not HAS_TRANSFORMERS:
            raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}")
        print("Loading RT-DETR...")
        self._rtdetr_model = RTDetrForObjectDetection.from_pretrained(
            self.cfg.rtdetr_model_id, device_map=self.device
        )
        self._rtdetr_processor = AutoProcessor.from_pretrained(self.cfg.rtdetr_model_id)

    def _load_vitpose(self):
        if self._vitpose_model is not None:
            return
        if not HAS_TRANSFORMERS:
            raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}")
        print("Loading ViTPose...")
        self._vitpose_model = VitPoseForPoseEstimation.from_pretrained(
            self.cfg.vitpose_model_id, device_map=self.device
        )
        self._vitpose_processor = AutoProcessor.from_pretrained(self.cfg.vitpose_model_id)

    def _load_hands(self):
        if self._hands_detector is not None:
            return
        if not HAS_MEDIAPIPE:
            raise RuntimeError(f"mediapipe not available: {MEDIAPIPE_ERR}")
        self._hands_detector = get_mediapipe_hands()

    def _detect_person_boxes(self, image: Image.Image) -> np.ndarray:
        self._load_rtdetr()
        inputs = self._rtdetr_processor(images=image, return_tensors="pt").to(self._rtdetr_model.device)
        with torch.no_grad():
            outputs = self._rtdetr_model(**inputs)
        target_sizes = torch.tensor([(image.height, image.width)])
        results = self._rtdetr_processor.post_process_object_detection(
            outputs, target_sizes=target_sizes, threshold=self.cfg.confidence_threshold
        )[0]
        person_mask = results["labels"] == 0
        return results["boxes"][person_mask].cpu().numpy()

    def _estimate_poses(self, image: Image.Image, boxes_xyxy: np.ndarray):
        if len(boxes_xyxy) == 0:
            return []
        self._load_vitpose()
        boxes_xywh = boxes_xyxy.copy()
        boxes_xywh[:, 2] -= boxes_xywh[:, 0]
        boxes_xywh[:, 3] -= boxes_xywh[:, 1]
        inputs = self._vitpose_processor(image, boxes=[boxes_xywh], return_tensors="pt").to(self._vitpose_model.device)
        with torch.no_grad():
            outputs = self._vitpose_model(**inputs)
        poses = self._vitpose_processor.post_process_pose_estimation(outputs, boxes=[boxes_xywh])
        return poses[0] if isinstance(poses, list) and len(poses) > 0 else []

    def _sam3_single_pass(self, video_frames: List[np.ndarray]):
        """
        One SAM3 propagation with all prompts (person + surfaces).
        Returns per-frame: persons dict and surfaces list.
        """
        self._load_sam3()
        pil_frames = [Image.fromarray(f) for f in video_frames]
        inference_session = self._sam3_processor.init_video_session(
            video=pil_frames,
            inference_device=self.device,
            processing_device="cpu",
            video_storage_device="cpu",
        )
        all_prompts = ["person"] + self.cfg.surface_prompts
        inference_session = self._sam3_processor.add_text_prompt(
            inference_session=inference_session, text=all_prompts
        )
        per_frame_persons = []
        per_frame_surfaces = []
        max_track = self.cfg.max_frames if self.cfg.max_frames > 0 else len(pil_frames)
        for model_outputs in self._sam3_model.propagate_in_video_iterator(
            inference_session=inference_session, max_frame_num_to_track=max_track
        ):
            processed = self._sam3_processor.postprocess_outputs(inference_session, model_outputs)
            prompt_to_obj_ids = processed.get("prompt_to_obj_ids", {})
            obj_ids = processed.get("object_ids", [])
            boxes = processed.get("boxes", [])
            masks = processed.get("masks", [])
            scores = processed.get("scores", [])

            persons = {}
            surfaces = []
            for prompt, ids in prompt_to_obj_ids.items():
                for oid in ids:
                    idx = list(obj_ids).index(oid) if oid in list(obj_ids) else -1
                    if idx < 0:
                        continue
                    if scores is not None and len(scores) > idx and scores[idx] < self.cfg.confidence_threshold:
                        continue
                    mask = masks[idx].cpu().numpy() if hasattr(masks[idx], "cpu") else np.array(masks[idx])
                    bbox = boxes[idx].cpu().numpy().tolist() if hasattr(boxes[idx], "cpu") else list(boxes[idx])
                    entry = {"bbox": bbox, "mask": mask, "score": float(scores[idx]) if scores is not None and len(scores) > idx else 1.0}
                    if prompt == "person":
                        persons[int(oid)] = entry
                    else:
                        entry["name"] = prompt
                        entry["center"] = mask_center(mask)
                        surfaces.append(entry)
            per_frame_persons.append(persons)
            per_frame_surfaces.append(surfaces)
        return per_frame_persons, per_frame_surfaces

    def _get_touch_candidates(self, image: np.ndarray, person_boxes: np.ndarray) -> List[dict]:
        self._load_hands()
        candidates = []
        pil = Image.fromarray(image)
        poses = self._estimate_poses(pil, person_boxes)
        for i, box in enumerate(person_boxes):
            if i < len(poses):
                kp = poses[i].get("keypoints", [])
                for wrist_name, wrist_idx in [("left_wrist", 9), ("right_wrist", 10)]:
                    if wrist_idx < len(kp):
                        x, y = float(kp[wrist_idx][0]), float(kp[wrist_idx][1])
                        candidates.append({
                            "person_idx": i,
                            "body_part": wrist_name,
                            "point": (int(x), int(y)),
                            "source": "body_pose",
                        })
            x1, y1, x2, y2 = map(int, box)
            x1, y1 = max(0, x1), max(0, y1)
            x2, y2 = min(image.shape[1], x2), min(image.shape[0], y2)
            crop = image[y1:y2, x1:x2]
            if crop.size == 0:
                continue
            hands = extract_hand_fingertips(crop, self._hands_detector)
            for hand in hands:
                for tip_name, tip_coords in hand["fingertips"].items():
                    candidates.append({
                        "person_idx": i,
                        "body_part": f"hand_{hand['hand_index']}_{tip_name}",
                        "point": (x1 + tip_coords[0], y1 + tip_coords[1]),
                        "source": "hand_landmark",
                    })
        return candidates

    def process_video(self, video_path: str, progress=None):
        print(f"[ForensicTouchTracker] Processing {video_path}")
        frames, fps, (W, H) = extract_frames(video_path, max_frames=self.cfg.max_frames)
        total = len(frames)
        print(f"  -> {total} frames @ {fps:.2f} fps, {W}x{H}")
        if total == 0:
            return [], str(self.output_dir)

        if progress:
            progress(0.05, desc="Extracting frames...")

        # Single SAM3 pass for persons + surfaces
        print("[SAM3] Single-pass tracking & surface segmentation...")
        person_tracks, surface_masks = self._sam3_single_pass(frames)
        print(f"  -> got persons in {len(person_tracks)} frames, surfaces in {len(surface_masks)} frames")
        if progress:
            progress(0.45, desc="Tracking & segmentation done")

        # Stage 3: per-frame pose + hands + contact inference with temporal grouping
        print("[Stage 3] Pose, hands, and temporal contact grouping...")
        active_contacts: Dict[Tuple[int, str], dict] = {}
        annotated_frames = []

        def emit_event(key, rec):
            points = rec["points"]
            x = int(np.median([p[0] for p in points]))
            y = int(np.median([p[1] for p in points]))
            dists = rec["dists"]
            med_dist = float(np.median(dists))
            conf = max(0.0, 1.0 - (med_dist / (self.cfg.contact_threshold_px * 2)))
            self.events.append(TouchEvent(
                timestamp_seconds_start=rec["time_start"],
                timestamp_seconds_end=rec["time_end"],
                frame_index_start=rec["frame_start"],
                frame_index_end=rec["frame_end"],
                person_id=key[0],
                body_part=key[1],
                touch_point=(x, y),
                surface=rec["surface"],
                confidence=round(conf, 4),
                bbox=rec["bbox"],
                video_resolution=(W, H),
                num_frames=rec["count"],
            ))

        for frame_idx, frame in enumerate(frames):
            if progress and frame_idx % 10 == 0:
                progress(0.45 + 0.50 * (frame_idx / total), desc=f"Frame {frame_idx}/{total}")
            t_sec = frame_idx / fps
            persons = person_tracks[frame_idx]
            surfaces = surface_masks[frame_idx]
            if len(persons) == 0:
                # flush all active contacts (no persons => no contact possible)
                for key, rec in list(active_contacts.items()):
                    if rec["count"] >= self.cfg.min_contact_frames:
                        emit_event(key, rec)
                    del active_contacts[key]
                annotated_frames.append(frame)
                continue

            pil = Image.fromarray(frame)
            person_boxes = self._detect_person_boxes(pil)
            if len(person_boxes) == 0:
                for key, rec in list(active_contacts.items()):
                    if rec["count"] >= self.cfg.min_contact_frames:
                        emit_event(key, rec)
                    del active_contacts[key]
                annotated_frames.append(frame)
                continue

            candidates = self._get_touch_candidates(frame, person_boxes)

            # Map each candidate to SAM3 person ID via IoU
            matched_ids = {}
            for cand in candidates:
                i = cand["person_idx"]
                best_iou = 0.0
                best_oid = None
                for oid, pdata in persons.items():
                    iou = box_iou_xyxy(pdata["bbox"], person_boxes[i])
                    if iou > best_iou:
                        best_iou = iou
                        best_oid = oid
                matched_ids[i] = best_oid

            # Determine which keys are still in contact this frame
            touched_keys = set()
            for cand in candidates:
                pid = matched_ids.get(cand["person_idx"])
                if pid is None:
                    continue
                point = cand["point"]
                best_surface = None
                best_dist = float("inf")
                for surf in surfaces:
                    d = distance_point_to_mask(point, surf["mask"])
                    if d < best_dist:
                        best_dist = d
                        best_surface = surf
                if best_surface is not None and best_dist <= self.cfg.contact_threshold_px:
                    key = (pid, cand["body_part"])
                    touched_keys.add(key)
                    if key not in active_contacts:
                        active_contacts[key] = {
                            "frame_start": frame_idx,
                            "time_start": t_sec,
                            "points": [],
                            "dists": [],
                            "surface": best_surface["name"],
                            "bbox": person_boxes[cand["person_idx"]].astype(int).tolist(),
                            "count": 0,
                        }
                    rec = active_contacts[key]
                    rec["frame_end"] = frame_idx
                    rec["time_end"] = t_sec
                    rec["points"].append(point)
                    rec["dists"].append(best_dist)
                    rec["count"] += 1

            # Emit events for keys that lost contact this frame
            for key, rec in list(active_contacts.items()):
                if key not in touched_keys:
                    if rec["count"] >= self.cfg.min_contact_frames:
                        emit_event(key, rec)
                    del active_contacts[key]

            annotated = self._annotate_frame(frame, persons, surfaces, candidates)
            annotated_frames.append(annotated)

        # Flush remaining active contacts at EOF
        for key, rec in list(active_contacts.items()):
            if rec["count"] >= self.cfg.min_contact_frames:
                emit_event(key, rec)
            del active_contacts[key]

        self._write_outputs(annotated_frames, fps, (W, H))
        print(f"[Done] {len(self.events)} events. Output: {self.output_dir}")
        return self.events, str(self.output_dir)

    def _annotate_frame(self, frame, persons, surfaces, candidates):
        out = frame.copy()
        for surf in surfaces:
            m = surf["mask"]
            if m.shape[:2] != out.shape[:2]:
                continue
            overlay = np.zeros_like(out)
            overlay[m > 0] = np.array([0, 255, 255], dtype=np.uint8)
            out = cv2.addWeighted(out, 1.0, overlay, 0.3, 0)
            cx, cy = surf["center"]
            cv2.putText(out, surf["name"], (cx, cy), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 120, 120), 2)
        for oid, pdata in persons.items():
            x1, y1, x2, y2 = map(int, pdata["bbox"])
            cv2.rectangle(out, (x1, y1), (x2, y2), (0, 255, 0), 2)
            cv2.putText(out, f"Person {oid}", (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
        for cand in candidates:
            pt = cand["point"]
            color = (255, 0, 0) if "hand" in cand["body_part"] else (0, 0, 255)
            cv2.circle(out, pt, 4, color, -1)
        return out

    def _write_outputs(self, annotated_frames, fps, resolution):
        log_path = self.output_dir / "forensic_touch_log.jsonl"
        with open(log_path, "w") as f:
            for ev in self.events:
                f.write(json.dumps(ev.to_dict(), default=str) + "\n")
        summary = {
            "total_events": len(self.events),
            "unique_persons": sorted({e.person_id for e in self.events}),
            "surfaces_touched": sorted({e.surface for e in self.events if e.surface}),
            "body_parts": sorted({e.body_part for e in self.events}),
            "resolution": list(resolution),
            "fps": fps,
        }
        with open(self.output_dir / "forensic_summary.json", "w") as f:
            json.dump(summary, f, indent=2)
        if self.cfg.save_debug_video and annotated_frames:
            vid_path = self.output_dir / "annotated_video.mp4"
            fourcc = cv2.VideoWriter_fourcc(*"mp4v")
            writer = cv2.VideoWriter(str(vid_path), fourcc, fps, resolution)
            for af in annotated_frames:
                writer.write(cv2.cvtColor(af, cv2.COLOR_RGB2BGR))
            writer.release()


# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------
def run_tracker(video_file, max_frames, contact_threshold, min_contact_frames):
    if video_file is None:
        return "No video uploaded.", None, None, None
    cfg = ForensicConfig(
        max_frames=int(max_frames),
        contact_threshold_px=int(contact_threshold),
        min_contact_frames=int(min_contact_frames),
        output_dir="./forensic_output",
    )
    tracker = ForensicTouchTracker(cfg)
    events, out_dir = tracker.process_video(video_file, progress=gr.Progress())

    if not events:
        md = "## No touch events detected.\n\nTry lowering the contact threshold or min-contact-frames."
    else:
        md = f"## Detected {len(events)} Forensic Touch Event(s)\n\n"
        md += "| Start | End | Person | Body Part | Surface | Conf | Frames | Touch Point |\n"
        md += "|-------|-----|--------|-----------|---------|------|--------|-------------|\n"
        for ev in events:
            md += f"| {ev.timestamp_seconds_start:.2f}s | {ev.timestamp_seconds_end:.2f}s | {ev.person_id} | {ev.body_part} | {ev.surface or 'unknown'} | {ev.confidence:.2f} | {ev.num_frames} | ({ev.touch_point[0]},{ev.touch_point[1]}) |\n"

    log_path = os.path.join(out_dir, "forensic_touch_log.jsonl")
    summary_path = os.path.join(out_dir, "forensic_summary.json")
    video_path_out = os.path.join(out_dir, "annotated_video.mp4")
    files_out = [p for p in [log_path, summary_path, video_path_out] if os.path.exists(p)]
    return md, files_out[0] if len(files_out) > 0 else None, files_out[1] if len(files_out) > 1 else None, files_out[2] if len(files_out) > 2 else None


with gr.Blocks(title="Forensic Touch Tracker") as demo:
    gr.Markdown(
        """
        # 🕵️ Forensic Touch Tracker
        **Track human movement and log touch points for fingerprint forensics.**

        Upload a surveillance or body-cam video. The pipeline:
        1. **Track persons** with persistent IDs across the entire video (SAM3 Video)
        2. **Segment forensic surfaces** in a single pass — door handles, countertops, walls, etc. (SAM3 Video)
        3. **Estimate body poses** (ViTPose) and **hand fingertips** (MediaPipe)
        4. **Log discrete touch events** when fingertips come near a surface for ≥N consecutive frames

        Outputs: structured JSONL forensic log, summary JSON, and an annotated debug video.
        """
    )
    with gr.Row():
        with gr.Column(scale=1):
            video_input = gr.Video(label="Upload Video", format="mp4")
            max_frames = gr.Number(value=0, label="Max Frames (0 = all)", precision=0)
            contact_threshold = gr.Number(value=25, label="Contact Threshold (px)", precision=0)
            min_contact_frames = gr.Number(value=3, label="Min Contact Frames", precision=0)
            run_btn = gr.Button("Run Analysis", variant="primary")
        with gr.Column(scale=2):
            results_md = gr.Markdown("## Results will appear here")
    with gr.Row():
        log_file = gr.File(label="Forensic Touch Log (JSONL)")
        summary_file = gr.File(label="Forensic Summary (JSON)")
        debug_video = gr.Video(label="Annotated Debug Video")

    run_btn.click(
        fn=run_tracker,
        inputs=[video_input, max_frames, contact_threshold, min_contact_frames],
        outputs=[results_md, log_file, summary_file, debug_video],
    )

if __name__ == "__main__":
    demo.launch()