""" 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, )