babylon.metrics.collector

Metrics collection system for Babylon/Babylon.

The MetricsCollector observes the simulation without interfering. It is the Party’s Central Statistical Bureau - recording material conditions for analysis by the planning committee.

Classes

`MetricEvent`(**data)	A single metric measurement.
`MetricsCollector`()	Centralized metrics collection and aggregation.
`TimerContext`(collector, name)	Context manager for timing code blocks.

class babylon.metrics.collector.MetricEvent(**data)[source]

Bases: BaseModel

A single metric measurement.

Immutable record of a moment in the simulation’s history.

Parameters:

name (str)
value (float)
timestamp (datetime)
tags (dict[str, str])
metadata (dict[str, Any])

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

value: float

timestamp: datetime

tags: dict[str, str]

metadata: dict[str, Any]

class babylon.metrics.collector.MetricsCollector[source]

Bases: object

Centralized metrics collection and aggregation.

This class implements the Singleton pattern to ensure a single source of truth for all metrics across the simulation.

The collector is thread-safe for concurrent metric recording.

Metrics Categories: - performance: Timing, throughput, latency - simulation: Game state metrics (P(S|A), P(S|R), Rent flow) - cache: Hit rates, evictions, memory usage - embedding: Generation times, batch sizes, errors

Return type:: MetricsCollector

static __new__(cls)[source]

Implement singleton pattern with thread safety.

Return type:: MetricsCollector

__init__()[source]

Initialize the metrics collector.

Return type:: None

property enabled: bool: Check if metrics collection is enabled.

record(name, value, tags=None, metadata=None)[source]

Record a metric value.

Parameters:

name (str) – Metric name (e.g., “simulation.p_revolution”)
value (float) – Numeric value
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)[source]

Increment a counter metric.

Parameters:

name (str) – Counter name
value (int) – Amount to increment

Return type:

None

gauge(name, value)[source]

Set a gauge metric to a specific value.

Parameters:

name (str) – Gauge name
value (float) – Current value

Return type:

None

time(name)[source]

Context manager for timing operations.

Parameters:: name (str) – Timer name
Return type:: TimerContext
Returns:: TimerContext that records duration on exit

Example

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

record_timing(name, duration)[source]

Record a timing measurement directly.

Parameters:

name (str) – Timer name
duration (float) – Duration in seconds

Return type:

None

record_metric(name, value, context='', object_id=None, context_level=None)[source]

Record a named metric with context.

Parameters:

name (str) – Metric name
value (float) – Metric value
context (str) – Optional context string
object_id (str | None) – Optional object identifier
context_level (str | None) – Optional context level

Return type:

None

record_cache_event(level, hit)[source]

Record a cache hit or miss event.

Parameters:

level (str) – Cache level (e.g., “L1”, “L2”, “embedding”)
hit (bool) – Whether this was a cache hit (True) or miss (False)

Return type:

None

record_token_usage(tokens)[source]

Record token usage.

Parameters:: tokens (int) – Number of tokens used
Return type:: None

record_memory_usage(memory_bytes)[source]

Record memory usage.

Parameters:: memory_bytes (float) – Memory usage in bytes
Return type:: None

get_counter(name)[source]

Get current counter value.

Return type:: int
Parameters:: name (str)

get_gauge(name)[source]

Get current gauge value.

Return type:: float | None
Parameters:: name (str)

get_timer_stats(name)[source]

Get statistics for a timer.

Return type:: dict[str, float]
Returns:: Dict with count, sum, mean, min, max
Parameters:: name (str)

get_metrics(name, limit=100)[source]

Get recent metric events.

Parameters:

name (str) – Metric name
limit (int) – Maximum number of events to return

Return type:

list[MetricEvent]

Returns:

List of recent MetricEvent objects

clear()[source]

Clear all collected metrics.

Used primarily for testing or resetting between game sessions.

Return type:: None

summary()[source]

Get a summary of all collected metrics.

Return type:: dict[str, Any]
Returns:: Dict with counters, gauges, and timer statistics

class babylon.metrics.collector.TimerContext(collector, name)[source]

Bases: object

Context manager for timing code blocks.

Parameters:

collector (MetricsCollector)
name (str)

__init__(collector, name)[source]

Parameters:

collector (MetricsCollector)
name (str)

Return type:

None