# app.py # Live Video Insight — real-time video understanding (English UI) # Defaults set to your requested live URLs: # 1) VDOT – Fairfax NO0451 (HLS) # 2) Red Bull Bike – RBMN (HLS) # Notes: # • Start/Stop buttons pinned at top; robust lifecycle. # • Preview fits card; no deprecated Streamlit args (uses width="stretch"). # • Optional credentials only for Custom URLs. # • Natural preview pacing; default sampling=10 FPS. # • Qwen2-VL chat template under the hood; AMP deprecation fixed. import os import sys import time import queue import threading from collections import deque from typing import List, Optional from urllib.parse import urlparse, urlunparse import io, base64 # (media store bypass için) # Windows/Tornado event loop stability for Streamlit websockets try: if sys.platform.startswith("win"): import asyncio asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) except Exception: pass import av import torch import streamlit as st from PIL import Image import gc # RAM temizliği için APP_TITLE = "Live Video Insight" APP_TAGLINE = "Real-time video understanding." MODEL_ID = os.getenv("VISION_ENGINE_ID", "Qwen/Qwen2-VL-2B-Instruct") # Optional: attach Streamlit ScriptRunContext to threads try: from streamlit.runtime.scriptrunner import add_script_run_ctx except Exception: add_script_run_ctx = None st.set_page_config(page_title=APP_TITLE, page_icon="🎥", layout="wide") # ==================== Theme & layout ==================== st.markdown(""" """, unsafe_allow_html=True) # ==================== State ==================== if "running" not in st.session_state: st.session_state.running = False if "stop_event" not in st.session_state: st.session_state.stop_event = threading.Event() # ==================== Default demo URLs (your picks) ==================== DEMO_OPTIONS = { "VDOT – Fairfax NO0451": "https://media-sfs1.vdotcameras.com:443/rtplive/NO0451/playlist.m3u8", "Red Bull Bike – RBMN": "https://rbmn-live.akamaized.net/hls/live/590964/BoRB-AT/master_3360.m3u8", } # ==================== Sidebar (Start/Stop on top) ==================== st.sidebar.markdown('', unsafe_allow_html=True) btn_cols = st.sidebar.columns(2) start_clicked = btn_cols[0].button("Start ▶️") stop_clicked = btn_cols[1].button("Stop ⏹") # Map button clicks to state transitions (idempotent, robust) if start_clicked and not st.session_state.running: # yeni run öncesi hafif temizlik try: if torch.cuda.is_available(): torch.cuda.empty_cache() except Exception: pass gc.collect() st.session_state.running = True st.session_state.stop_event.clear() if stop_clicked and st.session_state.running: st.session_state.stop_event.set() # Source mode: Demo vs Custom st.sidebar.markdown('', unsafe_allow_html=True) source_mode = st.sidebar.radio("Mode", ["Demo URLs", "Custom URL"], index=0, horizontal=True) if source_mode == "Demo URLs": demo_label = st.sidebar.selectbox("Pick a demo", list(DEMO_OPTIONS.keys()), index=0) source_final = DEMO_OPTIONS[demo_label] st.sidebar.caption(f"Selected demo URL:\n{source_final}") auth_enabled = False user = pwd = "" else: # Custom URL + optional credentials source_type = st.sidebar.selectbox("Type", ["Video URL (HTTP/HTTPS)", "IP Camera (RTSP)", "Local File"], index=0) default_hint = { "Video URL (HTTP/HTTPS)": list(DEMO_OPTIONS.values())[0], "IP Camera (RTSP)" : "rtsp://ip:port/stream", "Local File" : "C:\\videos\\video.mp4 or /path/video.mp4" }[source_type] source_input = st.sidebar.text_input("Address or path", value=default_hint, placeholder=default_hint, key="custom_source") cred_disabled = st.session_state.running auth_enabled = False; user = pwd = "" if source_type in ("Video URL (HTTP/HTTPS)", "IP Camera (RTSP)"): auth_enabled = st.sidebar.toggle("Use credentials (optional)", value=False, help="Enable if the URL requires username/password.", disabled=cred_disabled) if auth_enabled: user = st.sidebar.text_input("Username", value="", placeholder="user", disabled=cred_disabled) pwd = st.sidebar.text_input("Password", value="", placeholder="password", type="password", disabled=cred_disabled) source_final = source_input # Advanced groups with st.sidebar.expander("Analysis (advanced)", expanded=False): fps_sample = st.slider("Frame sampling (FPS)", 1, 20, 10, help="Frames per second sampled for analysis.", disabled=st.session_state.running) segment_seconds = st.slider("Segment length (sec)", 1, 8, 2, disabled=st.session_state.running) resp_len = st.slider("Response length (tokens)", 16, 128, 64, disabled=st.session_state.running) style = st.selectbox("Commentary style", ["Play-by-play (concise)", "Tactical analysis (succinct)", "Highlights only (major moments)"], disabled=st.session_state.running) with st.sidebar.expander("Performance (advanced)", expanded=False): preview_fps = st.slider("Preview FPS cap", 10, 30, 25, help="25–30 looks natural.", disabled=st.session_state.running) gpu_mode = st.toggle("GPU mode (if available)", value=True, help="Takes effect on next run.", disabled=st.session_state.running) eff_mode = st.toggle("Efficiency mode", value=True, help="Downscale frames to 512px; keep responses short.", disabled=st.session_state.running) fast_processor = st.toggle("Fast processor", value=True, help="Takes effect on next run.", disabled=st.session_state.running) with st.sidebar.expander("History (advanced)", expanded=False): console_max = st.slider("Live Console lines", 5, 300, 60, disabled=st.session_state.running) summary_max = st.slider("Segment Summary items", 5, 300, 60, disabled=st.session_state.running) # ==================== Header ==================== st.title(APP_TITLE) st.write(APP_TAGLINE) # ==================== Helpers ==================== def is_rtsp(url: str) -> bool: return url.strip().lower().startswith("rtsp") def apply_credentials(url: str, username: str, password: str) -> str: if not url or not username: return url p = urlparse(url) if p.scheme not in ("http", "https", "rtsp"): return url hostpart = p.netloc.split("@", 1)[-1] auth = f"{username}:{password}" if password else username return urlunparse(p._replace(netloc=f"{auth}@{hostpart}")) def safe_ts(frame, vstream) -> Optional[float]: if getattr(frame, "time", None) is not None: return float(frame.time) if frame.pts is not None and vstream and vstream.time_base: return float(frame.pts * vstream.time_base) return None def downscale(img: Image.Image, max_side: int = 512) -> Image.Image: w, h = img.size m = max(w, h) if m <= max_side: return img s = max_side / float(m) return img.resize((int(w*s), int(h*s)), Image.BILINEAR) # --- Media store bypass: st.image yerine data-URI --- def img_to_data_uri(img: Image.Image, quality: int = 80) -> str: buf = io.BytesIO() img.convert("RGB").save(buf, format="JPEG", quality=quality, optimize=True) b64 = base64.b64encode(buf.getvalue()).decode("ascii") return f"data:image/jpeg;base64,{b64}" def safe_image_data_uri(placeholder, img: Image.Image, ts: Optional[float] = None, quality: int = 80): try: uri = img_to_data_uri(img, quality=quality) cap = f"t = {ts:.2f}s" if ts is not None else "" html = f'''

''' placeholder.markdown(html, unsafe_allow_html=True) return True except Exception: return False # --- Hafıza/CPU temizliği yardımcıları --- def purge_queue(q: "queue.Queue"): try: while True: q.get_nowait() except queue.Empty: pass def finalize_and_free(decoder_obj, worker_obj, buffers: List, deques: List, queues: List): # Threadleri durdur/kat try: if worker_obj is not None: worker_obj.stop() except Exception: pass try: if decoder_obj is not None and hasattr(decoder_obj, "stop_event"): pass # decoder loop stop_event ile çıkıyor except Exception: pass try: if decoder_obj is not None: decoder_obj.join(timeout=1.0) except Exception: pass try: if worker_obj is not None: worker_obj.join(timeout=1.0) except Exception: pass # Buffer/queue temizle for b in buffers: try: if hasattr(b, "clear"): b.clear() except Exception: pass for d in deques: try: d.clear() except Exception: pass for q in queues: try: purge_queue(q) except Exception: pass # Torch/CUDA bellek try: if torch.cuda.is_available(): torch.cuda.empty_cache() except Exception: pass gc.collect() # ==================== Engine Loader (Qwen2-VL under the hood) ==================== @st.cache_resource(show_spinner=False) # suppress default status box def load_engine(model_id: str, use_gpu: bool, use_fast_processor: bool): from transformers import AutoProcessor, Qwen2VLForConditionalGeneration from qwen_vl_utils import process_vision_info # noqa cuda_built = getattr(torch.backends, "cuda", None) and torch.backends.cuda.is_built() cuda_ok = bool(use_gpu and torch.cuda.is_available() and cuda_built) device = torch.device("cuda") if cuda_ok else torch.device("cpu") amp_dtype = torch.float16 if device.type == "cuda" else torch.float32 processor = AutoProcessor.from_pretrained( model_id, trust_remote_code=True, use_fast=bool(use_fast_processor) ) model = Qwen2VLForConditionalGeneration.from_pretrained( model_id, trust_remote_code=True, dtype=amp_dtype ).to(device).eval() # Deterministic defaults gcfg = getattr(model, "generation_config", None) if gcfg is not None: for k in ("temperature", "top_p", "top_k"): if hasattr(gcfg, k): setattr(gcfg, k, None) tokenizer = getattr(processor, "tokenizer", None) if tokenizer is None: raise RuntimeError("Tokenizer not available from processor; please update transformers.") device_label = "CPU" if device.type == "cuda": try: device_label = f"GPU ({torch.cuda.get_device_name(0)})" except Exception: device_label = "GPU" if device.type == "cpu": try: torch.set_num_threads(max(1, os.cpu_count() // 2)) except Exception: pass cuda_hint = None if use_gpu and device.type == "cpu": cuda_ver = getattr(torch.version, "cuda", None) cuda_hint = ( f"CUDA not available to PyTorch. torch.version.cuda={cuda_ver}, " f"torch.backends.cuda.is_built={cuda_built}, torch.cuda.is_available={torch.cuda.is_available()}." " Install a CUDA-enabled PyTorch build from the official selector." ) return (tokenizer, processor, model, device), device_label, cuda_hint # ==================== Prompting ==================== SYSTEM_PROMPT = ( "You are a real-time video analyst. Respond ONLY in English. " "Use short, factual sentences grounded in the provided frames. " "Do not speculate about unseen causes or off-screen events. " "When nothing occurs, nothing say." ) def build_analysis_instruction(kind: str, t0: float, t1: float) -> str: if kind == "Tactical analysis (succinct)": return ( f"Segment {t0:.1f}–{t1:.1f}s. Provide 1–2 sentences on formations, movements, and spatial patterns. " "Be objective. No questions. No speculation." ) if kind == "Highlights only (major moments)": return ( f"Segment {t0:.1f}–{t1:.1f}s. Report only decisive highlights (e.g., goals/shots on target/turnovers/aces). " "One short sentence." ) return ( f"Segment {t0:.1f}–{t1:.1f}s. Describe visible actions in present tense. " "Use short factual sentences; avoid guesses." ) def pick_k(frames: List[Image.Image], k: int = 4) -> List[Image.Image]: if len(frames) <= k: return frames idxs = [round(i*(len(frames)-1)/(k-1)) for i in range(k)] return [frames[i] for i in idxs] def build_inputs_qwen(tokenizer, processor, images: List[Image.Image], instruction: str, device: torch.device): # Qwen2-VL chat template: system + user([{image...}, {text...}]) from qwen_vl_utils import process_vision_info messages = [ {"role": "system", "content": [{"type": "text", "text": SYSTEM_PROMPT}]}, {"role": "user", "content": ([{"type": "image", "image": im} for im in images] + [{"type": "text", "text": instruction}])} ] text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) image_inputs, video_inputs = process_vision_info(messages) inputs = processor(text=[text], images=image_inputs, videos=video_inputs, padding=True, return_tensors="pt") for k, v in list(inputs.items()): if torch.is_tensor(v): inputs[k] = v.to(device) return inputs # ==================== Decoder (Producer) ==================== class Decoder(threading.Thread): """pyAV decoder. Real-time pacing for HTTP/Local; RTSP flows as-is.""" def __init__(self, src: str, fps_target: int, buf: deque, stop_event: threading.Event, is_rtsp_src: bool): super().__init__(daemon=True) self.src = src; self.fps_target = fps_target; self.buf = buf self.stop_event = stop_event; self.is_rtsp_src = is_rtsp_src self.err: Optional[Exception] = None self._container = None; self._vstream = None def run(self): try: options = {"rtsp_transport": "tcp", "stimeout": "5000000"} if self.is_rtsp_src else {} self._container = av.open(self.src, options=options) self._vstream = next((s for s in self._container.streams if s.type == "video"), None) if self._vstream is None: raise RuntimeError("No video stream detected.") step = 1.0 / float(self.fps_target) next_t = 0.0 wall0 = None; ts0 = None for packet in self._container.demux(self._vstream): if self.stop_event.is_set(): break for f in packet.decode(): t = safe_ts(f, self._vstream) if t is None: continue if t + 1e-6 < next_t: continue if not self.is_rtsp_src: if wall0 is None: wall0 = time.time(); ts0 = t target_wall = wall0 + (t - ts0) now = time.time() if target_wall > now: dt = target_wall - now while dt > 0 and (not self.stop_event.is_set()): time.sleep(min(dt, 0.03)) now = time.time() dt = target_wall - now try: self.buf.append((f.to_image(), t)) except Exception: continue next_t += step except Exception as e: self.err = e finally: try: if self._container is not None: self._container.close() except Exception: pass # ==================== Inference Worker ==================== class InferenceWorker(threading.Thread): def __init__(self, engine, job_q: "queue.Queue", out_q: "queue.Queue", max_tokens: int): super().__init__(daemon=True) self.engine = engine; self.job_q = job_q; self.out_q = out_q self.max_tokens = int(max_tokens); self.alive = True def run(self): tokenizer, processor, model, device = self.engine while self.alive: try: frames, t0, t1, instruction = self.job_q.get(timeout=0.1) except queue.Empty: continue try: picks = pick_k(frames, 4) inputs = build_inputs_qwen(tokenizer, processor, picks, instruction, device) start = time.time() with torch.inference_mode(): if device.type == "cuda": with torch.amp.autocast("cuda", dtype=torch.float16): outputs = model.generate(**inputs, max_new_tokens=self.max_tokens, do_sample=False, use_cache=True) else: outputs = model.generate(**inputs, max_new_tokens=self.max_tokens, do_sample=False, use_cache=True) latency = time.time() - start input_len = inputs["input_ids"].shape[1] new_ids = outputs[0][input_len:] text = tokenizer.decode(new_ids, skip_special_tokens=True).strip() self.out_q.put((t0, t1, text, latency)) except Exception as e: self.out_q.put((t0, t1, f"[analysis error] {e}", None)) def stop(self): self.alive = False # ==================== Page layout ==================== left, right = st.columns([1.05, 1.0]) with left: st.subheader("Video") st.markdown('

', unsafe_allow_html=True) # The .video-frame wrapper ensures fit-to-card visuals st.markdown('

', unsafe_allow_html=True) video_img = st.empty() st.markdown('

', unsafe_allow_html=True) meta_row = st.columns([1,1,1]) fps_badge = meta_row[0].markdown('Sampling: -- fps', unsafe_allow_html=True) time_badge = meta_row[1].markdown('Time: 0.0s', unsafe_allow_html=True) seg_badge = meta_row[2].markdown('Segment: --', unsafe_allow_html=True) st.markdown('

', unsafe_allow_html=True) st.caption("If the URL requires credentials, switch to Custom mode and enable 'Use credentials'. RTSP over TCP is preferred.") st.markdown('

', unsafe_allow_html=True) with right: st.subheader("Live Console") console_box = st.empty() st.subheader("Segment Summary") summary_box = st.empty() st.markdown('

', unsafe_allow_html=True) # ==================== Misc UI helpers ==================== def render_scrollbox(lines: deque[str], dom_id: str) -> str: items = "".join(f'

{l}

' for l in reversed(lines)) return f'''

{items}

''' def safe_markdown(placeholder, html: str): try: placeholder.markdown(html, unsafe_allow_html=True); return True except Exception: return False def safe_image(placeholder, img, **kwargs): # Kept for compatibility; artık data-URI kullanılıyor. try: placeholder.image(img, width="stretch", **kwargs) return True except Exception: return False # ==================== RUN ==================== if st.session_state.running: # Pre-run validation if source_mode == "Custom URL": if not source_final.strip(): st.warning("Please provide a valid source.") st.session_state.running = False st.stop() if source_final.lower().startswith("rtsp"): pass elif source_final.lower().startswith(("http://", "https://")): pass elif source_final and os.path.isfile(source_final): pass else: st.error("Unsupported source.") st.session_state.running = False st.stop() else: pass final_source = source_final if source_mode == "Custom URL" and auth_enabled and source_final.lower().startswith(("http://", "https://", "rtsp")): final_source = apply_credentials(source_final, user.strip(), pwd) seg_len_frames = max(1, int(segment_seconds * fps_sample)) frames_buf: List[Image.Image] = [] times_buf: List[float] = [] console_buf: deque[str] = deque(maxlen=60 if 'console_max' not in locals() else console_max) summary_buf: deque[str] = deque(maxlen=60 if 'summary_max' not in locals() else summary_max) ema_latency: Optional[float] = None # Tiny spinner text (no white box) st.markdown('

Loading AI…

', unsafe_allow_html=True) (tokenizer, processor, model, device), device_label, cuda_hint = load_engine(MODEL_ID, gpu_mode, fast_processor) # KPIs k1, k2, k3, k4, k5 = st.columns(5) k1.markdown(f'

Sampling FPS
{fps_sample}

', unsafe_allow_html=True) k2.markdown(f'

Segment (sec)
{segment_seconds}

', unsafe_allow_html=True) k3.markdown(f'

Response length
{resp_len}

', unsafe_allow_html=True) k4.markdown(f'

Device
{device_label}

', unsafe_allow_html=True) lat_slot = k5.empty() lat_slot.markdown(f'

Avg latency
--

', unsafe_allow_html=True) if cuda_hint: st.warning(cuda_hint, icon="⚠️") # Queues & worker job_q: queue.Queue = queue.Queue(maxsize=1) out_q: queue.Queue = queue.Queue() worker = InferenceWorker((tokenizer, processor, model, device), job_q, out_q, resp_len) if add_script_run_ctx: add_script_run_ctx(worker) worker.start() # Decoder ring = deque(maxlen=300) # ~10–12s @25–30fps decoder = Decoder(final_source, fps_sample, ring, st.session_state.stop_event, is_rtsp(final_source)) if add_script_run_ctx: add_script_run_ctx(decoder) decoder.start() preview_period = 1.0 / float(preview_fps) last_preview = 0.0 try: while True: if st.session_state.stop_event.is_set(): break if decoder.err: raise decoder.err now = time.time() if ring and (now - last_preview) >= preview_period: img, ts = ring[-1] img_disp = downscale(img, 512 if eff_mode else 1024) # data-URI render (media store hatalarını önler) if not safe_image_data_uri(video_img, img_disp, ts, quality=80): break last_preview = now if not safe_markdown(fps_badge, f'Sampling: {fps_sample} fps'): break if not safe_markdown(time_badge, f'Time: {ts:.2f}s'): break frames_buf.append(img_disp); times_buf.append(ts) if not safe_markdown(seg_badge, f'Segment: {len(frames_buf)}/{seg_len_frames}'): break # Dispatch a job each filled segment if len(frames_buf) >= seg_len_frames: t0, t1 = times_buf[0], times_buf[-1] instruction = build_analysis_instruction(style, t0, t1) try: if job_q.full(): _ = job_q.get_nowait() job_q.put_nowait((frames_buf.copy(), t0, t1, instruction)) except queue.Full: pass frames_buf.clear(); times_buf.clear() # Drain model outputs try: while True: t0o, t1o, txt, lat = out_q.get_nowait() console_buf.append(f"[{t0o:.1f}–{t1o:.1f}s] {txt}") summary_buf.append(f"{t0o:.1f}–{t1o:.1f}s: {txt}") if lat is not None: ema_latency = lat if ema_latency is None else (0.7*ema_latency + 0.3*lat) lat_slot.markdown(f'

Avg latency
{ema_latency:.1f}s

', unsafe_allow_html=True) except queue.Empty: pass # One-shot render if not safe_markdown(console_box, render_scrollbox(console_buf, "console_box")): break if not safe_markdown(summary_box, render_scrollbox(summary_buf, "summary_box")): break time.sleep(0.002) except Exception as e: st.error(f"Could not process source: {e}") finally: # Signal threads to stop, then join st.session_state.stop_event.set() try: finalize_and_free( decoder_obj=decoder, worker_obj=worker, buffers=[frames_buf, times_buf], deques=[ring, console_buf, summary_buf], queues=[job_q, out_q], ) except Exception: pass st.session_state.running = False try: st.success("Stopped.") except Exception: pass # Son kez bellek temizliği try: if torch.cuda.is_available(): torch.cuda.empty_cache() except Exception: pass gc.collect() else: # Idle screen (refresh/stop'tan sonra da hafif temizlik yap) try: if torch.cuda.is_available(): torch.cuda.empty_cache() except Exception: pass gc.collect() right_col = st.container() right_col.info("Pick a demo on the left or switch to Custom URL, then press Start ▶️.") st.caption("Defaults: Frame sampling=10, Preview FPS=25. Use credentials only if your custom URL is protected.")