app.py

"""
dokoCame - Real-time video location identification service

Based on: https://github.com/freddyaboulton/fastrtc/tree/main/demo/object_detection
"""

import os
import time
import threading
import numpy as np
import cv2
from fastrtc import Stream, VideoStreamHandler, get_twilio_turn_credentials

from config.settings import settings


def is_hf_space():
    """Check if running on Hugging Face Spaces"""
    return os.environ.get("SPACE_ID") is not None


class AsyncAnalyzer:
    """非同期解析"""

    def __init__(self):
        self._ocr_texts = []
        self._vlm_keywords = []
        self._frame_count = 0
        self._last_ocr = 0
        self._last_vlm = 0
        self._ocr_busy = False
        self._vlm_busy = False
        self._lock = threading.Lock()
        self._ocr_engine = None
        self._vlm_analyzer = None
        self._init_done = False

    def _init(self):
        if self._init_done:
            return
        try:
            from core.ocr_engine import OCREngine
            from core.vlm_analyzer import VLMAnalyzer
            self._ocr_engine = OCREngine(lang=settings.ocr_lang)
            self._vlm_analyzer = VLMAnalyzer()
            self._init_done = True
        except:
            pass

    def _ocr_async(self, frame):
        if self._ocr_busy or not self._ocr_engine:
            return
        def run():
            try:
                self._ocr_busy = True
                texts = self._ocr_engine.detect_text_only(frame)
                with self._lock:
                    self._ocr_texts = [t for t in texts if t and t.strip()]
            except:
                pass
            finally:
                self._ocr_busy = False
        threading.Thread(target=run, daemon=True).start()

    def _vlm_async(self, frame):
        if self._vlm_busy or not self._vlm_analyzer:
            return
        if not self._vlm_analyzer.is_available:
            return
        def run():
            try:
                self._vlm_busy = True
                analysis = self._vlm_analyzer.analyze(frame)
                if analysis.success:
                    kw = self._vlm_analyzer.get_search_keywords(analysis)
                    with self._lock:
                        self._vlm_keywords = kw
            except:
                pass
            finally:
                self._vlm_busy = False
        threading.Thread(target=run, daemon=True).start()

    def process(self, frame):
        self._frame_count += 1
        now = time.time()
        if not self._init_done:
            self._init()
        if now - self._last_ocr > 1.0:  # 1秒間隔でOCR実行
            self._last_ocr = now
            self._ocr_async(frame)
        if now - self._last_vlm > 5.0:
            self._last_vlm = now
            self._vlm_async(frame)
        with self._lock:
            return {
                "ocr": self._ocr_texts.copy(),
                "vlm": self._vlm_keywords.copy(),
                "frame": self._frame_count,
                "ocr_busy": self._ocr_busy,
                "vlm_busy": self._vlm_busy,
            }


analyzer = AsyncAnalyzer()


class FloatingText:
    """ふわふわ浮かび上がって消えるテキスト管理"""

    def __init__(self, fade_duration=3.0):
        self._texts = []  # [(text, timestamp, y_offset), ...]
        self._fade_duration = fade_duration
        self._lock = threading.Lock()

    def add_texts(self, new_texts):
        """新しいテキストを追加"""
        now = time.time()
        with self._lock:
            # 既存のテキストと重複しないものだけ追加
            existing = {t[0] for t in self._texts}
            for text in new_texts:
                if text not in existing:
                    self._texts.append((text, now, 0))

    def get_visible_texts(self):
        """表示中のテキストと透明度を取得"""
        now = time.time()
        visible = []
        with self._lock:
            new_list = []
            for text, timestamp, y_offset in self._texts:
                age = now - timestamp
                if age < self._fade_duration:
                    # 透明度: 1.0 -> 0.0 にフェード
                    alpha = 1.0 - (age / self._fade_duration)
                    # 上に浮かび上がる効果（最大30px上昇）
                    float_y = int(age * 10)
                    visible.append((text, alpha, float_y))
                    new_list.append((text, timestamp, float_y))
            self._texts = new_list
        return visible


floating_texts = FloatingText(fade_duration=4.0)
_frame_count = 0


def detection(image):
    """
    Process video frame and add overlay.

    Args:
        image: numpy array (height, width, 3) RGB format
    Returns:
        numpy array (height, width, 3) RGB format
    """
    global _frame_count
    _frame_count += 1

    if image is None:
        return image

    try:
        output = image.copy()
        h, w = output.shape[:2]

        # 非同期解析を実行（OCR/VLM）
        result = analyzer.process(output)
        ocr_texts = result.get("ocr", [])

        # 新しいOCRテキストを追加
        if ocr_texts:
            floating_texts.add_texts(ocr_texts)

        # ふわふわ浮かぶテキストを描画
        visible = floating_texts.get_visible_texts()
        base_y = h - 50  # 画面下部から開始

        for i, (text, alpha, float_y) in enumerate(visible[:8]):  # 最大8個
            # 透明度に応じた色（フェードアウト効果）
            color_intensity = int(255 * alpha)
            color = (color_intensity, color_intensity, color_intensity)

            # テキストを短縮
            display_text = text[:25] + "..." if len(text) > 25 else text

            # 位置（上に浮かび上がる）
            y_pos = base_y - (i * 30) - float_y
            if y_pos < 30:
                continue

            # 背景（半透明風）
            text_size = cv2.getTextSize(display_text, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)[0]
            bg_alpha = int(150 * alpha)
            cv2.rectangle(output,
                          (25, y_pos - 20),
                          (35 + text_size[0], y_pos + 5),
                          (0, bg_alpha // 3, 0), -1)

            # テキスト描画
            cv2.putText(output, display_text, (30, y_pos),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, cv2.LINE_AA)

        print(f"[detection] Frame {_frame_count}, floating={len(visible)}")
        return output

    except Exception as e:
        print(f"[detection] ERROR: {e}")
        import traceback
        traceback.print_exc()
        return image


# 外カメラ設定
TRACK_CONSTRAINTS = {
    "facingMode": "environment",  # 外カメラ（背面）
    "width": {"ideal": 1280},
    "height": {"ideal": 720},
    "frameRate": {"ideal": 15},
}

# フッター削除 + ミラーリング無効化CSS
CUSTOM_CSS = """
footer { display: none !important; }
.built-with { display: none !important; }
video { transform: none !important; -webkit-transform: none !important; }
"""


if __name__ == "__main__":
    print("===== App Start:", time.strftime("%Y-%m-%d %H:%M:%S"), "=====")

    # RTC設定（HF Spacesの場合のみTwilio TURN使用）
    rtc_config = None
    if is_hf_space():
        print("[INFO] HF Spaces detected, trying Twilio TURN")
        try:
            rtc_config = get_twilio_turn_credentials()
            print("[INFO] Twilio TURN configured successfully")
        except Exception as e:
            print(f"[WARN] Twilio TURN failed: {e}")
            print("[INFO] Falling back to no TURN server")
    else:
        print("[INFO] Local mode, no TURN server")

    # Stream作成
    # Based on: https://fastrtc.org/userguide/video/
    # VideoStreamHandler で skip_frames=True を設定してラグ軽減
    stream = Stream(
        handler=VideoStreamHandler(detection, skip_frames=True, fps=30),
        modality="video",
        mode="send-receive",
        rtc_configuration=rtc_config,
        track_constraints=TRACK_CONSTRAINTS,
    )

    # UIにCSS追加
    if hasattr(stream, '_ui') and stream._ui:
        stream._ui.css = (stream._ui.css or "") + CUSTOM_CSS

    stream.ui.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
    )