trackers ¶

Utilities for logging metrics and artifacts to external trackers.

This module is inspired by the trackers implementation in https://github.com/huggingface/finetrainers and provides a minimal, shared interface that can be used across all FastVideo training pipelines.

Classes¶

fastvideo.training.trackers.BaseTracker ¶

BaseTracker()

Base tracker implementation.

The default tracker stores timing information but does not emit any logs.

Source code in fastvideo/training/trackers.py

def __init__(self) -> None:
    self._timed_metrics: dict[str, float] = {}

Methods:¶

fastvideo.training.trackers.BaseTracker.finish ¶

finish() -> None

Finalize the tracker session.

Source code in fastvideo/training/trackers.py

def finish(self) -> None:  # pragma: no cover - interface
    """Finalize the tracker session."""

fastvideo.training.trackers.BaseTracker.log ¶

log(metrics: dict[str, Any], step: int) -> None

Log metrics for the given step.

Source code in fastvideo/training/trackers.py

def log(self, metrics: dict[str, Any], step: int) -> None:  # pragma: no cover - interface
    """Log metrics for the given step."""
    # Merge timing metrics with provided metrics
    metrics = {**self._timed_metrics, **metrics}
    self._timed_metrics = {}

fastvideo.training.trackers.BaseTracker.log_artifacts ¶

log_artifacts(artifacts: dict[str, Any], step: int) -> None

Log tracker artifacts such as sampled media.

By default this is treated the same as :meth:log.

Source code in fastvideo/training/trackers.py

def log_artifacts(self, artifacts: dict[str, Any], step: int) -> None:
    """Log tracker artifacts such as sampled media.

    By default this is treated the same as :meth:`log`.
    """

    if artifacts:
        self.log(artifacts, step)

fastvideo.training.trackers.BaseTracker.log_file ¶

log_file(file_path: str, name: str | None = None) -> None

Attach a file to the tracker run (e.g. config YAML).

Source code in fastvideo/training/trackers.py

def log_file(
    self,
    file_path: str,
    name: str | None = None,
) -> None:
    """Attach a file to the tracker run (e.g. config YAML)."""

fastvideo.training.trackers.BaseTracker.video ¶

video(data: Any, *, caption: str | None = None, fps: int | None = None, format: str | None = None) -> Any | None

Create a tracker specific video artifact.

Trackers that do not support video artifacts should return None.

Source code in fastvideo/training/trackers.py

def video(
    self,
    data: Any,
    *,
    caption: str | None = None,
    fps: int | None = None,
    format: str | None = None,
) -> Any | None:
    """Create a tracker specific video artifact.

    Trackers that do not support video artifacts should return ``None``.
    """

    return None

fastvideo.training.trackers.DummyTracker ¶

DummyTracker()

Bases: BaseTracker

Tracker implementation used when logging is disabled.

Source code in fastvideo/training/trackers.py

def __init__(self) -> None:
    self._timed_metrics: dict[str, float] = {}

fastvideo.training.trackers.SequentialTracker ¶

SequentialTracker(trackers: Iterable[BaseTracker])

Bases: BaseTracker

A tracker that forwards logging calls to a sequence of trackers.

Source code in fastvideo/training/trackers.py

def __init__(self, trackers: Iterable[BaseTracker]) -> None:
    super().__init__()
    self._trackers: list[BaseTracker] = list(trackers)

fastvideo.training.trackers.SwanlabTracker ¶

SwanlabTracker(experiment_name: str, log_dir: str, *, config: dict[str, Any] | None = None, run_name: str | None = None)

Bases: BaseTracker

Tracker implementation for SwanLab.

Source code in fastvideo/training/trackers.py

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")

Methods:¶

fastvideo.training.trackers.SwanlabTracker.video ¶

video(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.

Source code in fastvideo/training/trackers.py

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)

fastvideo.training.trackers.Timer `dataclass` ¶

Timer(name: str, _start_time: float | None = None, _end_time: float | None = None)

Simple timer utility used by the trackers.

fastvideo.training.trackers.WandbTracker ¶

WandbTracker(experiment_name: str, log_dir: str, *, config: dict[str, Any] | None = None, entity: str | None = None, run_name: str | None = None)

Bases: BaseTracker

Tracker implementation for Weights & Biases.

Source code in fastvideo/training/trackers.py

def __init__(
    self,
    experiment_name: str,
    log_dir: str,
    *,
    config: dict[str, Any] | None = None,
    entity: str | None = None,
    run_name: str | None = None,
) -> None:
    super().__init__()

    import wandb

    pathlib.Path(log_dir).mkdir(parents=True, exist_ok=True)

    self._wandb = wandb
    self._run = wandb.init(
        entity=entity,
        project=experiment_name,
        dir=log_dir,
        config=(_sanitize_wandb_config(config) if config is not None else None),
        name=run_name,
    )
    logger.info("Initialized Weights & Biases tracker")

Functions:¶

fastvideo.training.trackers.initialize_trackers ¶

initialize_trackers(trackers: Iterable[str], *, experiment_name: str, config: dict[str, Any] | None, log_dir: str, entity: str | None = None, run_name: str | None = None) -> BaseTracker

Create tracker instances based on trackers configuration.

Source code in fastvideo/training/trackers.py

def initialize_trackers(
    trackers: Iterable[str],
    *,
    experiment_name: str,
    config: dict[str, Any] | None,
    log_dir: str,
    entity: str | None = None,
    run_name: str | None = None,
) -> BaseTracker:
    """Create tracker instances based on ``trackers`` configuration."""

    tracker_names = [tracker.lower() for tracker in trackers]
    if not tracker_names:
        return DummyTracker()

    unsupported = [name for name in tracker_names if name not in SUPPORTED_TRACKERS]
    if unsupported:
        raise ValueError(
            f"Unsupported tracker(s) provided: {unsupported}. Supported trackers: {sorted(SUPPORTED_TRACKERS)}")

    tracker_instances: list[BaseTracker] = []
    for tracker_name in tracker_names:
        if tracker_name == Trackers.NONE.value:
            tracker_instances.append(DummyTracker())
        elif tracker_name == Trackers.WANDB.value:
            tracker_instances.append(
                WandbTracker(
                    experiment_name,
                    os.path.abspath(log_dir),
                    config=config,
                    entity=entity,
                    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()

    if len(tracker_instances) == 1:
        return tracker_instances[0]

    return SequentialTracker(tracker_instances)

trackers ¶

Classes¶

fastvideo.training.trackers.BaseTracker ¶

Methods:¶

fastvideo.training.trackers.BaseTracker.finish ¶

fastvideo.training.trackers.BaseTracker.log ¶

fastvideo.training.trackers.BaseTracker.log_artifacts ¶

fastvideo.training.trackers.BaseTracker.log_file ¶

fastvideo.training.trackers.BaseTracker.video ¶

fastvideo.training.trackers.DummyTracker ¶

fastvideo.training.trackers.SequentialTracker ¶

fastvideo.training.trackers.SwanlabTracker ¶

Methods:¶

fastvideo.training.trackers.SwanlabTracker.video ¶

fastvideo.training.trackers.Timer dataclass ¶

fastvideo.training.trackers.WandbTracker ¶

Functions:¶

fastvideo.training.trackers.initialize_trackers ¶

fastvideo.training.trackers.Timer `dataclass` ¶