momentry_core/scripts/appearance_processor_v2.py

#!/opt/homebrew/bin/python3.11
"""
Appearance Expansion Processor V2

Calls swift_appearance_expansion which:
1. Reads pose.json (from pose expansion)
2. Expands appearance detection from pose frames
3. Stops when 3 consecutive frames have HSV similarity < 0.5
4. Outputs at 8Hz sampling (floor(fps/8))

Flow:
    face_processor.py → face.json
    store_traced_faces.py → face_traced.json (with trace_id)
    pose_processor.py → pose.json
    appearance_processor.py → appearance.json (this script)
"""

import sys
import os
import json
import argparse
import subprocess
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from redis_publisher import RedisPublisher

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SWIFT_BIN = os.path.join(SCRIPT_DIR, "swift_processors", ".build", "release", "swift_appearance_expansion")
SWIFT_BIN_DEBUG = os.path.join(SCRIPT_DIR, "swift_processors", ".build", "debug", "swift_appearance_expansion")

OUTPUT_DIR = os.environ.get("MOMENTRY_OUTPUT_DIR", "/Users/accusys/momentry/output")


def process_appearance(
    video_path: str,
    output_path: str,
    uuid: str = "",
    publisher: RedisPublisher = None,
) -> dict:
    """Process appearance expansion from pose frames.
    
    Args:
        video_path: Path to video file
        output_path: Path to output appearance.json
        uuid: File UUID for logging
        publisher: Redis publisher for progress updates
    """
    # Check if appearance.json already exists
    if os.path.exists(output_path):
        with open(output_path) as f:
            data = json.load(f)
        frame_count = len(data.get("frames", []))
        print(f"[Appearance] Output exists: {output_path} ({frame_count} frames)", file=sys.stderr)
        if publisher:
            publisher.progress("appearance", 100, 100, f"{frame_count} frames (exists)")
        return data
    
    # Determine file_uuid from output_path
    file_uuid = os.path.basename(output_path).replace(".appearance.json", "")
    
    # Find pose.json
    pose_path = os.path.join(OUTPUT_DIR, f"{file_uuid}.pose.json")
    
    if not os.path.exists(pose_path):
        print(f"[Appearance] ERROR: pose.json not found for {file_uuid}", file=sys.stderr)
        # Return empty result
        empty_result = {"frame_count": 0, "fps": 0.0, "frames": []}
        with open(output_path, "w") as f:
            json.dump(empty_result, f)
        return empty_result
    
    # Build swift_appearance_expansion if needed
    swift_bin = SWIFT_BIN if os.path.exists(SWIFT_BIN) else SWIFT_BIN_DEBUG
    if not os.path.exists(swift_bin):
        build_dir = os.path.join(SCRIPT_DIR, "swift_processors")
        print(f"[Appearance] Building swift_appearance_expansion in {build_dir}...", file=sys.stderr)
        result = subprocess.run(
            ["swift", "build", "-c", "release", "--product", "swift_appearance_expansion"],
            cwd=build_dir, capture_output=True, text=True
        )
        if result.returncode != 0:
            print(f"[Appearance] Build failed: {result.stderr}", file=sys.stderr)
            raise RuntimeError("Failed to build swift_appearance_expansion")
        swift_bin = SWIFT_BIN
    
    # Run swift_appearance_expansion
    cmd = [
        swift_bin,
        video_path,
        pose_path,
        output_path,
    ]
    if uuid:
        cmd.extend(["--uuid", uuid])
    
    print(f"[Appearance] Running: {' '.join(cmd)}", file=sys.stderr)
    
    t0 = time.time()
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    
    # Monitor progress
    last_progress = ""
    while proc.poll() is None:
        time.sleep(5)
        # Read stderr for progress
        try:
            import select
            if select.select([proc.stderr], [], [], 0)[0]:
                line = proc.stderr.readline().strip()
                if line and line != last_progress:
                    last_progress = line
                    print(f"[Appearance] {line}", file=sys.stderr)
                    if publisher and "frames" in line:
                        publisher.progress("appearance", 50, 100, line)
        except Exception:
            pass
    
    # Read remaining output
    stdout, stderr = proc.communicate()
    if stdout:
        print(stdout, file=sys.stderr)
    if stderr:
        print(stderr, file=sys.stderr)
    
    elapsed = time.time() - t0
    
    if proc.returncode != 0:
        print(f"[Appearance] ERROR: swift_appearance_expansion exited with code {proc.returncode}", file=sys.stderr)
        if publisher:
            publisher.error("appearance", f"Process failed with code {proc.returncode}")
        raise RuntimeError(f"swift_appearance_expansion failed: {proc.returncode}")
    
    # Load result
    if not os.path.exists(output_path):
        print(f"[Appearance] ERROR: Output file not created: {output_path}", file=sys.stderr)
        raise RuntimeError("Appearance output not created")
    
    with open(output_path) as f:
        result = json.load(f)
    
    frame_count = len(result.get("frames", []))
    print(f"[Appearance] Done: {frame_count} frames in {elapsed:.1f}s", file=sys.stderr)
    
    if publisher:
        publisher.progress("appearance", 100, 100, f"{frame_count} frames")
        publisher.complete("appearance", f"{frame_count} frames")
    
    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Appearance Expansion Processor")
    parser.add_argument("video_path", help="Video file path")
    parser.add_argument("output_path", help="Output appearance.json path")
    parser.add_argument("--uuid", "-u", default="", help="File UUID for logging")
    args = parser.parse_args()
    
    publisher = RedisPublisher(args.uuid) if args.uuid else None
    if publisher:
        publisher.info("appearance", "APPEARANCE_START")
    
    result = process_appearance(args.video_path, args.output_path, args.uuid, publisher)
    print(f"Appearance: {len(result.get('frames', []))} frames with appearance")