961 lines
34 KiB
Python
961 lines
34 KiB
Python
import os
|
|
from types import SimpleNamespace
|
|
import warnings
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import torch
|
|
from einops import rearrange
|
|
|
|
import fastvideo.entrypoints.video_generator as video_generator_module
|
|
from fastvideo.api import (
|
|
ConfigValidationError,
|
|
GenerationRequest,
|
|
GenerationResult,
|
|
GeneratorConfig,
|
|
InputConfig,
|
|
SamplingConfig,
|
|
load_run_config,
|
|
)
|
|
from fastvideo.api.sampling_param import SamplingParam
|
|
from fastvideo.entrypoints.video_generator import VideoGenerator, _resolve_output_size
|
|
from fastvideo.fastvideo_args import WorkloadType
|
|
from fastvideo.pipelines import ForwardBatch
|
|
from fastvideo.worker.gpu_worker import Worker
|
|
from fastvideo.worker.ray_distributed_executor import RayDistributedExecutor
|
|
|
|
|
|
def _new_video_generator() -> VideoGenerator:
|
|
# Bypass __init__ since we only test a pure helper method.
|
|
return VideoGenerator.__new__(VideoGenerator)
|
|
|
|
|
|
def _new_runtime_video_generator() -> VideoGenerator:
|
|
generator = _new_video_generator()
|
|
generator.fastvideo_args = SimpleNamespace(
|
|
model_path="test-model",
|
|
prompt_txt=None,
|
|
workload_type=SimpleNamespace(value="t2v"),
|
|
)
|
|
generator.executor = SimpleNamespace(
|
|
set_log_queue=lambda queue: None,
|
|
clear_log_queue=lambda: None,
|
|
)
|
|
generator.config = None
|
|
return generator
|
|
|
|
|
|
def _patch_from_fastvideo_args(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_from_fastvideo_args(cls, fastvideo_args, *, log_queue=None):
|
|
generator = cls.__new__(cls)
|
|
generator.fastvideo_args = fastvideo_args
|
|
generator.executor = None
|
|
generator.config = None
|
|
captured["fastvideo_args"] = fastvideo_args
|
|
captured["log_queue"] = log_queue
|
|
return generator
|
|
|
|
monkeypatch.setattr(
|
|
VideoGenerator,
|
|
"from_fastvideo_args",
|
|
classmethod(fake_from_fastvideo_args),
|
|
)
|
|
return captured
|
|
|
|
|
|
def _patch_fastvideo_args_from_kwargs(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_from_kwargs(cls, **kwargs):
|
|
captured["kwargs"] = kwargs
|
|
return SimpleNamespace(
|
|
model_path=kwargs["model_path"],
|
|
num_gpus=kwargs["num_gpus"],
|
|
workload_type=WorkloadType.from_string(kwargs.get("workload_type", "t2v")),
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"fastvideo.api.compat.FastVideoArgs.from_kwargs",
|
|
classmethod(fake_from_kwargs),
|
|
)
|
|
return captured
|
|
|
|
|
|
def _patch_sampling_param_from_pretrained(monkeypatch):
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls()
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
|
|
|
|
class _NoCpuMaterializationOutput:
|
|
|
|
@property
|
|
def shape(self):
|
|
raise AssertionError("metadata-only output path should not inspect output shape")
|
|
|
|
def cpu(self):
|
|
raise AssertionError("metadata-only output path should not move output to CPU")
|
|
|
|
def __mul__(self, other):
|
|
raise AssertionError("metadata-only output path should not build frames")
|
|
|
|
|
|
def _single_video_args(output_type="pil"):
|
|
return SimpleNamespace(
|
|
output_type=output_type,
|
|
pin_cpu_memory=False,
|
|
VSA_sparsity=0.0,
|
|
pipeline_config=SimpleNamespace(flow_shift=1.0, embedded_cfg_scale=1.0),
|
|
workload_type=SimpleNamespace(value="t2v"),
|
|
)
|
|
|
|
|
|
def _small_sampling_param(*, save_video=False, return_frames=False):
|
|
return SamplingParam(
|
|
num_frames=2,
|
|
height=16,
|
|
width=16,
|
|
fps=8,
|
|
num_inference_steps=1,
|
|
save_video=save_video,
|
|
return_frames=return_frames,
|
|
)
|
|
|
|
|
|
def _single_video_output_batch(output, *, extra=None):
|
|
return SimpleNamespace(
|
|
output=output,
|
|
extra=extra or {},
|
|
logging_info=None,
|
|
trajectory_latents=None,
|
|
trajectory_timesteps=None,
|
|
trajectory_decoded=None,
|
|
)
|
|
|
|
|
|
def _single_video_generator(output_batch, fastvideo_args):
|
|
generator = _new_video_generator()
|
|
generator.fastvideo_args = fastvideo_args
|
|
generator.executor = SimpleNamespace(execute_forward=lambda batch, args: output_batch)
|
|
generator.config = None
|
|
return generator
|
|
|
|
|
|
def test_resolve_output_size_uses_produced_pixel_geometry() -> None:
|
|
"""Report refined dimensions instead of the base request geometry."""
|
|
samples = torch.empty(1, 3, 5, 64, 96)
|
|
assert _resolve_output_size(samples, (32, 48, 1), pixel_output=True) == (64, 96, 5)
|
|
assert _resolve_output_size(samples, (32, 48, 1), pixel_output=False) == (32, 48, 1)
|
|
|
|
|
|
def test_prepare_output_path_file_sanitization(tmp_path):
|
|
vg = _new_video_generator()
|
|
target_dir = tmp_path / "dir"
|
|
raw_path = target_dir / "inv:al*id?.mp4"
|
|
|
|
result = vg._prepare_output_path(str(raw_path), prompt="ignored")
|
|
|
|
assert os.path.dirname(result) == str(target_dir)
|
|
assert os.path.basename(result) == "invalid.mp4"
|
|
assert os.path.isdir(target_dir)
|
|
|
|
|
|
def test_prepare_output_path_directory_prompt_derived(tmp_path):
|
|
vg = _new_video_generator()
|
|
out_dir = tmp_path / "outputs"
|
|
prompt = "Hello:/\\*?<>| world"
|
|
|
|
result = vg._prepare_output_path(str(out_dir), prompt=prompt)
|
|
|
|
assert os.path.dirname(result) == str(out_dir)
|
|
# spaces are preserved (collapsed) by sanitizer; here it becomes "Hello world.mp4"
|
|
assert os.path.basename(result) == "Hello world.mp4"
|
|
assert os.path.isdir(out_dir)
|
|
|
|
|
|
def test_prepare_output_path_non_mp4_treated_as_dir(tmp_path):
|
|
vg = _new_video_generator()
|
|
weird_dir = tmp_path / "foo.gif"
|
|
prompt = "My Video"
|
|
|
|
result = vg._prepare_output_path(str(weird_dir), prompt=prompt)
|
|
|
|
assert os.path.dirname(result) == str(weird_dir)
|
|
assert os.path.basename(result) == "My Video.mp4"
|
|
assert os.path.isdir(weird_dir)
|
|
|
|
|
|
def test_prepare_output_path_uniqueness_suffix(tmp_path):
|
|
vg = _new_video_generator()
|
|
out_dir = tmp_path / "outputs"
|
|
prompt = "Sample Name"
|
|
|
|
first = vg._prepare_output_path(str(out_dir), prompt=prompt)
|
|
# simulate existing file
|
|
os.makedirs(os.path.dirname(first), exist_ok=True)
|
|
with open(first, "wb") as f:
|
|
f.write(b"")
|
|
|
|
second = vg._prepare_output_path(str(out_dir), prompt=prompt)
|
|
assert os.path.basename(second) == "Sample Name_1.mp4"
|
|
|
|
# simulate second existing file as well
|
|
with open(second, "wb") as f:
|
|
f.write(b"")
|
|
third = vg._prepare_output_path(str(out_dir), prompt=prompt)
|
|
assert os.path.basename(third) == "Sample Name_2.mp4"
|
|
|
|
|
|
def test_prepare_output_path_empty_prompt_fallback(tmp_path):
|
|
vg = _new_video_generator()
|
|
out_dir = tmp_path / "outputs"
|
|
bad_prompt = ":/\\*?<>| .." # sanitizes to empty, should fallback to "video"
|
|
|
|
result = vg._prepare_output_path(str(out_dir), prompt=bad_prompt)
|
|
|
|
assert os.path.dirname(result) == str(out_dir)
|
|
assert os.path.basename(result) == "output.mp4"
|
|
|
|
|
|
def test_generate_single_video_metadata_only_skips_output_materialization(monkeypatch, tmp_path):
|
|
output_batch = _single_video_output_batch(
|
|
_NoCpuMaterializationOutput(),
|
|
extra={"peak_memory_mb": 42.0},
|
|
)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
empty_calls = []
|
|
real_empty = video_generator_module.torch.empty
|
|
|
|
def record_empty(*args, **kwargs):
|
|
empty_calls.append((args, kwargs))
|
|
return real_empty(*args, **kwargs)
|
|
|
|
def fail_make_grid(*args, **kwargs):
|
|
raise AssertionError("metadata-only output path should not build frame grids")
|
|
|
|
monkeypatch.setattr(video_generator_module.torch, "empty", record_empty)
|
|
monkeypatch.setattr(video_generator_module.torchvision.utils, "make_grid", fail_make_grid)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="metadata only",
|
|
sampling_param=_small_sampling_param(save_video=False, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["video_path"] is None
|
|
assert result["peak_memory_mb"] == 42.0
|
|
assert empty_calls == [((0, ), {"device": "cpu"})]
|
|
|
|
|
|
def test_generate_single_video_return_frames_still_materializes_output(tmp_path):
|
|
output = torch.ones((1, 3, 2, 16, 16), dtype=torch.float32) * 0.5
|
|
output_batch = _single_video_output_batch(output)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="return frames",
|
|
sampling_param=_small_sampling_param(save_video=False, return_frames=True),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
torch.testing.assert_close(result["samples"], output)
|
|
assert len(result["frames"]) == 2
|
|
assert result["video_path"] is None
|
|
|
|
|
|
def test_generate_single_video_frames_match_legacy_cpu_loop(tmp_path):
|
|
"""The on-device quantize path (#1362) must reproduce the legacy
|
|
per-frame CPU loop (make_grid -> permute -> *255 -> uint8) bit-exactly
|
|
for in-range fp32 pixels: same uint8 dtype, same HWC grid layout with
|
|
nrow=6 (batch>1), odd frame count. CPU-only: on CUDA the float->uint8
|
|
cast may differ by <=1 LSB, but on CPU both orderings run identical
|
|
fp32 ops, so exact equality is required."""
|
|
torch.manual_seed(0)
|
|
output = torch.rand((2, 3, 3, 16, 16), dtype=torch.float32)
|
|
output_batch = _single_video_output_batch(output)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
sampling_param = _small_sampling_param(save_video=False, return_frames=True)
|
|
sampling_param.num_frames = 3
|
|
sampling_param.num_videos_per_prompt = 2
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="grid parity",
|
|
sampling_param=sampling_param,
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
legacy_frames = []
|
|
for x in rearrange(output, "b c t h w -> t b c h w"):
|
|
grid = video_generator_module.torchvision.utils.make_grid(x, nrow=6)
|
|
grid = grid.permute(1, 2, 0).squeeze(-1)
|
|
legacy_frames.append((grid * 255).to(torch.uint8).contiguous().cpu().numpy())
|
|
|
|
torch.testing.assert_close(result["samples"], output)
|
|
assert len(result["frames"]) == 3
|
|
for got, want in zip(result["frames"], legacy_frames, strict=True):
|
|
assert got.dtype == np.uint8
|
|
assert got.shape == want.shape
|
|
np.testing.assert_array_equal(got, want)
|
|
|
|
|
|
def test_generate_single_video_frames_clamp_out_of_range_pixels(tmp_path):
|
|
"""VAE output slightly outside [0, 1] must saturate at 0/255 in the
|
|
uint8 frames. The pre-#1362 unclamped cast wrapped mod 256 (e.g.
|
|
1.5 -> 126). CPU-only."""
|
|
output = torch.full((1, 3, 2, 16, 16), 1.5, dtype=torch.float32)
|
|
output[:, :, 1] = -0.5
|
|
output_batch = _single_video_output_batch(output)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="clamp",
|
|
sampling_param=_small_sampling_param(save_video=False, return_frames=True),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
frames = result["frames"]
|
|
assert len(frames) == 2
|
|
# make_grid passes a single image through without grid padding, so
|
|
# every pixel comes from the (clamped) output tensor.
|
|
assert frames[0].dtype == np.uint8
|
|
assert frames[0].shape == (16, 16, 3)
|
|
assert (frames[0] == 255).all()
|
|
assert (frames[1] == 0).all()
|
|
|
|
|
|
def test_generate_single_video_save_video_still_builds_frames(monkeypatch, tmp_path):
|
|
output = torch.ones((1, 3, 2, 16, 16), dtype=torch.float32) * 0.5
|
|
output_batch = _single_video_output_batch(output)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
saved = {}
|
|
|
|
def fake_mimsave(path, frames, *, fps, format):
|
|
saved["path"] = path
|
|
saved["frame_count"] = len(frames)
|
|
saved["fps"] = fps
|
|
saved["format"] = format
|
|
|
|
monkeypatch.setattr(video_generator_module.imageio, "mimsave", fake_mimsave)
|
|
|
|
output_path = str(tmp_path / "saved.mp4")
|
|
result = generator._generate_single_video(
|
|
prompt="save only",
|
|
sampling_param=_small_sampling_param(save_video=True, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=output_path,
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["video_path"] == output_path
|
|
assert saved == {
|
|
"path": output_path,
|
|
"frame_count": 2,
|
|
"fps": 8,
|
|
"format": "mp4",
|
|
}
|
|
|
|
|
|
def test_generate_single_video_save_only_reports_refined_output_size(monkeypatch, tmp_path):
|
|
"""`GenerationResult.size` must describe the decoded media even when the
|
|
fp32 `samples` mirror is skipped (`return_frames=False`, the CLI save
|
|
flow). Refiner pipelines can change the final pixel geometry, so the size
|
|
has to come from `output_batch.output`, not the base request. CPU-only."""
|
|
# Refiner-style output: request asks for 2 frames of 16x16, pipeline
|
|
# produces 5 frames of 32x48.
|
|
output = torch.full((1, 3, 5, 32, 48), 0.5, dtype=torch.float32)
|
|
output_batch = _single_video_output_batch(output)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
saved = {}
|
|
|
|
def fake_mimsave(path, frames, *, fps, format):
|
|
saved["frame_count"] = len(frames)
|
|
|
|
monkeypatch.setattr(video_generator_module.imageio, "mimsave", fake_mimsave)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="refined save",
|
|
sampling_param=_small_sampling_param(save_video=True, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "refined.mp4"),
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["size"] == (32, 48, 5)
|
|
assert saved["frame_count"] == 5
|
|
|
|
|
|
def test_generate_single_video_audio_only_metadata_returns_audio_without_frames(tmp_path):
|
|
audio = torch.zeros((16, ), dtype=torch.float32)
|
|
output_batch = _single_video_output_batch(
|
|
_NoCpuMaterializationOutput(),
|
|
extra={
|
|
"audio_only": True,
|
|
"audio": audio,
|
|
"audio_sample_rate": 44100,
|
|
},
|
|
)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="audio only",
|
|
sampling_param=_small_sampling_param(save_video=False, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["audio"] is audio
|
|
assert result["audio_sample_rate"] == 44100
|
|
|
|
|
|
def test_generate_single_video_audio_only_save_skips_placeholder_materialization(tmp_path):
|
|
audio = torch.zeros((16, ), dtype=torch.float32)
|
|
output_batch = _single_video_output_batch(
|
|
_NoCpuMaterializationOutput(),
|
|
extra={
|
|
"audio_only": True,
|
|
"audio": audio,
|
|
"audio_sample_rate": 44100,
|
|
},
|
|
)
|
|
fastvideo_args = _single_video_args()
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
output_path = str(tmp_path / "audio.mp4")
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="audio only",
|
|
sampling_param=_small_sampling_param(save_video=True, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=output_path,
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["audio"] is audio
|
|
assert result["audio_sample_rate"] == 44100
|
|
assert result["video_path"] == str(tmp_path / "audio.wav")
|
|
|
|
|
|
def test_generate_single_video_ray_audio_only_save_preserves_worker_metadata(monkeypatch, tmp_path):
|
|
audio = torch.zeros((16, ), dtype=torch.float32)
|
|
worker_output = ForwardBatch(
|
|
data_type="audio",
|
|
output=torch.ones((1, 3, 1, 8, 8)),
|
|
extra={
|
|
"audio_only": True,
|
|
"audio": audio,
|
|
"audio_sample_rate": 44100,
|
|
},
|
|
)
|
|
worker = Worker.__new__(Worker)
|
|
worker.fastvideo_args = SimpleNamespace()
|
|
worker.pipeline = SimpleNamespace(forward=lambda batch, args: worker_output)
|
|
|
|
monkeypatch.setattr(RayDistributedExecutor, "__abstractmethods__", frozenset())
|
|
executor = RayDistributedExecutor.__new__(RayDistributedExecutor)
|
|
executor.shutdown = lambda: None
|
|
|
|
def collective_rpc(method, *, kwargs, **_):
|
|
assert method == "execute_forward"
|
|
return [worker.execute_forward(**kwargs)]
|
|
|
|
executor.collective_rpc = collective_rpc
|
|
fastvideo_args = _single_video_args()
|
|
generator = _new_video_generator()
|
|
generator.fastvideo_args = fastvideo_args
|
|
generator.executor = executor
|
|
generator.config = None
|
|
written = {}
|
|
|
|
def fake_write_pcm_wav(path, samples, sample_rate):
|
|
written.update(path=path, samples=samples, sample_rate=sample_rate)
|
|
|
|
monkeypatch.setattr(generator, "_write_pcm_wav", fake_write_pcm_wav)
|
|
sampling_param = _small_sampling_param(save_video=True, return_frames=False)
|
|
sampling_param.data_type = "audio"
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="audio only",
|
|
sampling_param=sampling_param,
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "audio.mp4"),
|
|
)
|
|
|
|
assert worker_output.output is not None
|
|
assert worker_output.output.numel() == 0
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["audio"] is audio
|
|
assert result["audio_sample_rate"] == 44100
|
|
assert result["video_path"] == str(tmp_path / "audio.wav")
|
|
assert written["path"] == str(tmp_path / "audio.wav")
|
|
assert written["samples"] is audio
|
|
assert written["sample_rate"] == 44100
|
|
|
|
|
|
def test_generate_single_video_latent_metadata_skips_cpu_materialization(tmp_path):
|
|
output_batch = _single_video_output_batch(_NoCpuMaterializationOutput())
|
|
fastvideo_args = _single_video_args(output_type="latent")
|
|
generator = _single_video_generator(output_batch, fastvideo_args)
|
|
|
|
result = generator._generate_single_video(
|
|
prompt="latent metadata",
|
|
sampling_param=_small_sampling_param(save_video=False, return_frames=False),
|
|
fastvideo_args=fastvideo_args,
|
|
output_path=str(tmp_path / "unused.mp4"),
|
|
)
|
|
|
|
assert result["samples"] is None
|
|
assert result["frames"] is None
|
|
assert result["video_path"] is None
|
|
|
|
|
|
def test_from_config_normalizes_and_translates(monkeypatch):
|
|
captured = _patch_from_fastvideo_args(monkeypatch)
|
|
_patch_fastvideo_args_from_kwargs(monkeypatch)
|
|
config = GeneratorConfig(model_path="test-model")
|
|
config.engine.num_gpus = 2
|
|
config.pipeline.workload_type = "t2v"
|
|
|
|
generator = VideoGenerator.from_config(config)
|
|
|
|
assert captured["fastvideo_args"].model_path == "test-model"
|
|
assert captured["fastvideo_args"].num_gpus == 2
|
|
assert captured["fastvideo_args"].workload_type.value == "t2v"
|
|
assert generator.config == config
|
|
|
|
|
|
def test_from_file_loads_generator_from_run_config(tmp_path, monkeypatch):
|
|
captured = _patch_from_fastvideo_args(monkeypatch)
|
|
_patch_fastvideo_args_from_kwargs(monkeypatch)
|
|
config_path = tmp_path / "run.yaml"
|
|
config_path.write_text(
|
|
"generator:\n"
|
|
" model_path: test-model\n"
|
|
" engine:\n"
|
|
" num_gpus: 3\n"
|
|
"request:\n"
|
|
" prompt: hello\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
VideoGenerator.from_file(str(config_path))
|
|
|
|
assert captured["fastvideo_args"].model_path == "test-model"
|
|
assert captured["fastvideo_args"].num_gpus == 3
|
|
|
|
|
|
def test_from_pretrained_convenience_kwargs_do_not_warn(monkeypatch):
|
|
captured = _patch_from_fastvideo_args(monkeypatch)
|
|
fastvideo_args_capture = _patch_fastvideo_args_from_kwargs(monkeypatch)
|
|
|
|
with warnings.catch_warnings(record=True) as caught:
|
|
warnings.simplefilter("always")
|
|
generator = VideoGenerator.from_pretrained(
|
|
"test-model",
|
|
num_gpus=4,
|
|
use_fsdp_inference=False,
|
|
text_encoder_cpu_offload=True,
|
|
pin_cpu_memory=True,
|
|
dit_cpu_offload=False,
|
|
vae_cpu_offload=False,
|
|
)
|
|
|
|
assert not caught
|
|
assert captured["fastvideo_args"].model_path == "test-model"
|
|
assert captured["fastvideo_args"].num_gpus == 4
|
|
assert fastvideo_args_capture["kwargs"]["use_fsdp_inference"] is False
|
|
assert fastvideo_args_capture["kwargs"]["text_encoder_cpu_offload"] is True
|
|
assert fastvideo_args_capture["kwargs"]["pin_cpu_memory"] is True
|
|
assert fastvideo_args_capture["kwargs"]["dit_cpu_offload"] is False
|
|
assert fastvideo_args_capture["kwargs"]["vae_cpu_offload"] is False
|
|
assert generator.config is not None
|
|
assert generator.config.model_path == "test-model"
|
|
assert generator.config.engine.num_gpus == 4
|
|
|
|
|
|
def test_from_pretrained_legacy_only_kwargs_warn(monkeypatch):
|
|
captured = _patch_from_fastvideo_args(monkeypatch)
|
|
_patch_fastvideo_args_from_kwargs(monkeypatch)
|
|
|
|
with pytest.warns(DeprecationWarning, match="legacy-only kwargs"):
|
|
generator = VideoGenerator.from_pretrained(
|
|
"test-model",
|
|
num_gpus=4,
|
|
workload_type="t2v",
|
|
)
|
|
|
|
assert captured["fastvideo_args"].model_path == "test-model"
|
|
assert captured["fastvideo_args"].num_gpus == 4
|
|
assert captured["fastvideo_args"].workload_type.value == "t2v"
|
|
assert generator.config is not None
|
|
assert generator.config.pipeline.workload_type == "t2v"
|
|
|
|
|
|
def test_generate_uses_typed_request_path(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
_patch_sampling_param_from_pretrained(monkeypatch)
|
|
captured = {}
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["prompt"] = prompt
|
|
captured["sampling_param"] = sampling_param
|
|
captured["kwargs"] = kwargs
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
result = generator.generate(
|
|
GenerationRequest(
|
|
prompt="hello world",
|
|
sampling=SamplingConfig(num_frames=81, height=480, width=832),
|
|
))
|
|
|
|
assert isinstance(result, GenerationResult)
|
|
assert captured["prompt"] == "hello world"
|
|
assert captured["sampling_param"].num_frames == 81
|
|
assert captured["sampling_param"].height == 480
|
|
assert captured["sampling_param"].width == 832
|
|
assert result.video_path == "outputs/test.mp4"
|
|
|
|
|
|
def test_generate_rejects_stage_override_outside_registered_stage(monkeypatch) -> None:
|
|
"""Validate typed stage overrides at the public generation entrypoint."""
|
|
generator = _new_runtime_video_generator()
|
|
monkeypatch.setattr(
|
|
"fastvideo.registry.get_preset_selection",
|
|
lambda _model_path: ("lingbot_video_moe_refiner_t2v", "lingbot_video"),
|
|
)
|
|
with pytest.raises(ConfigValidationError, match="stage_overrides.denoise.t_thresh"):
|
|
generator.generate({
|
|
"prompt": "hello world",
|
|
"stage_overrides": {
|
|
"denoise": {
|
|
"t_thresh": 0.7
|
|
}
|
|
},
|
|
})
|
|
|
|
|
|
def test_generate_preserves_schema_defaults_for_dataclass_request(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls(
|
|
negative_prompt="model default",
|
|
num_frames=61,
|
|
height=448,
|
|
width=832,
|
|
)
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["sampling_param"] = sampling_param
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
generator.generate(
|
|
GenerationRequest(
|
|
prompt="hello world",
|
|
negative_prompt=None,
|
|
sampling=SamplingConfig(num_frames=125, height=720, width=1280),
|
|
))
|
|
|
|
assert captured["sampling_param"].negative_prompt is None
|
|
assert captured["sampling_param"].num_frames == 125
|
|
assert captured["sampling_param"].height == 720
|
|
assert captured["sampling_param"].width == 1280
|
|
|
|
|
|
def test_generate_mapping_request_preserves_model_defaults_for_omitted_fields(monkeypatch, ):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls(
|
|
negative_prompt="model default",
|
|
num_frames=61,
|
|
height=448,
|
|
width=832,
|
|
fps=16,
|
|
guidance_scale=3.0,
|
|
)
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["sampling_param"] = sampling_param
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
generator.generate({
|
|
"prompt": "hello world",
|
|
})
|
|
|
|
assert captured["sampling_param"].negative_prompt == "model default"
|
|
assert captured["sampling_param"].num_frames == 61
|
|
assert captured["sampling_param"].height == 448
|
|
assert captured["sampling_param"].width == 832
|
|
assert captured["sampling_param"].fps == 16
|
|
assert captured["sampling_param"].guidance_scale == 3.0
|
|
|
|
|
|
def test_generate_honors_post_load_request_mutations(monkeypatch, tmp_path):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
config_path = tmp_path / "run.yaml"
|
|
config_path.write_text(
|
|
"generator:\n"
|
|
" model_path: test-model\n"
|
|
"request:\n"
|
|
" prompt: hello world\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls(seed=1024, num_frames=61, height=448, width=832)
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["sampling_param"] = sampling_param
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
config = load_run_config(config_path)
|
|
config.request.sampling.seed = 7
|
|
|
|
generator.generate(config.request)
|
|
|
|
assert captured["sampling_param"].seed == 7
|
|
|
|
|
|
def test_generate_honors_post_load_mutations_matching_schema_defaults(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
config_path = tmp_path / "run.yaml"
|
|
config_path.write_text(
|
|
"generator:\n"
|
|
" model_path: test-model\n"
|
|
"request:\n"
|
|
" prompt: hello world\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls(guidance_scale=3.0)
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["sampling_param"] = sampling_param
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
config = load_run_config(config_path)
|
|
config.request.sampling.guidance_scale = 1.0
|
|
|
|
generator.generate(config.request)
|
|
|
|
assert captured["sampling_param"].guidance_scale == 1.0
|
|
|
|
|
|
def test_generate_removes_deleted_loaded_stage_overrides(monkeypatch, tmp_path):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
config_path = tmp_path / "run.yaml"
|
|
config_path.write_text(
|
|
"generator:\n"
|
|
" model_path: test-model\n"
|
|
"request:\n"
|
|
" prompt: hello world\n"
|
|
" stage_overrides:\n"
|
|
" refine:\n"
|
|
" t_thresh: 0.8\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def fake_from_pretrained(cls, model_path):
|
|
return cls(t_thresh=0.5)
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured["sampling_param"] = sampling_param
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
config = load_run_config(config_path)
|
|
del config.request.stage_overrides["refine"]
|
|
|
|
generator.generate(config.request)
|
|
|
|
assert captured["sampling_param"].t_thresh == 0.5
|
|
|
|
|
|
def test_generate_video_legacy_call_uses_legacy_impl(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
captured = {}
|
|
|
|
def fake_generate_video_impl(
|
|
prompt=None,
|
|
sampling_param=None,
|
|
mouse_cond=None,
|
|
keyboard_cond=None,
|
|
grid_sizes=None,
|
|
**kwargs,
|
|
):
|
|
captured["prompt"] = prompt
|
|
captured["sampling_param"] = sampling_param
|
|
captured["mouse_cond"] = mouse_cond
|
|
captured["keyboard_cond"] = keyboard_cond
|
|
captured["grid_sizes"] = grid_sizes
|
|
captured["kwargs"] = kwargs
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
with pytest.warns(DeprecationWarning):
|
|
result = generator.generate_video(
|
|
prompt="legacy prompt",
|
|
num_frames=49,
|
|
output_path="outputs/legacy",
|
|
save_video=False,
|
|
log_queue="queue-token",
|
|
)
|
|
|
|
assert captured["prompt"] == "legacy prompt"
|
|
assert captured["sampling_param"].num_frames == 49
|
|
assert captured["sampling_param"].output_path == "outputs/legacy"
|
|
assert captured["sampling_param"].save_video is False
|
|
assert result["video_path"] == "outputs/test.mp4"
|
|
|
|
|
|
def test_generate_video_legacy_call_routes_compat_kwargs(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
generator.fastvideo_args.pipeline_config = SimpleNamespace(embedded_cfg_scale=1.0)
|
|
captured = {}
|
|
|
|
def fake_generate_video_impl(
|
|
prompt=None,
|
|
sampling_param=None,
|
|
mouse_cond=None,
|
|
keyboard_cond=None,
|
|
grid_sizes=None,
|
|
**kwargs,
|
|
):
|
|
captured["prompt"] = prompt
|
|
captured["sampling_param"] = sampling_param
|
|
captured["kwargs"] = kwargs
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
with pytest.warns(DeprecationWarning):
|
|
result = generator.generate_video(
|
|
prompt="legacy prompt",
|
|
neg_prompt="custom negative",
|
|
embedded_cfg_scale=7.5,
|
|
)
|
|
|
|
assert captured["prompt"] == "legacy prompt"
|
|
assert captured["sampling_param"].negative_prompt == "custom negative"
|
|
assert captured["kwargs"]["fastvideo_args"].pipeline_config.embedded_cfg_scale == 7.5
|
|
assert not hasattr(captured["sampling_param"], "embedded_cfg_scale")
|
|
assert result["video_path"] == "outputs/test.mp4"
|
|
|
|
|
|
def test_generate_batch_prompt_file_returns_typed_results(tmp_path, monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
_patch_sampling_param_from_pretrained(monkeypatch)
|
|
prompt_file = tmp_path / "prompts.txt"
|
|
prompt_file.write_text("first prompt\nsecond prompt\n", encoding="utf-8")
|
|
output_dir = tmp_path / "outputs"
|
|
captured_prompts = []
|
|
|
|
def fake_generate_single_video(prompt, sampling_param=None, **kwargs):
|
|
captured_prompts.append(prompt)
|
|
return {"prompts": prompt, "video_path": kwargs["output_path"]}
|
|
|
|
monkeypatch.setattr(generator, "_generate_single_video", fake_generate_single_video)
|
|
|
|
results = generator.generate({
|
|
"inputs": {
|
|
"prompt_path": str(prompt_file)
|
|
},
|
|
"output": {
|
|
"output_path": str(output_dir),
|
|
"save_video": False,
|
|
"return_frames": False,
|
|
},
|
|
})
|
|
|
|
assert isinstance(results, list)
|
|
assert [result.prompt for result in results] == ["first prompt", "second prompt"]
|
|
assert [result.prompt_index for result in results] == [0, 1]
|
|
assert captured_prompts == ["first prompt", "second prompt"]
|
|
|
|
|
|
def test_generate_batched_request_fans_out_media_inputs(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
_patch_sampling_param_from_pretrained(monkeypatch)
|
|
captured: list[tuple[str | None, str | None, str | None]] = []
|
|
|
|
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
|
|
captured.append((prompt, sampling_param.image_path, sampling_param.video_path))
|
|
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
|
|
|
|
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
|
|
|
|
results = generator.generate(
|
|
GenerationRequest(
|
|
prompt=["first prompt", "second prompt"],
|
|
inputs=InputConfig(
|
|
image_path=["first.png", "second.png"],
|
|
video_path=["first.mp4", "second.mp4"],
|
|
),
|
|
))
|
|
|
|
assert [result.prompt for result in results] == ["first prompt", "second prompt"]
|
|
assert captured == [
|
|
("first prompt", "first.png", "first.mp4"),
|
|
("second prompt", "second.png", "second.mp4"),
|
|
]
|
|
|
|
|
|
def test_generate_batched_request_rejects_mismatched_media_inputs(monkeypatch):
|
|
generator = _new_runtime_video_generator()
|
|
_patch_sampling_param_from_pretrained(monkeypatch)
|
|
|
|
with pytest.raises(ValueError, match="image_path"):
|
|
generator.generate(
|
|
GenerationRequest(
|
|
prompt=["first prompt", "second prompt"],
|
|
inputs=InputConfig(image_path=["first.png"]),
|
|
))
|