app.py

# 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("""
<style>
:root { --bg:#0e0f12; --fg:#e8eaed; --muted:#cfd3dc; --brand:#ff7a1a; --panel:#171a1f; --border:#232730; --ghost:#2a2f3a; }
html, body, .stApp, [data-testid="stAppViewContainer"], [data-testid="stHeader"], [data-testid="stToolbar"]{
  background-color:var(--bg) !important; color:var(--fg) !important;
}
a, a:visited { color:var(--brand) !important; }
[data-testid="stSidebar"] { background:var(--bg) !important; border-right:1px solid var(--border) !important; }
/* FORCE all sidebar text visible */
section[data-testid="stSidebar"] * { color: var(--fg) !important; }
/* Inputs */
section[data-testid="stSidebar"] input[type="text"], section[data-testid="stSidebar"] input[type="password"]{
  color: var(--fg) !important; background: var(--panel) !important;
  border:1px solid var(--border) !important; border-radius:10px !important;
}
section[data-testid="stSidebar"] input::placeholder { color:#9aa3b2 !important; }
/* Select */
section[data-testid="stSidebar"] [data-baseweb="select"] > div {
  color: var(--fg) !important; background: var(--panel) !important;
  border:1px solid var(--border) !important; border-radius:10px !important;
}
/* Sliders */
section[data-testid="stSidebar"] [data-testid="stThumbValue"]{
  color: var(--bg) !important; background: var(--brand) !important; border-radius:8px !important;
}
section[data-testid="stSidebar"] .stSlider [role="slider"]{
  background: var(--brand) !important; border: 2px solid var(--brand) !important;
}
section[data-testid="stSidebar"] .stSlider [data-baseweb="slider"] > div > div{
  background: linear-gradient(to right, var(--brand), var(--brand)) !important;
}
/* Buttons: full width without deprecated flags */
section[data-testid="stSidebar"] .stButton>button {
  width: 100% !important; border-radius:12px; font-weight:700; background:var(--brand); color:var(--bg); border:0;
}
section[data-testid="stSidebar"] .stButton>button:hover { filter:brightness(0.95); }
/* Cards / KPIs */
.card { background:var(--panel); border:1px solid var(--border); border-radius:16px; padding:14px; }
.kpi  { background:var(--panel); border:1px solid var(--border); border-radius:12px; padding:14px; }
.badge{ display:inline-block; padding:2px 8px; border-radius:999px; border:1px solid var(--ghost); color:#dbe0ea; font-size:12px; }
h1,h2,h3 { color:var(--brand); }
hr.soft { border:0; border-top:1px solid var(--border); margin:12px 0 8px; }
/* Fixed-height, scrollable, latest-first */
.scrollbox {
  max-height: 320px; overflow-y: auto; padding: 8px 10px;
  background:var(--panel); border:1px solid var(--border); border-radius:12px;
  display: flex; flex-direction: column; gap: 10px;
}
.scrollline { margin:0; line-height:1.25rem; color: var(--muted); }
/* Preview image fits the card */
.video-frame img {
  width: 100% !important;
  height: auto !important;
  object-fit: contain !important;
  border-radius: 12px;
}
/* Tiny spinner (no white Streamlit status box) */
.tiny-spinner { color: var(--muted); font-size: 0.9rem; }
.sidebar-title { font-weight:700; margin-top:6px; }
</style>
""", 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('<div class="sidebar-title">Controls</div>', 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('<div class="sidebar-title">Source</div>', 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'''
        <figure style="margin:0">
          <img src="{uri}" alt="frame" style="width:100%;height:auto;object-fit:contain;border-radius:12px;" />
          <figcaption style="color:#9aa3b2;font-size:0.85rem;margin-top:4px">{cap}</figcaption>
        </figure>
        '''
        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('<div class="card">', unsafe_allow_html=True)
    # The .video-frame wrapper ensures fit-to-card visuals
    st.markdown('<div class="video-frame">', unsafe_allow_html=True)
    video_img = st.empty()
    st.markdown('</div>', unsafe_allow_html=True)

    meta_row = st.columns([1,1,1])
    fps_badge  = meta_row[0].markdown('<span class="badge">Sampling: -- fps</span>', unsafe_allow_html=True)
    time_badge = meta_row[1].markdown('<span class="badge">Time: 0.0s</span>', unsafe_allow_html=True)
    seg_badge  = meta_row[2].markdown('<span class="badge">Segment: --</span>', unsafe_allow_html=True)
    st.markdown('<hr class="soft"/>', 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('</div>', unsafe_allow_html=True)

with right:
    st.subheader("Live Console")
    console_box = st.empty()
    st.subheader("Segment Summary")
    summary_box = st.empty()
    st.markdown('<div class="card"></div>', unsafe_allow_html=True)

# ==================== Misc UI helpers ====================
def render_scrollbox(lines: deque[str], dom_id: str) -> str:
    items = "".join(f'<p class="scrollline">{l}</p>' for l in reversed(lines))
    return f'''
    <div id="{dom_id}" class="scrollbox">{items}</div>
    <script>
      const el = window.parent.document.getElementById('{dom_id}');
      if (el) el.scrollTop = 0;
    </script>
    '''

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('<div class="tiny-spinner">Loading AI…</div>', 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'<div class="kpi"><b>Sampling FPS</b><br/><big>{fps_sample}</big></div>', unsafe_allow_html=True)
    k2.markdown(f'<div class="kpi"><b>Segment (sec)</b><br/><big>{segment_seconds}</big></div>', unsafe_allow_html=True)
    k3.markdown(f'<div class="kpi"><b>Response length</b><br/><big>{resp_len}</big></div>', unsafe_allow_html=True)
    k4.markdown(f'<div class="kpi"><b>Device</b><br/><big>{device_label}</big></div>', unsafe_allow_html=True)
    lat_slot = k5.empty()
    lat_slot.markdown(f'<div class="kpi"><b>Avg latency</b><br/><big>--</big></div>', 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'<span class="badge">Sampling: {fps_sample} fps</span>'): break
                if not safe_markdown(time_badge, f'<span class="badge">Time: {ts:.2f}s</span>'): break

                frames_buf.append(img_disp); times_buf.append(ts)
                if not safe_markdown(seg_badge, f'<span class="badge">Segment: {len(frames_buf)}/{seg_len_frames}</span>'): 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"<b>[{t0o:.1f}–{t1o:.1f}s]</b> {txt}")
                    summary_buf.append(f"<b>{t0o:.1f}–{t1o:.1f}s</b>: {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'<div class="kpi"><b>Avg latency</b><br/><big>{ema_latency:.1f}s</big></div>', 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.")