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] = {}

Functions

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 artifacts such as videos or images.

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 artifacts such as videos or images.

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

    if artifacts:
        self.log(artifacts, step)

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.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, 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,
    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(
        project=experiment_name,
        dir=log_dir,
        config=config,
        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, 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,
    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,
                    run_name=run_name,
                ))

    if not tracker_instances:
        return DummyTracker()

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

    return SequentialTracker(tracker_instances)