ValidationCallback(*, pipeline_target: str, dataset_file: str, every_steps: int = 100, sampling_steps: list[int] | None = None, guidance_scale: float | None = None, num_frames: int | None = None, output_dir: str | None = None, sampling_timesteps: list[int] | None = None, **pipeline_kwargs: Any)
Bases: Callback
Generic validation callback driven entirely by YAML
config.
Works with any pipeline that follows the
PipelineCls.from_pretrained(...) + pipeline.forward()
contract.
Source code in fastvideo/train/callbacks/validation.py
| def __init__(
self,
*,
pipeline_target: str,
dataset_file: str,
every_steps: int = 100,
sampling_steps: list[int] | None = None,
guidance_scale: float | None = None,
num_frames: int | None = None,
output_dir: str | None = None,
sampling_timesteps: list[int] | None = None,
**pipeline_kwargs: Any,
) -> None:
self.pipeline_target = str(pipeline_target)
self.dataset_file = str(dataset_file)
self.every_steps = int(every_steps)
self.sampling_steps = ([int(s) for s in sampling_steps] if sampling_steps else [40])
self.guidance_scale = (float(guidance_scale) if guidance_scale is not None else None)
self.num_frames = (int(num_frames) if num_frames is not None else None)
self.output_dir = (str(output_dir) if output_dir is not None else None)
self.sampling_timesteps = ([int(s) for s in sampling_timesteps] if sampling_timesteps is not None else None)
self.pipeline_kwargs = dict(pipeline_kwargs)
# Set after on_train_start.
self._pipeline: Any | None = None
self._pipeline_key: tuple[Any, ...] | None = None
self._sampling_param: SamplingParam | None = None
self.tracker: Any = DummyTracker()
self.validation_random_generator: (torch.Generator | None) = None
self.seed: int = 0
|