workflows

stream3r/worker/main.py

@@ -83,11 +83,14 @@ def main() -> None:
     runtime = get_runtime()
 
     queues = [Queue(name, connection=runtime.redis) for name in args.queues]
+    if not queues:
+        raise ValueError("No queues configured for worker")
     for queue in queues:
         logger.info("Listening on queue '%s'", queue.name)
 
     worker = Stream3RWorker(
         queues,
+        connection=runtime.redis,
         name=settings.worker_name,
         default_timeout=settings.default_job_timeout,
     )

stream3r/worker/tasks.py

@@ -160,7 +160,13 @@ def _register_scene_media_entries(runtime: WorkerRuntime, scene_id: str, entries
         logger.exception("Failed to register scene media entries for scene %s", scene_id)
 
 
-def _resolve_frame_entry(entry: Any, *, index: int, dest_dir: Path) -> FrameRecord:
+def _resolve_frame_entry(
+    entry: Any,
+    *,
+    index: int,
+    dest_dir: Path,
+    runtime: WorkerRuntime | None = None,
+) -> FrameRecord:
     metadata: dict[str, Any] = {}
     timestamp = None
     source = None
@@ -190,6 +196,13 @@ def _resolve_frame_entry(entry: Any, *, index: int, dest_dir: Path) -> FrameReco
                 raise FileNotFoundError(f"Frame path does not exist: {path}")
             destination = dest_dir / (path.name if path.suffix else f"{frame_id}.jpg")
             shutil.copy2(path, destination)
+        elif storage_key := entry.get("storage_key") or entry.get("file") or entry.get("key"):
+            if runtime is None:
+                raise ValueError("Frame entry provided storage key but runtime is not available")
+            filename = Path(str(storage_key)).name or f"{frame_id}.jpg"
+            destination = dest_dir / filename
+            runtime.storage.download_to_path(str(storage_key), destination)
+            source = str(storage_key)
         elif url := entry.get("url"):
             source = url
             suffix = Path(url).suffix or ".jpg"
@@ -198,10 +211,9 @@ def _resolve_frame_entry(entry: Any, *, index: int, dest_dir: Path) -> FrameReco
         elif content := entry.get("content"):
             destination = dest_dir / f"{frame_id}.jpg"
             _write_base64(content, destination)
-        else:
-            raise ValueError("Frame entry must include 'path', 'url', or 'content'")
     else:
-        raise TypeError("Unsupported frame specification")
+        raise ValueError("Frame entry must include 'path', 'url', or 'content'")
+
 
     if destination.suffix.lower() not in IMAGE_EXTENSIONS:
         destination = destination.with_suffix(".png")
@@ -231,7 +243,9 @@ def _collect_frames(
         for entry in frames_payload:
             if frame_limit and frame_limit > 0 and len(records) >= frame_limit:
                 break
-            records.append(_resolve_frame_entry(entry, index=len(records), dest_dir=frames_dir))
+            records.append(
+                _resolve_frame_entry(entry, index=len(records), dest_dir=frames_dir, runtime=runtime)
+            )
     else:
         directory = payload.get("frames_dir") or payload.get("images_dir")
         if directory: