mineproductivity.simulation¶
Auto-generated API reference for the Simulation package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Simulation.
simulation ¶
mineproductivity.simulation -- the platform's projection layer,
built directly on top of digital_twin.
Answers a question none of the seven packages below it were ever meant
to answer: given a scenario -- a hypothetical or historical starting
condition and a set of parameters -- how does this mining system
evolve over time, and how confident should we be in that projection?
digital_twin represents what a system currently looks like;
decision defines a stable interface for reasoning about
hypothetical business outcomes; neither implements an actual
forward-projection algorithm, by design. simulation supplies that
missing algorithmic layer -- scenario management, time-stepped or
event-stepped execution, experiment orchestration, and calibration
against history -- while still declining to choose which Monte
Carlo, discrete-event, or system-dynamics algorithm is correct for any
given site, leaving that choice to pluggable, independently-versioned
models (design spec §1, §13-§16).
The complete package (design spec §6's twenty-one modules) is
implemented: SimulationModel/SimulationContext (§8, no shared
abstract execution method by design); SimulationMetadata/
SimulationCategory (§29); Scenario/ScenarioStatus as a
versioned, governed artifact (§9); SimulationRun (a
core.BaseEntity[str] subclass, the second entity-shaped package in
the series)/RunStatus/SimulationExecutor (§10);
SimulationState (§10)/SimulationClock/TimeProgressionMode
(§11); seed_from_replay (§12); the four interface-only ABCs --
MonteCarloModel, DiscreteEventModel, SystemDynamicsModel,
CalibrationModel -- with zero concrete subclasses by design
(§13-§16, ADR-0009); Experiment/ExperimentRunner (§17);
ScenarioComparator/SensitivityAnalyzer, which delegate every
statistical judgment to analytics (§19-§20);
by_category/by_scope discovery (§22);
SimulationRunRepository as a literal type alias over
core.BaseRepository[SimulationRun, str] (§24);
SimulationStateCache (§26); the
SimulationResult/ExperimentResult family (§18);
REGISTRY/register (§21); and the full exception hierarchy. It
holds no KPI formulas, no statistical computation of its own, no
business-decision logic, no optimization search, no AI-agent
reasoning, no rendering, and no telemetry ingestion (§3).
simulation depends on core, ontology, events,
registry, plugins, connectors, kpis, analytics,
decision, and digital_twin -- and MUST NEVER import
optimization, agents, or visualization, none of which this
package may see.
SimulationRunRepository ¶
The storage contract for run instances, keyed by their own identity
-- a literal type alias over
core.BaseRepository[SimulationRun, str], never a parallel
simulation-specific repository ABC (design spec §24).
SimulationContext ¶
Bundles the collaborators and evidence a SimulationModel may
need -- the simulation-layer counterpart to
digital_twin.TwinContext (spec 08 §8), one layer up. Carries
the KPIResult/AnalyticsResult/DecisionResult evidence
already gathered, plus access to events.EventStore for
replay-based seeding (design spec §12). Evidence is gathered once,
at the boundary, by the caller -- the executor never reaches back
into a lower package on its own initiative (§4's runtime request
flow).
Examples:
>>> class _FakeStore: ...
>>> context = SimulationContext(event_store=_FakeStore())
>>> context.kpi_results
()
>>> context.decision_results
()
SimulationModel ¶
Bases: ABC
The root of every registrable simulation model type --
'Simulation-as-object,' the direct counterpart of
kpis.BaseKPI/analytics.AnalyticsModel/decision.DecisionModel/
digital_twin.Twin, four/three/two/one layers down respectively.
Deliberately carries no shared abstract execution method (see the
module docstring); each category base (design spec §13-§15)
declares its own.
Statelessness. Every SimulationModel subclass, of every
category, is stateless -- no instance attribute is mutated by any
category method (design spec §29, §32); statefulness in this
package lives entirely in SimulationRun (§10), never in a
model implementation. A concrete MonteCarloModel in particular
derives all randomness from the random_seed supplied per
_trial call, never from internal generator state (§33, §34's
recorded anti-pattern).
SimulationStateCache ¶
Caches the SimulationState seeded for one
(scenario_code, as_of) key (design spec §26).
Thread safety. Reads/writes are keyed by
(scenario_code, as_of); independent keys never contend beyond
the single internal lock's brief critical section, and writes to
the same key serialize on it (design spec §33) -- the same one-lock
posture digital_twin.TwinStateCache already takes one layer
down.
Examples:
>>> from datetime import datetime, timezone
>>> cache = SimulationStateCache()
>>> as_of = AsOf(utc=datetime(2026, 7, 8, tzinfo=timezone.utc))
>>> cache.get("FLEET.NightShiftSurge", as_of) is None
True
>>> state = SimulationState(
... attributes={"events_replayed": 3},
... simulated_time=datetime(2026, 7, 8, tzinfo=timezone.utc),
... )
>>> cache.put("FLEET.NightShiftSurge", as_of, state)
>>> cache.get("FLEET.NightShiftSurge", as_of) is state
True
CalibrationModel ¶
Bases: ABC
The contract a future calibration plugin implements: fit a
registered SimulationModel's parameters against historical
ground truth. THIS MODULE SHIPS NO CONCRETE SUBCLASS (design spec
§16, §35's interface-purity proof).
SimulationClock ¶
Represents how simulated time advances for a given
TimeProgressionMode (design spec §11): a fixed-timestep clock
advances by a constant dt each step; a next-event clock
advances directly by the delta the model's _advance returned;
a trial-based clock does not advance continuous time at all --
each trial is independent, and 'time' within a trial is whatever
the concrete MonteCarloModel implementation defines it to be.
Examples:
>>> clock = SimulationClock(mode=TimeProgressionMode.FIXED_TIMESTEP, dt=timedelta(minutes=5))
>>> clock.advance(datetime(2026, 7, 8)).minute
5
>>> next_event = SimulationClock(mode=TimeProgressionMode.NEXT_EVENT)
>>> next_event.advance(datetime(2026, 7, 8), delta=timedelta(minutes=17)).minute
17
>>> trial = SimulationClock(mode=TimeProgressionMode.TRIAL_BASED)
>>> trial.advance(datetime(2026, 7, 8)) == datetime(2026, 7, 8)
True
advance ¶
The next simulated instant after current. In
NEXT_EVENT mode delta is the model-supplied time until
the next scheduled event (design spec §14) and is required; in
FIXED_TIMESTEP mode the constructor-fixed dt governs
and delta is ignored; in TRIAL_BASED mode continuous
time never advances.
TimeProgressionMode ¶
Bases: Enum
Which of the three interface-only methodologies (design spec
§13-§15) a SimulationClock is paced for.
ScenarioComparator ¶
Compares simulation outcomes across two or more scenarios by
delegating statistical treatment to analytics -- never
computing descriptive statistics itself (design spec §19).
compare ¶
For each scenario key, extracts the numeric attribute of
interest from its SimulationState sequence and calls
analytics.describe() -- never re-implements
mean/percentile computation here (design spec §19).
DiscreteEventModel ¶
Bases: SimulationModel, ABC
The contract a future discrete-event plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS (design spec §14, §35's interface-purity proof).
ScenarioConflictError ¶
Bases: RegistrationError
A governance action attempted to re-register an existing,
Active Scenario code with different parameters/initial
conditions without a version bump and a Superseded transition
for the prior version -- the Scenario-layer analogue of
:class:SimulationVersionConflictError, since a Scenario is a
separate governed artifact from a SimulationModel
implementation (design spec §9, §25).
SimulationExecutionError ¶
Bases: MineProductivityError
SimulationExecutor raised for a batch of steps/trials that
should have been structurally valid -- distinct from a
legitimately-incomplete-input case (design spec §8's 'qualify,
don't coerce' rule), which returns a SimulationResult carrying
a warning instead of raising (§28).
SimulationRunNotFoundError ¶
Bases: NotFoundError
SimulationRunRepository.get(run_id) found no run for that
id, or REGISTRY.get(code) found no registered
SimulationModel for that code (design spec §6).
SimulationValidationError ¶
Bases: ValidationError
A SimulationMetadata, Scenario, or SimulationState
failed validation (design spec §29, §9, §10) -- e.g. an empty
code, a Scenario with no model_code, or a
SimulationState with empty attributes.
SimulationVersionConflictError ¶
Bases: RegistrationError
A plugin attempted to re-register an existing SimulationModel
type code with materially different metadata without a version
bump, mirroring digital_twin.TwinVersionConflictError (spec 08
§6) and decision.DecisionVersionConflictError (spec 07 §6).
SimulationExecutor ¶
Orchestrates one SimulationRun: fetches the current instance
from a SimulationRunRepository, seeds the initial
SimulationState (snapshot, cached replay, or the run's own
provisioned state), dispatches to the registered model's
category-specific method, advances the SimulationClock, and
persists each produced state -- design spec §10's sequence diagram,
exactly.
execute ¶
Execute scenario for the run stored under run_id.
A run already Completed/Failed is terminal (design spec
§10): execution is skipped and a warning-carrying
SimulationResult returned, never a raise. A zero-length
time_horizon is a legitimately incomplete input (§28):
zero steps execute, the seeded state is final, and the result
carries a warning.
Raises:
| Type | Description |
|---|---|
SimulationRunNotFoundError
|
If no run is stored under |
SimulationValidationError
|
If the model's category is not forward-executable
( |
SimulationExecutionError
|
If the model's category method raised for a structurally
valid input (the run is marked |
Experiment
dataclass
¶
Bases: BaseValueObject
A named collection of SimulationRun\ s produced from one
Scenario -- many independent Monte Carlo trials, or one run per
point in a sensitivity sweep (design spec §17, §20).
Examples:
>>> from datetime import timedelta
>>> scenario = Scenario(
... code="FLEET.NightShiftSurge", model_code="MONTECARLO.HaulCycleVariability",
... time_horizon=timedelta(hours=12),
... )
>>> Experiment(name="surge-x2", scenario=scenario, run_ids=("RUN-1", "RUN-2")).run_ids
('RUN-1', 'RUN-2')
ExperimentRunner ¶
Runs a Scenario across many trials (Monte Carlo, design spec
§13) or many parameter values (a sensitivity sweep, §20),
collecting the resulting runs into one Experiment. Trials are
dispatched concurrently in this reference implementation (§33);
run_ids preserve trial order regardless of completion order.
Zero trials is a legitimately incomplete input (§28): an empty
Experiment is returned, never a raise.
run_trials ¶
Provision and execute trials independent runs of
scenario, each with its own distinct random_seed (the
trial index -- §13's reproducibility anchor), returning the
Experiment naming them in trial order.
SimulationCategory ¶
Bases: Enum
Closed enum -- adding a member is a governance-reviewed change,
mirroring digital_twin.TwinCategory's/decision.DecisionCategory's
closed-enum rule (spec 08 §26, spec 07 §30).
SimulationMetadata
dataclass
¶
SimulationMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, version='1.0.0')
Bases: BaseMetadata
The minimal registration schema for a discoverable
SimulationModel type (design spec §29). code names a
type (e.g. "MONTECARLO.HaulCycleVariability"), never a
run -- the same distinction spec 08 §26 draws between
TwinMetadata.code and Twin.id, applied here between a model
type's code and a SimulationRun.id.
BaseMetadata.name has no default upstream and a model type's
code already serves as its identifier, so name defaults to
code (via :meth:_normalize) whenever a caller does not supply
one explicitly -- the same convention TwinMetadata/
DecisionMetadata already established.
Examples:
>>> meta = SimulationMetadata(
... code="MONTECARLO.HaulCycleVariability",
... category=SimulationCategory.MONTE_CARLO,
... description="Randomized haul-cycle variability trials.",
... )
>>> meta.name
'MONTECARLO.HaulCycleVariability'
>>> meta.version
'1.0.0'
>>> SimulationMetadata(code="", category=SimulationCategory.MONTE_CARLO, description="x")
Traceback (most recent call last):
...
mineproductivity.simulation.exceptions.SimulationValidationError: SimulationMetadata.code must not be empty
MonteCarloModel ¶
Bases: SimulationModel, ABC
The contract a future Monte Carlo plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS (design spec §13, §35's interface-purity proof).
Each call to :meth:_trial is expected to be independent and
reproducible given the same random_seed --
ExperimentRunner (design spec §17) supplies a distinct seed per
trial; the concrete model never manages its own seed state
internally (§29's statelessness rule, §34's recorded anti-pattern).
ExperimentResult
dataclass
¶
ExperimentResult(run_id='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), *, final_state=None, trial_results=())
Bases: SimulationResult
The outcome of one Experiment -- an aggregation, not a
statistical characterization; the actual descriptive/inferential
treatment of trial_results is analytics' job (design spec
§19-§20), not this type's.
Examples:
>>> result = ExperimentResult(trial_results=(SimulationResult(run_id="RUN-1"),))
>>> len(result.trial_results)
1
SimulationResult
dataclass
¶
SimulationResult(run_id='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), *, final_state=None)
Bases: BaseValueObject
The shared envelope every concrete simulation outcome composes
-- mirrors digital_twin.TwinResult's role (spec 08 §25), one
layer up.
Examples:
>>> SimulationResult(run_id="RUN-1").run_id
'RUN-1'
>>> SimulationResult(warnings=("zero trials requested",)).warnings
('zero trials requested',)
RunStatus ¶
Bases: Enum
A SimulationRun's own operational lifecycle -- distinct in
shape from digital_twin.TwinStatus (spec 08 §10), which tracks
a twin instance's synchronization freshness; RunStatus instead
tracks execution progress, with values shaped by what
SimulationExecutor (design spec §10) actually does.
COMPLETED and FAILED are terminal -- no transition leads
out of either (§10's state diagram).
SimulationRun
dataclass
¶
Bases: BaseEntity[str]
The root of one executing or completed simulation --
'Run-as-entity,' following digital_twin.Twin's own precedent
(spec 08 §3.3, §8) exactly: id (inherited) is the run's
identity, and representing a state change means producing a NEW
SimulationRun instance via :meth:with_state, never mutating
fields in place.
Thread safety. SimulationRun instances are immutable --
trivially safe to read and share across threads (design spec §32).
Examples:
>>> from datetime import datetime, timezone
>>> run = SimulationRun(
... id="RUN-1",
... scenario_code="FLEET.NightShiftSurge",
... state=SimulationState(
... attributes={"provisioned": True},
... simulated_time=datetime(2026, 7, 8, tzinfo=timezone.utc),
... ),
... )
>>> run.status
<RunStatus.SCHEDULED: 'scheduled'>
>>> run.with_state(run.state, status=RunStatus.RUNNING).status
<RunStatus.RUNNING: 'running'>
>>> run.status # the original instance is untouched
<RunStatus.SCHEDULED: 'scheduled'>
with_state ¶
Returns a NEW SimulationRun instance with state (and
optionally status) replacing the current ones -- the
dataclasses.replace-style helper core.BaseEntity's own
docstring anticipates, identical to Twin.with_state()
(spec 08 §8).
Scenario
dataclass
¶
Scenario(code, *, version='1.0.0', status=(lambda: ScenarioStatus.PROPOSED)(), model_code, parameters=dict(), time_horizon, initial_state=None, as_of=None)
Bases: BaseValueObject
A named, versioned simulation configuration -- the
business-policy-grade governed artifact this package owns,
analogous to decision.Policy (spec 07 §12) one layer down.
An Active Scenario is a public contract: it is never edited
in place. A changed scenario is published as a new version; the
prior version transitions to Superseded, never silently
repointed (design spec §9, §25, §34's recorded anti-pattern).
Examples:
>>> scenario = Scenario(
... code="FLEET.NightShiftSurge",
... model_code="MONTECARLO.HaulCycleVariability",
... parameters={"trucks_added": 3},
... time_horizon=timedelta(hours=12),
... )
>>> scenario.status
<ScenarioStatus.PROPOSED: 'proposed'>
>>> scenario.version
'1.0.0'
>>> scenario.initial_state is None
True
ScenarioStatus ¶
Bases: Enum
The Scenario lifecycle -- mirrors decision.DecisionStatus
(spec 07 §12) exactly, applied here to governed
simulation-configuration artifacts rather than to business
policies.
SensitivityAnalyzer ¶
Sweeps one or more Scenario parameters across an
Experiment, handing the resulting outcomes to analytics for
distributional treatment -- the sensitivity computation itself is
analytics' job, not this package's, mirroring
ScenarioComparator's identical delegation (design spec §19,
§20).
sweep ¶
Produces one SimulationRun per value in values, each
a copy of base_scenario with parameter overridden --
the resulting Experiment's run_ids are ordered to match
values' order, so a caller can zip them back together for
analytics' own correlation/regression primitives to consume
(design spec §20). Zero values is a legitimately incomplete
input (§28): an empty Experiment is returned, never a
raise.
summarize ¶
Hands a sweep's numeric outcomes to analytics for
distributional treatment -- analytics.distribution() for
shape, analytics.confidence_interval() for the interval
around the mean (design spec §20). This method owns no
arithmetic of its own; both computations, including their own
input validation, are analytics' entirely.
SimulationState
dataclass
¶
Bases: BaseValueObject
One run's simulated attributes and simulated time as of the last
executed step or trial (design spec §10). A frozen value object,
not an entity -- the entity (continuous identity over an
execution's lifetime) is SimulationRun itself; SimulationState
is what a run currently points to, exactly the identity/value
distinction digital_twin.Twin/TwinState already draw one
layer down (spec 08 §12).
simulated_time is simulated, never wall-clock, time -- advanced
by SimulationClock (§11) under whichever TimeProgressionMode
the executing model's category prescribes.
Examples:
>>> from datetime import timezone
>>> state = SimulationState(
... attributes={"queue_len": 4, "tonnes_moved": 1850.0},
... simulated_time=datetime(2026, 7, 8, tzinfo=timezone.utc),
... )
>>> state.attributes["queue_len"]
4
>>> SimulationState(attributes={}, simulated_time=datetime(2026, 7, 8, tzinfo=timezone.utc))
Traceback (most recent call last):
...
mineproductivity.simulation.exceptions.SimulationValidationError: SimulationState.attributes must not be empty
SystemDynamicsModel ¶
Bases: SimulationModel, ABC
The contract a future system-dynamics plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS (design spec §15, §35's interface-purity proof).
register ¶
Register cls into :data:REGISTRY, keyed by cls.meta.code.
Raises:
| Type | Description |
|---|---|
SimulationValidationError
|
If |
SimulationVersionConflictError
|
If |
by_category ¶
A specification satisfied by every run whose published
scenario's registered model belongs to category -- compose with
SimulationRunRepository.list() and &/|/~ freely.
A run whose scenario is unpublished, or whose model is
unregistered, never matches (and never raises, design spec §22).
Examples:
by_scope ¶
A specification satisfied by every run carrying every key/value
pair in scope, resolved against the run's own scenario_code
/status fields first and its open state.attributes mapping
otherwise (a subset match, mirroring digital_twin.by_scope's
semantics one layer down).
Examples:
seed_from_replay ¶
Reconstructs an initial SimulationState from real event
history via EventStore.replay(as_of) -- the concrete mechanism
behind Business Objective 3 (design spec §2): 'simulate forward
from last Tuesday's actual conditions' is seed_from_replay
followed by handing the result to a Scenario.
The reconstructed state is a deterministic census of the replayed
history: events_replayed (total envelope count),
event_counts (per payload-type counts, alphabetized), and --
when any envelope exists -- last_event_time_utc (ISO 8601).
simulated_time is as_of.utc when supplied, else the latest
replayed envelope's event_time_utc.
Raises:
| Type | Description |
|---|---|
SimulationValidationError
|
If |