[bugfix] Accept log_queue in StreamingVideoGenerator.from_fastvideo_args (#1937)

VideoGenerator.from_config calls from_fastvideo_args with a log_queue keyword
argument. The StreamingVideoGenerator override did not accept it, so
StreamingVideoGenerator.from_pretrained raised TypeError. The override now
accepts log_queue and passes it to the executor through VideoGenerator.__init__.
A new test covers the call.

fastvideo/entrypoints/streaming_generator.py

@@ -82,13 +82,14 @@ class StreamingVideoGenerator(VideoGenerator):
                  fastvideo_args: FastVideoArgs,
                  executor_class: type[Executor],
                  log_stats: bool,
-                 use_queue_mode: bool = True):
+                 use_queue_mode: bool = True,
+                 log_queue=None):
         if use_queue_mode and issubclass(executor_class, MultiprocExecutor):
             # Copy so callers that reuse these args for a standard VideoGenerator
             # do not inherit the streaming queues; RPC mode never needs them.
             fastvideo_args = copy.copy(fastvideo_args)
             fastvideo_args.enable_streaming_ipc_queues = True
-        super().__init__(fastvideo_args, executor_class, log_stats)
+        super().__init__(fastvideo_args, executor_class, log_stats, log_queue=log_queue)
         self.accumulated_frames: list[np.ndarray] = []
         self.sampling_param: SamplingParam | None = None
         self.batch: ForwardBatch | None = None
@@ -104,7 +105,7 @@ class StreamingVideoGenerator(VideoGenerator):
         self.block_idx: int = 0
 
     @classmethod
-    def from_fastvideo_args(cls, fastvideo_args: FastVideoArgs) -> "StreamingVideoGenerator":
+    def from_fastvideo_args(cls, fastvideo_args: FastVideoArgs, *, log_queue=None) -> "StreamingVideoGenerator":
         reject_external_launcher(
             fastvideo_args.distributed_executor_backend,
             entrypoint="StreamingVideoGenerator",
@@ -114,6 +115,7 @@ class StreamingVideoGenerator(VideoGenerator):
             fastvideo_args=fastvideo_args,
             executor_class=executor_class,
             log_stats=False,
+            log_queue=log_queue,
         )
 
     def reset(

fastvideo/tests/worker/test_multiproc_executor.py

@@ -445,3 +445,26 @@ def test_streaming_generator_enables_queues_only_for_queue_mode(monkeypatch, use
 
     assert seen[0].enable_streaming_ipc_queues is expected
     assert args.enable_streaming_ipc_queues is False  # the caller's args are never mutated
+
+
+def test_streaming_generator_from_fastvideo_args_forwards_log_queue(monkeypatch) -> None:
+    from fastvideo.entrypoints.streaming_generator import StreamingVideoGenerator
+    from fastvideo.entrypoints.video_generator import VideoGenerator
+    from fastvideo.worker.executor import Executor
+
+    seen: list[tuple[FastVideoArgs, object]] = []
+
+    def fake_init(self, fastvideo_args, executor_class, log_stats, **kwargs):
+        seen.append((fastvideo_args, kwargs.get("log_queue")))
+        self.executor = MultiprocExecutor.__new__(MultiprocExecutor)
+
+    monkeypatch.setattr(VideoGenerator, "__init__", fake_init)
+    monkeypatch.setattr(Executor, "get_class", staticmethod(lambda fastvideo_args: MultiprocExecutor))
+    args = FastVideoArgs(model_path="test/model")
+    sentinel = object()
+
+    StreamingVideoGenerator.from_fastvideo_args(args, log_queue=sentinel)
+
+    assert seen[0][1] is sentinel  # VideoGenerator.from_config's log_queue= call site works again
+    assert seen[0][0] is not args and seen[0][0].enable_streaming_ipc_queues is True
+    assert args.enable_streaming_ipc_queues is False