From 2b164405c3d15ed2d222d0582d6db09ffc5ee6b2 Mon Sep 17 00:00:00 2001 From: William Lin Date: Wed, 7 Oct 2026 15:58:03 -0700 Subject: [PATCH] [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 | 8 ++++--- .../tests/worker/test_multiproc_executor.py | 23 +++++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/fastvideo/entrypoints/streaming_generator.py b/fastvideo/entrypoints/streaming_generator.py index 258b7d580..8a7913057 100644 --- a/fastvideo/entrypoints/streaming_generator.py +++ b/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( diff --git a/fastvideo/tests/worker/test_multiproc_executor.py b/fastvideo/tests/worker/test_multiproc_executor.py index 58320e72c..a7dcacf97 100644 --- a/fastvideo/tests/worker/test_multiproc_executor.py +++ b/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