diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 2a4baf8..05a6e2e 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -17,7 +17,7 @@ from .worker import Worker # Bump on every agent change. The page embeds this and /status reports it; the UI # warns to reload when they differ — so a stale browser-cached page can't be # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.) -VERSION = "2026-07-01.5 · name the real fetch-failure reason" +VERSION = "2026-07-01.6 · stream videos via ffmpeg-from-URL (no full download)" logbuf.install() cfg = Config.from_env() diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py index 1ac9925..6138a00 100644 --- a/agent/fc_agent/client.py +++ b/agent/fc_agent/client.py @@ -121,6 +121,16 @@ class FcClient: r.raise_for_status() return r.content + def is_reachable(self) -> bool: + """Cheap 'is curator responding at all right now?' check. Used to decide, + when a video can't be sampled, between a transient outage (keep retrying — + survives a redeploy) and an unprocessable file (fail it, don't loop).""" + try: + r = self.s.get(f"{self.base}/api/gpu/status", timeout=5) + return r.status_code < 500 + except requests.RequestException: + return False + def queue_status(self) -> dict: # Short timeout: this backs the UI /status poll, so a busy curator must # not hang the page for long (the GPU meters poll /gpu separately). diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py index 027cdc1..cad5b34 100644 --- a/agent/fc_agent/config.py +++ b/agent/fc_agent/config.py @@ -41,6 +41,8 @@ class Config: frame_dedupe_distance: int # video frames whose dHash differs by < this many # bits are near-duplicates, dropped before detect; # higher keeps more frames, 0 disables + ffmpeg_timeout: float # hard ceiling (s) for ffmpeg-from-URL video sampling; + # generous so a SLOW media link still completes @classmethod def from_env(cls) -> Config: @@ -69,4 +71,5 @@ class Config: max_regions=int(os.environ.get("MAX_REGIONS", "128")), dedupe_iou=float(os.environ.get("DEDUPE_IOU", "0.85")), frame_dedupe_distance=int(os.environ.get("FRAME_DEDUPE_DISTANCE", "8")), + ffmpeg_timeout=float(os.environ.get("FFMPEG_TIMEOUT", "1200")), ) diff --git a/agent/fc_agent/media.py b/agent/fc_agent/media.py index 435edfa..4af1f1e 100644 --- a/agent/fc_agent/media.py +++ b/agent/fc_agent/media.py @@ -85,32 +85,48 @@ def load_image(data: bytes) -> Image.Image: return to_rgb(Image.open(io.BytesIO(data))) -def sample_frames( - data: bytes, interval_seconds: float, max_frames: int +# ffmpeg reconnect flags — resume a dropped HTTP transfer (a slow/contended media +# store can cut a long stream) instead of failing the whole job. Relies only on +# HTTP + Range, which every FC deployment serves → environment-agnostic. +_RECONNECT = [ + "-reconnect", "1", "-reconnect_streamed", "1", + "-reconnect_on_network_error", "1", "-reconnect_delay_max", "5", +] + + +def _collect_frames( + tmp: str, interval: float, cap: int ) -> list[tuple[float, Image.Image]]: - """Extract up to max_frames frames at one-every-interval_seconds via ffmpeg. - Returns [(timestamp_seconds, frame)]. Empty on failure (caller falls back).""" + out: list[tuple[float, Image.Image]] = [] + names = sorted(n for n in os.listdir(tmp) if n.startswith("f_")) + for i, name in enumerate(names[:cap]): + with Image.open(os.path.join(tmp, name)) as im: + out.append((round(i * interval, 2), to_rgb(im))) + return out + + +def sample_frames_from_url( + url: str, interval_seconds: float, max_frames: int, + *, headers: str = "", timeout: float = 1200.0, +) -> list[tuple[float, Image.Image]]: + """Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads + only the video index + up to max_frames worth of content, so the agent never + downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in + RAM and get cut off mid-download). Reconnect flags resume a dropped transfer; + the generous timeout lets it finish however slow the link is — completion over + speed, and environment-agnostic (any HTTP+Range source). Empty on failure.""" interval = max(0.5, float(interval_seconds or 4.0)) cap = max(1, int(max_frames or 64)) + hdr = ["-headers", headers] if headers else [] with tempfile.TemporaryDirectory() as tmp: - src = os.path.join(tmp, "in") - with open(src, "wb") as fh: - fh.write(data) pattern = os.path.join(tmp, "f_%05d.jpg") try: subprocess.run( - [ - "ffmpeg", "-nostdin", "-loglevel", "error", "-i", src, - "-vf", f"fps=1/{interval}", "-frames:v", str(cap), - "-q:v", "3", pattern, - ], - check=True, timeout=600, + ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr, + "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap), + "-q:v", "3", pattern], + check=True, timeout=timeout, ) except (subprocess.SubprocessError, FileNotFoundError): return [] - out: list[tuple[float, Image.Image]] = [] - names = sorted(n for n in os.listdir(tmp) if n.startswith("f_")) - for i, name in enumerate(names[:cap]): - with Image.open(os.path.join(tmp, name)) as im: - out.append((round(i * interval, 2), to_rgb(im))) - return out + return _collect_frames(tmp, interval, cap) diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 7ccd7b3..b6e5a30 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -132,6 +132,11 @@ class Worker: def __init__(self, cfg: Config): self.cfg = cfg self.client = FcClient(cfg.fc_url, cfg.token, cfg.agent_id) + # Sent to ffmpeg-from-URL so video sampling works whether or not a + # deployment gates /images behind the bearer token (FC's is public). + self._auth_header = ( + f"Authorization: Bearer {cfg.token}\r\n" if cfg.token else "" + ) self._lock = threading.Lock() self._running = False self._auto = bool(cfg.auto_scale) # autoscale the downloader count @@ -480,26 +485,42 @@ class Worker: def _download_decode(self, job: dict): """Fetch the image bytes and decode → [(frame_time, PIL.Image)]. Videos are sampled into frames (ffmpeg). Records the download + decode timings.""" - _t = time.monotonic() - data = self.client.fetch_image(job["image_url"]) - self._record("download", time.monotonic() - _t) - _t = time.monotonic() if media.is_video(job.get("mime", "")): - frames = media.sample_frames( - data, job.get("frame_interval_seconds", 4.0), + # Stream the video: ffmpeg reads the media URL directly and Range-reads + # only the frames it needs, so we NEVER pull the whole file (VR/4K + # originals are 800MB+ — buffering that in RAM and getting cut off + # mid-download was the failure loop). Environment-agnostic + resilient. + _t = time.monotonic() + url = f"{self.cfg.fc_url}{job['image_url']}" + frames = media.sample_frames_from_url( + url, job.get("frame_interval_seconds", 4.0), job.get("max_frames", 64), - ) or [(None, media.load_image(data))] - # Temporal dedup: a near-static video sampled into many frames re-runs - # the whole detect+embed chain on ~identical frames. Drop near-dup - # frames HERE (decode stage, CPU) so the GPU never sees them. + headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout, + ) + self._record("decode", time.monotonic() - _t) + if not frames: + # Couldn't sample. If curator is up, the file is unprocessable → + # a job fault (fail it, don't re-lease forever). If curator is + # unreachable, it's transient → let the loop back off + retry + # (survives a redeploy). ConnectionError is caught as transient. + if self.client.is_reachable(): + raise RuntimeError("no frames sampled from video (unprocessable)") + raise requests.ConnectionError("curator unreachable during video sampling") + # Temporal dedup: a near-static video re-runs the whole detect+embed + # chain on ~identical frames — drop near-dups HERE (CPU) pre-GPU. if len(frames) > 1 and self.cfg.frame_dedupe_distance > 0: kept = media.dedupe_frames(frames, self.cfg.frame_dedupe_distance) if len(kept) < len(frames): log.info("job %s: video frames %d→%d (near-dup dedup)", job.get("job_id"), len(frames), len(kept)) frames = kept - else: - frames = [(None, media.load_image(data))] + return frames + # Stills: download the bytes and decode. + _t = time.monotonic() + data = self.client.fetch_image(job["image_url"]) + self._record("download", time.monotonic() - _t) + _t = time.monotonic() + frames = [(None, media.load_image(data))] self._record("decode", time.monotonic() - _t) return frames