Statistics Module

Data collection and analysis components for simulation output.

Counter

Event counting with rate calculation.

class simcraft.Counter[source]

Bases: object

Counter for discrete events.

Tracks the number of occurrences and calculates rates.

Parameters:

sim (Optional[Simulation]) – Parent simulation for rate calculations
name (str) – Counter name

Examples

>>> counter = Counter(sim, name="arrivals")
>>> counter.increment()
>>> counter.increment(5)
>>> print(counter.value)
6
>>> print(counter.rate)  # per time unit

name: str = ''

property value: int: Get current count.

property elapsed_time: float: Get elapsed time since start/reset.

property rate: float: Get rate (count per time unit).

increment(amount=1)[source]

Increment the counter.

Parameters:: amount (int) – Amount to increment by
Returns:: New counter value
Return type:: int

decrement(amount=1)[source]

Decrement the counter.

Parameters:: amount (int) – Amount to decrement by
Returns:: New counter value
Return type:: int

reset()[source]

Reset counter to zero.

Return type:: None

__init__(name='', _value=0, _start_time=0.0, _sim=None)

Parameters:

name (str)
_value (int)
_start_time (float)
_sim (Optional['Simulation'])

Return type:

None

class simcraft.statistics.counter.WindowedCounter[source]

Bases: object

Counter that tracks values over a sliding time window.

Useful for calculating rates over recent time periods.

Parameters:

sim (Simulation) – Parent simulation
window_size (float) – Size of sliding window in time units
name (str) – Counter name

Examples

>>> counter = WindowedCounter(sim, window_size=60.0, name="arrivals_per_hour")
>>> counter.increment()
>>> # ... time passes ...
>>> print(counter.window_rate)  # Rate over last 60 time units

__init__(sim, window_size, name='')[source]

Initialize windowed counter.

Parameters:

sim (Simulation)
window_size (float)
name (str)

Return type:

None

property name: str: Get counter name.

property window_size: float: Get window size.

increment(amount=1)[source]

Increment the counter.

Parameters:: amount (int) – Amount to increment by
Return type:: None

property window_count: int: Get count within the window.

property window_rate: float: Get rate within the window (count per time unit).

reset()[source]

Reset the counter.

Return type:: None

Tally

Discrete observation collector with statistical analysis using Welford’s algorithm.

class simcraft.Tally[source]

Bases: object

Collects discrete observations and calculates statistics.

Efficiently calculates running statistics without storing all observations (unless keep_history is True).

Parameters:

sim (Optional[Simulation]) – Parent simulation
name (str) – Tally name
keep_history (bool) – Whether to store all observations

Examples

>>> tally = Tally(name="service_time")
>>> for time in service_times:
...     tally.observe(time)
>>> print(tally.mean, tally.std)

name: str = ''

keep_history: bool = False

property count: int: Get number of observations.

property sum: float: Get sum of observations.

property mean: float: Get mean of observations.

property variance: float: Get sample variance.

property std: float: Get sample standard deviation.

property min: float: Get minimum observation.

property max: float: Get maximum observation.

property range: float: Get range (max - min).

observe(value)[source]

Record an observation.

Parameters:: value (float) – Observed value
Return type:: None

observe_batch(values)[source]

Record multiple observations.

Parameters:: values (List[float]) – Observed values
Return type:: None

percentile(p)[source]

Calculate percentile from stored history.

Parameters:: p (float) – Percentile (0-100)
Returns:: Percentile value
Return type:: float
Raises:: ValueError – If history not kept or empty

confidence_interval(confidence=0.95)[source]

Calculate confidence interval for the mean.

Parameters:: confidence (float) – Confidence level (e.g., 0.95 for 95%)
Returns:: (lower, upper) confidence bounds
Return type:: Tuple[float, float]

get_history()[source]

Get observation history.

Returns:: List of (time, value) tuples
Return type:: List[Tuple[float, float]]

reset()[source]

Reset all statistics.

Return type:: None

summary()[source]

Get summary statistics.

Returns:: Dictionary of statistics
Return type:: dict

__init__(name='', keep_history=False, _sim=None, _count=0, _mean=0.0, _m2=0.0, _min=inf, _max=-inf, _sum=0.0, _history=<factory>)

Parameters:

name (str)
keep_history (bool)
_sim (Optional['Simulation'])
_count (int)
_mean (float)
_m2 (float)
_min (float)
_max (float)
_sum (float)
_history (List[Tuple[float, float]])

Return type:

None

class simcraft.statistics.tally.BatchTally[source]

Bases: Tally

Tally that groups observations into batches for analysis.

Useful for batch means analysis and detecting initialization bias.

Parameters:

batch_size (int) – Number of observations per batch
name (str) – Tally name

__init__(batch_size=100, name='', sim=None)[source]

Initialize batch tally.

Parameters:

batch_size (int)
name (str)
sim (Optional['Simulation'])

Return type:

None

property batch_count: int: Get number of completed batches.

property batch_means: List[float]: Get list of batch means.

observe(value)[source]

Record an observation.

Parameters:: value (float)
Return type:: None

batch_variance()[source]

Calculate variance of batch means.

Useful for estimating variance of the overall mean.

Return type:: float

reset()[source]

Reset all statistics.

Return type:: None

TimeSeries

Time-weighted statistics collection (equivalent to O2DES HourCounter).

class simcraft.TimeSeries[source]

Bases: object

Time-weighted statistics collector (HourCounter equivalent).

Tracks a value that changes over time and calculates time-weighted statistics like average count, utilization, and average duration.

Parameters:

sim (Simulation) – Parent simulation for time tracking
name (str) – Time series name
initial_value (float) – Initial value
keep_history (bool) – Whether to store value history

Examples

>>> ts = TimeSeries(sim, name="queue_length")
>>> ts.observe_change(1)   # Entity enters queue
>>> ts.observe_change(1)   # Another entity
>>> # ... time passes ...
>>> ts.observe_change(-1)  # Entity leaves
>>> print(ts.average_value)  # Time-weighted average
>>> print(ts.average_duration)  # Average time per entry

__init__(sim, name='', initial_value=0.0, keep_history=False)[source]

Initialize time series.

Parameters:

sim (Simulation)
name (str)
initial_value (float)
keep_history (bool)

Return type:

None

property name: str: Get time series name.

property current_value: float: Get current value.

property elapsed_time: float: Get total elapsed time.

property cumulative_value: float: Get cumulative time-weighted value.

property average_value: float: Get time-weighted average value (average count).

property average_duration: float

Get average duration per entry.

For counting values (queue length), this gives the average time an item spends in the system.

property min_value: float: Get minimum observed value.

property max_value: float: Get maximum observed value.

property increment_rate: float: Get increment rate (per time unit).

property decrement_rate: float: Get decrement rate (per time unit).

property increment_count: int: Get number of increments.

property decrement_count: int: Get number of decrements.

property utilization: float

Get utilization ratio (fraction of time value > 0).

For capacity tracking, this gives the busy ratio.

observe_change(delta)[source]

Observe a change in value.

Parameters:: delta (float) – Change amount (positive or negative)
Returns:: New value after change
Return type:: float

observe_value(value)[source]

Set value directly (observe absolute value).

Parameters:: value (float) – New value
Return type:: None

histogram(bins=10, min_val=None, max_val=None)[source]

Calculate time-weighted histogram from history.

Parameters:

bins (int) – Number of bins
min_val (Optional[float]) – Minimum bin value
max_val (Optional[float]) – Maximum bin value

Returns:

List of (bin_start, bin_end, time_fraction)

Return type:

List[Tuple[float, float, float]]

Raises:

ValueError – If history not kept

percentile(p)[source]

Calculate time-weighted percentile from history.

Parameters:: p (float) – Percentile (0-100)
Returns:: Percentile value
Return type:: float

get_history()[source]

Get value history.

Returns:: List of (time, value) tuples
Return type:: List[Tuple[float, float]]

reset(initial_value=0.0)[source]

Reset statistics.

Parameters:: initial_value (float) – New initial value
Return type:: None

summary()[source]

Get summary statistics.

Returns:: Dictionary of statistics
Return type:: dict

class simcraft.statistics.time_series.CapacityTimeSeries[source]

Bases: TimeSeries

TimeSeries specialized for capacity tracking.

Tracks busy/idle states for a resource with fixed capacity.

Parameters:

sim (Simulation) – Parent simulation
capacity (int) – Total capacity
name (str) – Time series name

__init__(sim, capacity, name='')[source]

Initialize capacity time series.

Parameters:

sim (Simulation)
capacity (int)
name (str)

Return type:

None

property capacity: int: Get total capacity.

property available: int: Get available capacity.

property utilization: float: Get utilization ratio.

property is_busy: bool: Check if all capacity is in use.

property is_idle: bool: Check if no capacity is in use.

acquire(amount=1)[source]

Acquire capacity.

Parameters:: amount (int) – Amount to acquire
Returns:: True if successful
Return type:: bool

release(amount=1)[source]

Release capacity.

Parameters:: amount (int) – Amount to release
Returns:: True if successful
Return type:: bool

Monitor

Unified data collection with multiple metric types.

class simcraft.Monitor[source]

Bases: object

General-purpose data collection and monitoring.

Provides a unified interface for tracking counters, tallies, and time series, with support for data export and reporting.

Parameters:

sim (Simulation) – Parent simulation
name (str) – Monitor name

Examples

>>> monitor = Monitor(sim, name="performance")
>>>
>>> # Add collectors
>>> monitor.add_counter("arrivals")
>>> monitor.add_tally("service_time")
>>> monitor.add_time_series("queue_length")
>>>
>>> # Record data
>>> monitor.counters["arrivals"].increment()
>>> monitor.tallies["service_time"].observe(5.2)
>>> monitor.time_series["queue_length"].observe_change(1)
>>>
>>> # Get report
>>> report = monitor.report()

__init__(sim, name='')[source]

Initialize monitor.

Parameters:

sim (Simulation)
name (str)

Return type:

None

property name: str: Get monitor name.

property counters: Dict[str, Counter]: Get all counters.

property tallies: Dict[str, Tally]: Get all tallies.

property time_series: Dict[str, TimeSeries]: Get all time series.

add_counter(name)[source]

Add a counter.

Parameters:: name (str) – Counter name
Returns:: The created counter
Return type:: Counter

add_tally(name, keep_history=False)[source]

Add a tally.

Parameters:

name (str) – Tally name
keep_history (bool) – Whether to store observations

Returns:

The created tally

Return type:

Tally

add_time_series(name, initial_value=0.0, keep_history=False)[source]

Add a time series.

Parameters:

name (str) – Time series name
initial_value (float) – Initial value
keep_history (bool) – Whether to store history

Returns:

The created time series

Return type:

TimeSeries

add_custom_metric(name, calculator)[source]

Add a custom metric.

Parameters:

name (str) – Metric name
calculator (Callable) – Function that returns the metric value

Return type:

None

get_counter(name)[source]

Get counter by name.

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

get_tally(name)[source]

Get tally by name.

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

get_time_series(name)[source]

Get time series by name.

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

enable_event_logging()[source]

Enable event logging.

Return type:: None

log_event(event_type, data=None)[source]

Log an event.

Parameters:

event_type (str) – Type of event
data (Any) – Event data

Return type:

None

get_events(event_type=None)[source]

Get logged events.

Parameters:: event_type (Optional[str]) – Filter by event type
Returns:: List of (time, type, data) tuples
Return type:: List[Tuple[float, str, Any]]

report()[source]

Generate a summary report.

Returns:: Dictionary with all statistics
Return type:: Dict[str, Any]

reset()[source]

Reset all statistics.

Return type:: None

to_json(indent=2)[source]

Export report as JSON.

Parameters:: indent (int) – JSON indentation
Returns:: JSON string
Return type:: str

to_dataframe()[source]

Export time series data as pandas DataFrame.

Returns:: DataFrame with time series data
Return type:: pd.DataFrame
Raises:: ImportError – If pandas is not available

class simcraft.statistics.monitor.SimulationRecorder[source]

Bases: object

Records simulation state at regular intervals.

Useful for generating time-series plots and animations.

Parameters:

sim (Simulation) – Parent simulation
interval (float) – Recording interval

__init__(sim, interval=1.0)[source]

Initialize recorder.

Parameters:

sim (Simulation)
interval (float)

Return type:

None

add_collector(name, collector)[source]

Add a data collector.

Parameters:

name (str) – Data series name
collector (Callable) – Function that returns the current value

Return type:

None

start()[source]

Start recording.

Return type:: None

stop()[source]

Stop recording.

Return type:: None

get_records()[source]

Get all records.

Return type:: List[Dict[str, Any]]

to_dataframe()[source]

Export records as pandas DataFrame.

Returns:: DataFrame with recorded data
Return type:: pd.DataFrame

clear()[source]

Clear all records.

Return type:: None