babylon.metrics.interfaces

Metrics collector interfaces and protocols.

This module defines the contract that all metrics collectors must implement, enabling dependency injection and testability throughout the codebase.

Classes

MetricsCollectorProtocol(*args, **kwargs)

Protocol defining the contract for metrics collectors.

class babylon.metrics.interfaces.MetricsCollectorProtocol(*args, **kwargs)

Bases: Protocol

Protocol defining the contract for metrics collectors.

This protocol ensures both the real MetricsCollector and any test doubles (spies, mocks) implement the same interface.

The protocol follows the “Dumb Spy” pattern for test doubles: implementations should record what was called, not calculate statistics. Statistical analysis belongs in the production MetricsCollector only.

Example

def process_data(collector: MetricsCollectorProtocol) -> None:
    collector.increment("items_processed")
    with collector.time("processing_duration"):
        # ... do work ...
        pass

record(name, value, tags=None, metadata=None)

Record a metric value.

Parameters:

• name (str) – Metric name (e.g., “simulation.p_revolution”).

• value (float) – Numeric value to record.

• tags (dict[str, str] | None) – Key-value pairs for filtering/grouping.

• metadata (dict[str, Any] | None) – Additional context.

Return type:

None

increment(name, value=1)

Increment a counter.

Parameters:

• name (str) – Counter name.

• value (int) – Amount to increment (default 1).

Return type:

None

gauge(name, value)

Set a gauge value.

Gauges represent point-in-time values that can go up or down.

Parameters:

• name (str) – Gauge name.

• value (float) – Current value.

Return type:

None

time(name)

Context manager for timing operations.

Parameters:

name (str) – Timer name.

Return type:

AbstractContextManager[Any]

Returns:

Context manager that records duration on exit.

Example

with collector.time("embedding.generation"):
    embeddings = generate_embeddings(texts)

summary()

Get aggregated summary of all metrics.

Return type:

dict[str, Any]

Returns:

Dict containing counters, gauges, timer statistics, etc.

clear()

Clear all recorded metrics.

Used primarily for testing or resetting between game sessions.

Return type:

None

__init__(*args, **kwargs)