Compare commits

...
Author SHA1 Message Date
SolitaryThinker bb3b3d0657 [ci]: collect the tracker unit tests in the unit lane
fastvideo/tests/training/test_trackers.py (CPU-only, fake-swanlab
fixtures) was not collected by any lane - the training lanes pick
specific subdirs and run_unit_test skipped tests/training entirely.
2026-07-06 23:10:15 -07:00
SolitaryThinker 67a8f7ac85 [fix]: complete SwanLab tracker integration 2026-07-06 23:10:15 -07:00
kared d133aef55d [feat]: add swanlab tracker 2026-07-06 23:10:15 -07:00
10 changed files with 561 additions and 13 deletions
+91
View File
@@ -0,0 +1,91 @@
# Training Trackers
FastVideo can send training metrics and validation media to Weights & Biases
or SwanLab. Tracking runs only on global rank 0, and local tracker files are
stored under `<output_dir>/tracker`.
## Supported Trackers
| Value | Backend | Installation |
|-------|---------|--------------|
| `wandb` | Weights & Biases | Included with FastVideo |
| `swanlab` | SwanLab | Install the optional `swanlab` dependency |
| `none` | Disable external tracking | No additional package |
You can enable more than one backend, for example `trackers: [wandb, swanlab]`.
Metrics and validation media are converted to the artifact type required by
each backend.
## Install SwanLab
For a published FastVideo installation, install the SwanLab extra:
```bash
uv pip install "fastvideo[swanlab]"
```
For an editable source checkout, include the same extra during installation:
```bash
uv pip install -e ".[swanlab]"
```
If FastVideo is already installed, you can install the compatible SDK directly:
```bash
uv pip install "swanlab>=0.6.7"
```
Authenticate once before starting a training run:
```bash
swanlab login
```
See the [SwanLab login documentation](https://docs.swanlab.cn/en/api/cli-swanlab-login.html)
for non-interactive and self-hosted setups.
## Configure Tracking
Select SwanLab in the YAML config used by the modular training framework:
```yaml
training:
checkpoint:
output_dir: outputs/my_run
tracker:
trackers: [swanlab]
project_name: my_project
run_name: my_run
```
To log to both supported services:
```yaml
training:
tracker:
trackers: [wandb, swanlab]
project_name: my_project
run_name: my_run
```
An empty or omitted `trackers` list selects W&B when `project_name` is set.
Use an explicit `none` entry to disable external tracking:
```yaml
training:
tracker:
trackers: [none]
```
## Validation Videos
SwanLab currently accepts GIF video artifacts. FastVideo converts validation
MP4 files and in-memory video arrays to GIF automatically before logging them.
For video files, FastVideo uses the sampling frame rate supplied by the caller,
or the source file's frame rate when no value is supplied. In-memory arrays use
the frame rate supplied by the caller. Both forms fall back to 16 FPS when no
frame rate is available.
For details about configuring validation callbacks, see
[Training Infrastructure](train_infra.md#callbacks-pluggable-hooks).
+3
View File
@@ -161,6 +161,9 @@ training:
decay_interval_steps: 0
```
See [Training Trackers](trackers.md) to configure Weights & Biases or SwanLab,
including SwanLab installation and authentication.
### `callbacks` — Pluggable hooks
Callbacks run at specific points in the training loop (before/after optimizer
+1 -1
View File
@@ -280,7 +280,7 @@ def run_self_forcing_tests():
@app.function(gpu="L40S:1", image=image, timeout=900)
def run_unit_test():
run_test(
"pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ ./fastvideo/tests/stages/ ./fastvideo/tests/ops/ --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
"pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ ./fastvideo/tests/stages/ ./fastvideo/tests/ops/ ./fastvideo/tests/training/test_trackers.py --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
)
+225
View File
@@ -0,0 +1,225 @@
from __future__ import annotations
import builtins
from io import BytesIO
from pathlib import Path
import sys
from types import ModuleType
from typing import Any
import numpy as np
from PIL import Image
import pytest
from fastvideo.training import trackers as trackers_module
from fastvideo.training.trackers import (
_coerce_video_fps,
_prepare_video_array,
BaseTracker,
SequentialTracker,
SwanlabTracker,
WandbTracker,
)
class _FakeSwanlab(ModuleType):
def __init__(self) -> None:
super().__init__("swanlab")
self.init_kwargs: dict[str, Any] | None = None
self.logged: list[tuple[dict[str, Any], int]] = []
self.video_paths: list[str] = []
self.finished = False
def init(self, **kwargs: Any) -> object:
self.init_kwargs = kwargs
return object()
def log(self, metrics: dict[str, Any], step: int) -> None:
self.logged.append((metrics, step))
def finish(self) -> None:
self.finished = True
def Video(self, file_path: str, caption: str | None = None) -> dict[str, Any]: # noqa: N802
self.video_paths.append(file_path)
return {
"content": Path(file_path).read_bytes(),
"caption": caption,
}
def _install_fake_swanlab(monkeypatch: pytest.MonkeyPatch) -> _FakeSwanlab:
swanlab = _FakeSwanlab()
monkeypatch.setitem(sys.modules, "swanlab", swanlab)
return swanlab
def test_prepare_video_array_converts_tchw_to_thwc() -> None:
video = np.arange(2 * 3 * 2 * 4, dtype=np.uint8).reshape(2, 3, 2, 4)
prepared = _prepare_video_array(video)
np.testing.assert_array_equal(prepared, video.transpose(0, 2, 3, 1))
def test_default_video_fps_is_16() -> None:
assert _coerce_video_fps(None) == 16
def test_wandb_video_uses_shared_default_fps(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
wandb = ModuleType("wandb")
video_calls: list[dict[str, Any]] = []
wandb.init = lambda **_: object()
def video(data: Any, **kwargs: Any) -> dict[str, Any]:
video_calls.append(kwargs)
return {"data": data, **kwargs}
wandb.Video = video
monkeypatch.setitem(sys.modules, "wandb", wandb)
tracker = WandbTracker("project", str(tmp_path))
tracker.video(np.zeros((2, 3, 2, 2), dtype=np.uint8))
assert video_calls == [{"fps": 16, "format": "mp4"}]
def test_prepare_video_array_tiles_and_pads_batches() -> None:
video = np.zeros((3, 1, 3, 1, 1), dtype=np.uint8)
video[0] = 1
video[1] = 2
video[2] = 3
prepared = _prepare_video_array(video)
assert prepared.shape == (1, 1, 4, 3)
np.testing.assert_array_equal(
prepared[0, 0],
np.array(
[
[1, 1, 1],
[2, 2, 2],
[3, 3, 3],
[0, 0, 0],
],
dtype=np.uint8,
),
)
def test_swanlab_video_converts_array_to_gif_and_logs_artifact(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
swanlab = _install_fake_swanlab(monkeypatch)
tracker = SwanlabTracker("project", str(tmp_path), run_name="run")
video = np.zeros((2, 3, 2, 2), dtype=np.uint8)
video[0, 0] = 255
video[1, 2] = 255
artifact = tracker.video(video, caption="validation sample", fps=5, format="mp4")
tracker.log_artifacts({"validation/videos": [artifact]}, step=7)
assert artifact["content"].startswith((b"GIF87a", b"GIF89a"))
assert artifact["caption"] == "validation sample"
assert not Path(swanlab.video_paths[0]).exists()
with Image.open(BytesIO(artifact["content"])) as image:
assert image.n_frames == 2
assert image.info["duration"] == 200
assert swanlab.logged == [({"validation/videos": [artifact]}, 7)]
@pytest.mark.parametrize(("requested_fps", "expected_fps"), [(None, 12.0), (2, 2.0)])
def test_swanlab_video_uses_requested_or_source_fps_during_gif_conversion(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
requested_fps: int | None,
expected_fps: float,
) -> None:
_install_fake_swanlab(monkeypatch)
tracker = SwanlabTracker("project", str(tmp_path))
frames = [np.zeros((2, 2, 3), dtype=np.uint8), np.ones((2, 2, 3), dtype=np.uint8)]
written_fps: list[float] = []
write_gif = trackers_module._write_gif
monkeypatch.setattr(trackers_module, "_read_video_file", lambda _: (frames, 12.0))
def capture_fps(file_path: str, video_frames: Any, fps: float) -> None:
written_fps.append(fps)
write_gif(file_path, video_frames, fps)
monkeypatch.setattr(trackers_module, "_write_gif", capture_fps)
artifact = tracker.video(tmp_path / "validation.mp4", fps=requested_fps)
assert artifact["content"].startswith((b"GIF87a", b"GIF89a"))
assert written_fps == [expected_fps]
def test_swanlab_video_passes_existing_gif_through(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
swanlab = _install_fake_swanlab(monkeypatch)
tracker = SwanlabTracker("project", str(tmp_path))
gif_path = tmp_path / "validation.gif"
Image.new("RGB", (2, 2)).save(gif_path, format="GIF")
artifact = tracker.video(gif_path, caption="existing")
assert swanlab.video_paths == [str(gif_path)]
assert gif_path.exists()
assert artifact["caption"] == "existing"
def test_swanlab_missing_dependency_error_is_actionable(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.delitem(sys.modules, "swanlab", raising=False)
real_import = builtins.__import__
def missing_swanlab(name: str, *args: Any, **kwargs: Any) -> Any:
if name == "swanlab":
raise ModuleNotFoundError("No module named 'swanlab'", name="swanlab")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", missing_swanlab)
with pytest.raises(ModuleNotFoundError, match=r"fastvideo\[swanlab\]"):
SwanlabTracker("project", str(tmp_path))
class _RecordingTracker(BaseTracker):
def __init__(self, name: str) -> None:
super().__init__()
self.name = name
self.artifact_logs: list[dict[str, Any]] = []
def video(
self,
data: Any,
*,
caption: str | None = None,
fps: int | None = None,
format: str | None = None,
) -> str:
return f"{self.name}:{data}"
def log_artifacts(self, artifacts: dict[str, Any], step: int) -> None:
self.artifact_logs.append(artifacts)
def test_sequential_tracker_logs_backend_specific_video_artifacts() -> None:
first = _RecordingTracker("first")
second = _RecordingTracker("second")
tracker = SequentialTracker([first, second])
artifact = tracker.video("sample.mp4")
tracker.log_artifacts({"validation/videos": [artifact]}, step=3)
assert first.artifact_logs == [{"validation/videos": ["first:sample.mp4"]}]
assert second.artifact_logs == [{"validation/videos": ["second:sample.mp4"]}]
+1
View File
@@ -404,6 +404,7 @@ class ValidationCallback(Callback):
art = self.tracker.video(
fname,
caption=cap,
fps=sp.fps,
)
if art is not None:
video_logs.append(art)
+1 -1
View File
@@ -1180,7 +1180,7 @@ class DistillationPipeline(TrainingPipeline):
artifacts = []
for filename, caption in zip(video_filenames, all_captions, strict=True):
video_artifact = self.tracker.video(filename, caption=caption)
video_artifact = self.tracker.video(filename, caption=caption, fps=sampling_param.fps)
if video_artifact is not None:
artifacts.append(video_artifact)
if artifacts:
+234 -10
View File
@@ -7,22 +7,28 @@ interface that can be used across all FastVideo training pipelines.
from __future__ import annotations
from collections.abc import Iterable, Iterator
import contextlib
import copy
import math
import os
import pathlib
import tempfile
import time
from dataclasses import dataclass
from enum import Enum
from typing import Any
from collections.abc import Iterable, Iterator
import numpy as np
import torch
from fastvideo.logger import init_logger
logger = init_logger(__name__)
_DEFAULT_VIDEO_FPS = 16
_MISSING_ARTIFACT = object()
def _sanitize_wandb_config(value: Any) -> Any:
"""Best-effort conversion of nested config objects to W&B-safe values."""
@@ -56,6 +62,140 @@ def _sanitize_wandb_config(value: Any) -> Any:
return repr(value)
def _prepare_video_array(data: Any) -> np.ndarray:
"""Convert W&B-style TCHW/BTCHW video data into GIF-ready frames."""
if isinstance(data, torch.Tensor):
data = data.detach().cpu().numpy()
video = np.asarray(data)
if video.ndim == 4:
video = video.reshape(1, *video.shape)
elif video.ndim != 5:
raise ValueError("Video data must have shape [T, C, H, W] or [B, T, C, H, W]")
batch_size, num_frames, channels, height, width = video.shape
if batch_size == 0 or num_frames == 0:
raise ValueError("Video data must contain at least one batch item and one frame")
if channels not in (1, 3, 4):
raise ValueError(f"Video data must have 1, 3, or 4 channels; got {channels}")
if video.dtype != np.uint8:
logger.warning("Converting video data to uint8 for SwanLab")
video = video.astype(np.uint8)
# Match wandb.Video's batch tiling so the same input has a familiar layout
# in either tracker.
if batch_size & (batch_size - 1):
padded_batch_size = 1 << batch_size.bit_length()
padding = np.zeros(
(padded_batch_size - batch_size, num_frames, channels, height, width),
dtype=video.dtype,
)
video = np.concatenate((video, padding), axis=0)
num_rows = 1 << ((batch_size.bit_length() - 1) // 2)
num_columns = video.shape[0] // num_rows
video = video.reshape(num_rows, num_columns, num_frames, channels, height, width)
video = np.transpose(video, axes=(2, 0, 4, 1, 5, 3))
video = video.reshape(num_frames, num_rows * height, num_columns * width, channels)
if channels == 1:
video = video[..., 0]
return np.ascontiguousarray(video)
def _read_video_file(file_path: str) -> tuple[list[np.ndarray], float | None]:
"""Decode a video file and return its frames and encoded frame rate."""
import imageio.v2 as imageio
reader = imageio.get_reader(file_path)
try:
metadata = reader.get_meta_data() or {}
frames = [np.asarray(frame) for frame in reader]
finally:
reader.close()
if not frames:
raise ValueError(f"Video file contains no frames: {file_path}")
raw_source_fps = metadata.get("fps")
try:
source_fps = float(str(raw_source_fps))
except ValueError:
source_fps = None
if source_fps is not None and (not math.isfinite(source_fps) or source_fps <= 0):
source_fps = None
return frames, source_fps
def _coerce_video_fps(fps: int | float | None) -> float:
if fps is None:
return _DEFAULT_VIDEO_FPS
value = float(fps)
if not math.isfinite(value) or value <= 0:
raise ValueError(f"Video fps must be a positive finite number; got {fps!r}")
return value
def _write_gif(file_path: str, frames: Any, fps: float) -> None:
"""Encode frames as an animated GIF for SwanLab's video API."""
from PIL import Image
images = []
for frame in frames:
array = np.asarray(frame)
if array.dtype != np.uint8:
array = array.astype(np.uint8)
if array.ndim == 3 and array.shape[-1] == 1:
array = array[..., 0]
if array.ndim not in (2, 3) or (array.ndim == 3 and array.shape[-1] not in (3, 4)):
raise ValueError(f"GIF frames must be grayscale, RGB, or RGBA; got shape {array.shape}")
images.append(Image.fromarray(array))
if not images:
raise ValueError("Video data must contain at least one frame")
duration_ms = max(1, round(1000 / fps))
images[0].save(
file_path,
format="GIF",
save_all=True,
append_images=images[1:],
duration=duration_ms,
loop=0,
)
@dataclass(frozen=True)
class _SequentialArtifact:
"""Backend-specific versions of one artifact created by a tracker group."""
values: tuple[Any | None, ...]
def _select_tracker_artifact(value: Any, tracker_index: int) -> Any:
"""Resolve nested sequential artifacts for one child tracker."""
if isinstance(value, _SequentialArtifact):
selected = value.values[tracker_index]
return _MISSING_ARTIFACT if selected is None else selected
if isinstance(value, dict):
selected_dict = {}
for key, item in value.items():
selected = _select_tracker_artifact(item, tracker_index)
if selected is not _MISSING_ARTIFACT:
selected_dict[key] = selected
return selected_dict if selected_dict else _MISSING_ARTIFACT
if isinstance(value, list):
selected_list = [
selected for item in value
if (selected := _select_tracker_artifact(item, tracker_index)) is not _MISSING_ARTIFACT
]
return selected_list if selected_list else _MISSING_ARTIFACT
if isinstance(value, tuple):
selected_tuple = tuple(selected for item in value
if (selected := _select_tracker_artifact(item, tracker_index)) is not _MISSING_ARTIFACT)
return selected_tuple if selected_tuple else _MISSING_ARTIFACT
return value
@dataclass
class Timer:
"""Simple timer utility used by the trackers."""
@@ -222,8 +362,7 @@ class WandbTracker(BaseTracker):
kwargs: dict[str, Any] = {}
if caption is not None:
kwargs["caption"] = caption
if fps is not None:
kwargs["fps"] = fps
kwargs["fps"] = fps if fps is not None else _DEFAULT_VIDEO_FPS
if format is not None:
kwargs["format"] = format
else:
@@ -254,8 +393,10 @@ class SequentialTracker(BaseTracker):
self._timed_metrics = {}
def log_artifacts(self, artifacts: dict[str, Any], step: int) -> None:
for tracker in self._trackers:
tracker.log_artifacts(artifacts, step)
for tracker_index, tracker in enumerate(self._trackers):
tracker_artifacts = _select_tracker_artifact(artifacts, tracker_index)
if tracker_artifacts is not _MISSING_ARTIFACT:
tracker.log_artifacts(tracker_artifacts, step)
self._timed_metrics = {}
def log_file(
@@ -278,16 +419,91 @@ class SequentialTracker(BaseTracker):
fps: int | None = None,
format: str | None = None,
) -> Any | None:
for tracker in self._trackers:
video = tracker.video(data, caption=caption, fps=fps, format=format)
if video is not None:
return video
return None
videos = tuple(tracker.video(data, caption=caption, fps=fps, format=format) for tracker in self._trackers)
if all(video is None for video in videos):
return None
return _SequentialArtifact(videos)
class SwanlabTracker(BaseTracker):
"""Tracker implementation for SwanLab."""
def __init__(
self,
experiment_name: str,
log_dir: str,
*,
config: dict[str, Any] | None = None,
run_name: str | None = None,
) -> None:
super().__init__()
try:
import swanlab
except ModuleNotFoundError as error:
if error.name != "swanlab":
raise
raise ModuleNotFoundError("SwanLab tracking requires the optional 'swanlab' dependency. "
"Install it with `uv pip install 'fastvideo[swanlab]'` (or "
"`uv pip install -e '.[swanlab]'` from a source checkout).") from error
pathlib.Path(log_dir).mkdir(parents=True, exist_ok=True)
self._swanlab = swanlab
self._run = swanlab.init(
project=experiment_name,
experiment_name=run_name,
config=(_sanitize_wandb_config(config) if config is not None else None),
logdir=log_dir,
)
logger.info("Initialized SwanLab tracker")
def log(self, metrics: dict[str, Any], step: int) -> None:
metrics = {**self._timed_metrics, **metrics}
if metrics:
self._swanlab.log(metrics, step=step)
self._timed_metrics = {}
def finish(self) -> None:
self._swanlab.finish()
def video(
self,
data: Any,
*,
caption: str | None = None,
fps: int | None = None,
format: str | None = None,
) -> Any:
"""Create a SwanLab GIF artifact from a file or W&B-style array."""
del format # SwanLab currently supports GIF artifacts only.
if isinstance(data, str | os.PathLike):
source_path = os.fspath(data)
if pathlib.Path(source_path).suffix.lower() == ".gif":
return self._swanlab.Video(source_path, caption=caption)
frames, source_fps = _read_video_file(source_path)
if fps is not None:
video_fps = _coerce_video_fps(fps)
else:
video_fps = source_fps if source_fps is not None else _coerce_video_fps(None)
else:
frames = _prepare_video_array(data)
video_fps = _coerce_video_fps(fps)
file_descriptor, gif_path = tempfile.mkstemp(suffix=".gif")
os.close(file_descriptor)
try:
_write_gif(gif_path, frames, video_fps)
return self._swanlab.Video(gif_path, caption=caption)
finally:
pathlib.Path(gif_path).unlink(missing_ok=True)
class Trackers(str, Enum):
NONE = "none"
WANDB = "wandb"
SWANLAB = "swanlab"
SUPPORTED_TRACKERS = {tracker.value for tracker in Trackers}
@@ -324,6 +540,14 @@ def initialize_trackers(
config=config,
run_name=run_name,
))
elif tracker_name == Trackers.SWANLAB.value:
tracker_instances.append(
SwanlabTracker(
experiment_name,
os.path.abspath(log_dir),
config=config,
run_name=run_name,
))
if not tracker_instances:
return DummyTracker()
+1 -1
View File
@@ -831,7 +831,7 @@ class TrainingPipeline(LoRAPipeline, ABC):
artifacts = []
for filename, caption in zip(video_filenames, all_captions, strict=True):
video_artifact = self.tracker.video(filename, caption=caption)
video_artifact = self.tracker.video(filename, caption=caption, fps=sampling_param.fps)
if video_artifact is not None:
artifacts.append(video_artifact)
if artifacts:
+1
View File
@@ -169,6 +169,7 @@ nav:
- Training:
- Overview: training/overview.md
- Training Infrastructure: training/train_infra.md
- Experiment Tracking: training/trackers.md
- Data Preprocessing: training/data_preprocess.md
- Fine-tuning: training/finetune.md
- Examples:
+3
View File
@@ -120,6 +120,9 @@ flash-attn-4 = { git = "https://github.com/Dao-AILab/flash-attention.git", rev =
# flash-attn: uv pip install flash-attn==2.8.1 --no-cache-dir --no-build-isolation
swanlab = [
"swanlab>=0.6.7",
]
lint = [
"pre-commit==4.0.1",