Metrics
Centralized metrics tracking for SeQUeNCe simulations.
This module provides a global registry for recording simulation events.
Metrics are disabled by default; call enable() to opt in to recording.
- class sequence.utils.metrics.CollectContext(owner_name: str, storage: InMemoryStorage, delivery_owner: str | None = None, target_pairs: int | None = None, reservation_start_time: int | None = None, throughput: float | None = None)
Context passed to metrics when collecting trial results.
- owner_name
Node name for counter and fidelity metrics.
- Type:
str
- storage
In-memory store of recorded events for the trial.
- delivery_owner
Node name for delivery-time metrics.
- Type:
str | None
- target_pairs
Number of delivered pairs required to compute delivery time.
- Type:
int | None
- reservation_start_time
Simulation time when the reservation started (ps).
- Type:
int | None
- throughput
Application throughput supplied at collection time.
- Type:
float | None
- class sequence.utils.metrics.CounterMetric(prefix: str, failure_event: EventType, success_event: EventType, rate_field: str)
Tracks failure/success counts and a running success rate.
- collect(ctx: CollectContext) dict[str, Any]
Return failure, success, and success-rate counts for the trial owner.
- Parameters:
ctx – Collection context with the owner name to report.
- Returns:
Mapping of prefixed failure, success, and success-rate keys.
- property event_types: frozenset[EventType]
Event types this metric reacts to during recording.
- Returns:
Frozen set of event types handled by on_record.
- failures(owner_name: str) int
Return the failure count for an owner.
- Parameters:
owner_name – Name of the node or component to query.
- Returns:
Number of recorded failure events for the owner.
- on_record(event_type: EventType, owner_name: str, kwargs: dict[str, Any]) None
Increment failure or success counts and update the rate field in kwargs.
- Parameters:
event_type – Recorded event type; must match failure or success event.
owner_name – Name of the node or component that owns the event.
kwargs – Event payload; updated with the current success rate.
- property output_keys: frozenset[str]
Keys produced by
collect().- Returns:
Frozen set of keys written into per-trial result dictionaries.
- reset() None
Clear per-owner failure and success counts.
- success_rate(owner_name: str) float
Return the success rate for an owner.
- Parameters:
owner_name – Name of the node or component to query.
- Returns:
Ratio of successes to total attempts, or 0.0 if there are no attempts.
- successes(owner_name: str) int
Return the success count for an owner.
- Parameters:
owner_name – Name of the node or component to query.
- Returns:
Number of recorded success events for the owner.
- class sequence.utils.metrics.DeliveryTimeMetric(key: str = 'delivery_time', delivery_event: EventType | None = None)
Time to deliver N pairs relative to reservation start.
- collect(ctx: CollectContext) dict[str, Any]
Compute elapsed time to deliver the target number of pairs.
- Parameters:
ctx – Collection context with delivery owner, target pair count, reservation start time, and stored delivery events.
- Returns:
Mapping with delivery time in seconds, or NaN if data is insufficient.
- property event_types: frozenset[EventType]
Event types this metric reacts to during recording.
- Returns:
Frozen set of event types handled by on_record.
- property output_keys: frozenset[str]
Keys produced by
collect().- Returns:
Frozen set of keys written into per-trial result dictionaries.
- class sequence.utils.metrics.EventType(name: str)
Identifies a type of recordable simulation event.
- name
Unique string identifier for the event type.
- Type:
str
- class sequence.utils.metrics.EventTypes
Namespace for built-in simulation event types.
- class sequence.utils.metrics.FidelityMetric(key: str, event: EventType, field: str)
Collects fidelity values from matching success events.
- collect(ctx: CollectContext) dict[str, Any]
Collect fidelity values from matching success events for the owner.
- Parameters:
ctx – Collection context with owner name and event storage.
- Returns:
Mapping of the configured key to a list of fidelity values.
- property event_types: frozenset[EventType]
Event types this metric reacts to during recording.
- Returns:
Frozen set of event types handled by on_record.
- property output_keys: frozenset[str]
Keys produced by
collect().- Returns:
Frozen set of keys written into per-trial result dictionaries.
- class sequence.utils.metrics.InMemoryStorage
Store metric records in memory for the lifetime of the simulation.
- append(record: dict[str, Any]) None
Append a metric record to storage.
- Parameters:
record – Event record with event type, owner, simulation time, and fields.
- clear() None
Remove all stored records.
- get_all() list[dict[str, Any]]
Return all stored metric records.
- Returns:
Copy of every record appended since construction or last clear().
- get_by_event(event_type: EventType) list[dict[str, Any]]
Return records matching an event type.
- Parameters:
event_type – Event type to filter on.
- Returns:
Records whose event_type matches the given value.
- get_by_owner(owner_name: str) list[dict[str, Any]]
Return records for a given owner.
- Parameters:
owner_name – Owner name to filter on.
- Returns:
Records whose owner_name matches the given value.
- class sequence.utils.metrics.Metric
Base class for metrics that aggregate recorded events.
- abstractmethod collect(ctx: CollectContext) dict[str, Any]
Return trial result keys and values for this metric.
- Parameters:
ctx – Collection context with owner, storage, and trial parameters.
- Returns:
Mapping of output keys to per-trial values.
- abstract property event_types: frozenset[EventType]
Event types this metric reacts to during recording.
- Returns:
Frozen set of event types handled by on_record.
- on_record(event_type: EventType, owner_name: str, kwargs: dict[str, Any]) None
Update metric state when a matching event is recorded.
This method is called for every metric when it is recorded. Default implementation is a no-op. This method is completely optional for any metric subclasses. Should only be implemented if you need a hook when a metric is recorded.
- Parameters:
event_type – Type of the recorded event.
owner_name – Name of the node or component that owns the event.
kwargs – Mutable event payload; metrics may add derived fields.
- abstract property output_keys: frozenset[str]
Keys produced by
collect().- Returns:
Frozen set of keys written into per-trial result dictionaries.
- reset() None
Clear per-trial metric state.
This method is called for every metric by default when all metrics are reset. Default implementation is a no-op. This method is completely optional for any metric subclasses. Should only be implemented if you need a hook when the metric is reset.
- class sequence.utils.metrics.RateMetric(key: str = 'app_throughput')
Collects a rate value supplied at trial collection time (e.g. throughput).
- collect(ctx: CollectContext) dict[str, Any]
Return the throughput value supplied at collection time.
- Parameters:
ctx – Collection context; uses throughput when set.
- Returns:
Mapping with the configured rate key and throughput or NaN.
- property event_types: frozenset[EventType]
Event types this metric reacts to during recording.
- Returns:
Frozen set of event types handled by on_record.
- property output_keys: frozenset[str]
Keys produced by
collect().- Returns:
Frozen set of keys written into per-trial result dictionaries.
- sequence.utils.metrics.aggregate_trial_metrics(trials: list[dict[str, Any]], *, list_metric_cap: int | None = 500) dict[str, float]
Aggregate trial metrics across multiple trials.
- Parameters:
trials – Per-trial metric dictionaries from collect_trial_metrics.
list_metric_cap – Maximum list elements per trial to include when aggregating list metrics.
- Returns:
Mapping of
avg_*andstd_*keys to aggregated statistics.
- sequence.utils.metrics.clear_registry() None
Remove all metrics from the registry.
Intended for test isolation; also clears the counter lookup table.
- sequence.utils.metrics.collect_trial_metrics(owner_name: str, *, delivery_owner: str | None = None, target_pairs: int | None = None, reservation_start_time: int | None = None, throughput: float | None = None) dict[str, Any]
Collect per-trial metrics for a node from the metrics module.
- Parameters:
owner_name – Node name to collect counter and fidelity metrics for.
delivery_owner – Node name used for delivery-time metrics; defaults to owner_name.
target_pairs – Number of delivered pairs required to compute delivery time.
reservation_start_time – Simulation time when the reservation started (ps).
throughput – Application throughput to include in collected metrics.
- Returns:
Mapping of metric output keys to per-trial values.
- sequence.utils.metrics.configure(storage_type: str = 'in_memory') None
Configure metrics storage.
- Available storage options:
“in_memory”: The default, uses InMemoryStorage.
- Parameters:
storage_type – Storage backend identifier.
- sequence.utils.metrics.enable(metrics_to_enable: list[Metric]) None
Enable metrics recording for the given metrics.
The event types required by each metric are automatically derived from each metric’s event_types property and added to the recording filter. Metrics with empty event_types (e.g. RateMetric) contribute no events to the filter; enabling them alone will not cause any events to be recorded. That will only affect what is collected in collect_trial_metrics.
- Parameters:
metrics_to_enable – Metrics whose event types should be recorded.
- sequence.utils.metrics.get_counter(prefix: str) CounterMetric
Return a registered counter metric by prefix.
- Parameters:
prefix – Counter metric prefix used at registration time.
- Returns:
The matching counter metric.
- sequence.utils.metrics.get_event_type(name: str) EventType
Return a registered event type by name.
- Parameters:
name – Name of the event type to look up.
- Returns:
The registered event type.
- sequence.utils.metrics.list_event_types() list[EventType]
Return all registered event types.
- Returns:
All event types registered so far.
- sequence.utils.metrics.list_metrics() list[Metric]
Return registered metrics in registration order.
- Returns:
Registered metrics in the order they were added.
- sequence.utils.metrics.record(event_type: EventType, owner_name: str, **kwargs: Any) None
Record a metrics event if metrics are enabled for this event type.
- Parameters:
event_type – Type of simulation event to record.
owner_name – Name of the node or component that owns the event.
**kwargs – Additional event-specific fields stored with the record.
- sequence.utils.metrics.register_event_type(name: str) EventType
Register an event type.
- Parameters:
name – Unique name for the event type.
- Returns:
The registered event type, or the existing instance if already registered.
- sequence.utils.metrics.register_metric(metric: Metric) None
Register a metric.
- Parameters:
metric – Metric instance to add to the registry.
- Raises:
ValueError – If output keys collide with an existing metric.
- sequence.utils.metrics.register_time_provider(provider: Timeline) None
Register the active time source for recorded events.
- Parameters:
provider – Object supplying timestamps via
now().
- sequence.utils.metrics.reset_metrics() None
Reset per-trial state for all registered metrics.
Clears accumulated counts and other state held by each metric between trials.