SR-Forge

Observers & Events¶

Legacy — use Hooks instead

The Observer / EventBus / Observable machinery described on this page is the deprecated side-channel mechanism. New code should write a Hook subclass and pass it to Trainer(hooks=[…]) or EpochRunner(hooks=[…]) directly. Subclassing Observer or Observable emits a DeprecationWarning.

The classes on this page continue to ship unchanged so existing scripts run. Internally, bus.subscribe(observer) wraps each observer in an ObserverAdapter — a Hook subclass — and binds it to the runner / trainer's HookPoints, so the dispatch path is the same as for native hooks. See Trainers & Runners → Training Hooks.

Observers are pluggable components that react to training events — logging metrics, saving checkpoints, updating progress bars. They decouple monitoring from the training loop: the trainer doesn't know (or care) who is listening.

How It Works¶

The system has three parts:

Events — frozen dataclasses carrying data (RunnerBatchFinished, TrainerEpochFinished, etc.)
Observables — components that emit events (trainers and runners)
Observers — components that handle events (progress bars, loggers, checkpointers)

Events flow through the EventBus — a global dispatcher that routes events from observables to matching observers.

TrainingEpochRunner (Observable)
    │  emits RunnerBatchFinished
    ▼
  EventBus
    │  dispatches to matching subscribers
    ├──► ProgressBar.on_runner_batch_finished()
    └──► BatchImageLogger.on_runner_batch_finished()

Built-in Events¶

Runner Events¶

Emitted by TrainingEpochRunner, ValidationEpochRunner, and BenchmarkRunner:

Event	Fields	When
`RunnerEpochStarted`	`epoch`, `dataset_size`, `batch_size`, `num_batches`	Start of each epoch
`RunnerBatchFinished`	`epoch`, `batch`, `entry`, `batch_scores`, `criterion`, `epoch_scores`	After each batch
`RunnerEpochFinished`	`epoch`, `epoch_scores`	End of each epoch

Trainer Events¶

Emitted by PyTorchTrainer:

Event	Fields	When
`TrainingBegan`	`total_epochs`, `initial_epoch`, `train_batches_per_epoch`, `val_batches_per_epoch`, `best_losses`	Before the first epoch
`TrainerEpochFinished`	`model`, `train_loss` (`MetricScores`), `val_loss` (`MetricScores`), `epoch`, `lr_scheduler`, `runner_state` (dict; the runner's checkpointable state including optimizer + scaler)	After each epoch (train + val)

Built-in Observers¶

Observer	Listens To	Description
`ProgressBar`	Runner events	CLI progress bar during an epoch
`LossLogger`	`TrainerEpochFinished`	Logs train/val losses to the tracker
`PyTorchModelSaver`	`TrainingBegan`, `TrainerEpochFinished`	Saves best and last checkpoints
`BatchImageLogger`	`RunnerBatchFinished`	Logs sample prediction images to the tracker
`BatchImageSaver`	`RunnerBatchFinished`	Saves prediction images to disk

Subscribing Observers¶

Observers are resolved from config and explicitly subscribed to the event bus:

observers = resolve(cfg.observers)
GlobalSettings().event_bus.subscribe(observers)

subscribe() accepts a single observer or a list. Each observer's _bus_scope (set via the scope constructor parameter) controls which events it receives.

In YAML:

observers:
  - _target: srforge.observers.ProgressBar
    params:
      name: "T"
      scope: train          # only receives events from the training runner
  - _target: srforge.observers.ProgressBar
    params:
      name: "V"
      scope: val            # only receives events from the validation runner
  - _target: srforge.observers.LossLogger     # no scope = receives all events
  - _target: srforge.observers.PyTorchModelSaver

Scoping¶

Runners and trainers emit events with a scope tag (e.g., "train", "val"). Observers filter based on their own scope:

Observer Scope	Event Scope	Delivered?
`None` (no scope)	Any	Yes
`"train"`	`"train"`	Yes
`"train"`	`"val"`	No
`"train"`	`None` (trainer-level)	Yes

Rules:

No scope = wildcard — receives everything.
Scoped observer receives matching events plus all scopeless events (trainer-level events like TrainingBegan have no scope).

Set the scope on a runner via its scope parameter:

training_runner:
  _target: srforge.training.runners.TrainingEpochRunner
  params:
    scope: train             # events from this runner have scope="train"
    ...

validation_runner:
  _target: srforge.training.runners.ValidationEpochRunner
  params:
    scope: val               # events from this runner have scope="val"
    ...

`scope` vs `_scope` vs `_bus_scope`¶

Three related identifiers live under the hood — only the first is part of the public API:

Name	Lives on	Source	Purpose
`scope` (constructor kwarg)	`Observer.__init__` and `EpochRunner.__init__`	The YAML / Python caller	Public knob. Sets the observer's filter or the runner's emission tag.
`Observer._bus_scope`	every `Observer` instance	`super().__init__(scope=...)`	The filter the `EventBus` reads when deciding whether to deliver a runner event. `None` = wildcard.
`Observable._scope`	every `EpochRunner` / `Trainer`	runner / trainer `__init__`	The tag emitted with each event, picked up by `Observable.attach_bus(..., scope=...)`.
`Observable._bus_scope`	every `Observable` instance	`attach_bus(bus, scope=...)`	Internal: scope passed through to the bus on every emit.

In normal use you only ever set scope=... — _bus_scope and _scope are the framework's plumbing. See srforge/observers/base.py (Observer.__init__, Observable.__init__, Observable.attach_bus) for the wiring.

Writing a Custom Observer¶

1. Define the class¶

from srforge.observers.base import Observer
from srforge.events.runner import RunnerBatchFinished

class GradientMonitor(Observer):
    EVENTS = [RunnerBatchFinished]

    def __init__(self, model, tracker=None, log_every=100, **kwargs):
        super().__init__(**kwargs)              # passes scope to base
        self.model = model
        self.log_every = log_every
        self._step = 0
        if tracker is None:
            from srforge.tracking import NullTracker
            tracker = NullTracker()
        self.tracker = tracker

    def on_runner_batch_finished(self, event: RunnerBatchFinished):
        self._step += 1
        if self._step % self.log_every != 0:
            return
        total_norm = sum(
            p.grad.norm().item() for p in self.model.parameters()
            if p.grad is not None
        )
        self.tracker.log_metrics(
            {"grad_norm": total_norm}, step=event.epoch
        )

2. Key requirements¶

Subclass Observer and call super().__init__(**kwargs) (this sets self._bus_scope from the scope keyword).
Declare EVENTS — a list of event types this observer handles. ObserverMeta uses it for warnings (no handler for a listed event, handler for an event not listed) but it is not what wires handlers to event types.
Type-annotate the event parameter — this is the primary mechanism: ObserverMeta inspects every on_* method, reads the first non-self parameter's annotation, and registers the handler for that event class. The name on_<snake_case> is a fallback used only when the annotation is missing. If name and annotation disagree, the annotation wins (with a warning).
Accept tracker as an optional constructor parameter for observers that need to log metrics or save files. Default to NullTracker() if not provided.

Why annotation-first?

A handler that looks right by name but binds the wrong event class is silent — it never fires. Annotation-first wiring catches the typo at class-construction time (no Event subclass means the metaclass logs a warning and ignores the handler). See srforge/observers/base.py:66-100 for the exact resolution order.

3. Use in config¶

observers:
  - _target: my_project.observers.GradientMonitor
    params:
      model: ${ref:model}
      tracker: ${ref:tracker}    # injects the resolved tracker instance
      log_every: 50
      scope: train

No script changes needed — the observer is resolved and subscribed like any other.

The EventBus¶

The EventBus is a standalone dispatcher stored on GlobalSettings().event_bus. You rarely interact with it directly, but here's how it works:

from srforge import GlobalSettings

bus = GlobalSettings().event_bus

# Subscribe an observer (reads scope from observer._bus_scope)
bus.subscribe(observer)

# Subscribe with explicit scope override
bus.subscribe(observer, scope="train")

# Publish an event (done by Observable.notify internally)
bus.publish(event, scope="train")

Observable¶

Trainers and runners are Observable subclasses. They auto-attach to the global event bus on creation. When they call self.notify(event), the event is:

Delivered to any directly-attached observers (via add_observer)
Forwarded to the EventBus with the observable's scope tag

Most of the time you don't need direct observer attachment — the event bus handles everything.

Creating Custom Events¶

Define a frozen dataclass inheriting from Event:

from dataclasses import dataclass
from srforge.events.base import Event

@dataclass(frozen=True)
class MyCustomEvent(Event):
    epoch: int
    custom_metric: float

Emit it from an Observable:

self.notify(MyCustomEvent(epoch=42, custom_metric=0.95))

Handle it in an Observer:

class MyObserver(Observer):
    EVENTS = [MyCustomEvent]

    def on_my_custom_event(self, event: MyCustomEvent):
        print(f"Epoch {event.epoch}: {event.custom_metric}")

The handler name on_my_custom_event is derived from the event class name MyCustomEvent via CamelCase-to-snake_case conversion.

Next: Extending SR-Forge — Write custom models, transforms, losses, and more