mineproductivity.digital_twin¶
Auto-generated API reference for the Digital Twin package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Digital Twin.
digital_twin ¶
mineproductivity.digital_twin -- the platform's stateful
representation layer, built directly on top of decision.
Answers a question none of the six packages below it were ever meant to
answer: what is the current, holistic condition of this physical
asset, operation, or process, and how do we keep that understanding
synchronized with reality over time? kpis computes a single
metric's value; analytics characterizes a series statistically;
decision turns a statistical judgment into a recommended action --
all point-in-time or series-oriented computations over
already-recorded facts. digital_twin holds a continuously-updated
virtual counterpart of a mine, a piece of equipment, a plant, a
conveyor, a fleet, or another real-world system, synchronized from the
immutable event stream events already owns, and exposes that
representation -- plus an interface for simulating hypothetical futures
over it -- to whatever consumes it next (design spec §1).
The complete package (design spec §6's fifteen modules) is implemented:
Twin (a core.BaseEntity[str] subclass -- the first package in
the series whose central abstraction is entity-shaped rather than
stateless, §8)/TwinContext; TwinMetadata/TwinCategory
(§26); the eleven category base classes (§9); TwinStatus (§10);
TwinState (§12)/TwinSnapshot (§13); TwinSynchronizer/
SyncPolicy (§11, §15); TelemetryReading (§16); the
interface-only TwinSimulationModel with zero concrete subclasses by
design (§14, ADR-0008); by_category/by_scope discovery
factories (§18); TwinRepository as a literal type alias over
core.BaseRepository[Twin, str] (§20); TwinStateCache (§22);
the TwinResult/SyncResult/TwinSimulationResult family
(§25); REGISTRY/register (§17); and the full exception
hierarchy. It holds no KPI formulas, no statistical computation, no
business-decision logic, no optimization search, and no AI-agent
reasoning -- all of those already exist, one or more layers down, and
are consumed rather than re-implemented (§3).
digital_twin depends on core, ontology, events,
registry, plugins, connectors, kpis, analytics, and
decision -- and MUST NEVER import simulation, optimization,
agents, or visualization, none of which this package may see.
TwinRepository ¶
The storage contract for twin instances, keyed by their own
identity -- a literal type alias over
core.BaseRepository[Twin, str], never a parallel digital-twin
repository ABC (design spec §20, §31's recorded anti-pattern).
Twin
dataclass
¶
Bases: BaseEntity[str], ABC
The root of every registrable twin type -- 'Twin-as-object,' the
direct counterpart of kpis.BaseKPI/analytics.AnalyticsModel/
decision.DecisionModel, one/two/three layers down respectively.
Unlike those three (deliberately stateless, spec 06 §8, spec 07 §8),
Twin subclasses core.BaseEntity[str] directly: id
(inherited) is the twin's identity, and -- per BaseEntity's own
documented contract -- representing a state change means producing
a NEW Twin instance via :meth:with_state, never mutating
fields in place. This mirrors the platform's event-first philosophy
exactly: a twin's state is a projection of the event stream, not a
mutable record (design spec §3.3).
scope is set once at provisioning time and is immutable
thereafter (frozen into a read-only mapping at construction) -- a
twin that starts representing a different real-world instance is a
new Twin (a new id), never the same twin re-scoped in place
(design spec §9, §31).
Thread safety. Twin instances are immutable -- trivially
safe to read and share across threads; no locking is ever needed to
read a twin's state/status (design spec §29). A concrete
_apply implementation MUST itself be safe to call concurrently
for different id\ s on any shared collaborators it closes
over (§30).
with_state ¶
Returns a NEW Twin instance with state (and
optionally status) replacing the current ones -- the
dataclasses.replace-style helper core.BaseEntity's own
docstring anticipates for representing a state change.
TwinContext ¶
Bundles the collaborators and evidence a Twin's _apply
may need -- the digital-twin-layer counterpart to
decision.DecisionContext (spec 07 §8), one layer up, extended
with decision_results since digital_twin is the first
package to consume Decision's own outputs directly (design spec §8).
Examples:
>>> class _FakeStore: ...
>>> context = TwinContext(event_store=_FakeStore())
>>> context.kpi_results
()
>>> context.as_of is None
True
TwinStateCache ¶
Caches the TwinContext evidence gathered for one
(twin_id, as_of) key (design spec §22).
Thread safety. Reads/writes are keyed by (twin_id, 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 §30) -- the same one-lock posture
decision.DecisionAuditTrail already takes for its own narrow
mutable surface.
Examples:
>>> from datetime import datetime, timezone
>>> class _FakeStore: ...
>>> cache = TwinStateCache()
>>> as_of = AsOf(utc=datetime(2026, 7, 8, tzinfo=timezone.utc))
>>> cache.get("CONV-7", as_of) is None
True
>>> context = TwinContext(event_store=_FakeStore())
>>> cache.put("CONV-7", as_of, context)
>>> cache.get("CONV-7", as_of) is context
True
ConveyorTwin
dataclass
¶
EquipmentTwin
dataclass
¶
Bases: Twin, ABC
A single piece of equipment -- scope typically names one
ontology.RigidHaulTruck/HydraulicShovel/etc. instance.
FleetTwin
dataclass
¶
GeologicalTwin
dataclass
¶
HaulageTwin
dataclass
¶
Bases: Twin, ABC
The haulage system (routes, cycle state) distinct from any one
EquipmentTwin -- an aggregate view over a haul fleet's
operation.
MineTwin
dataclass
¶
Bases: Twin, ABC
A whole-mine virtual counterpart -- aggregates state across every other twin category scoped to the same mine.
PlantTwin
dataclass
¶
Bases: Twin, ABC
A processing/beneficiation plant as a whole, distinct from
ProcessingPlantTwin's narrower processing-circuit framing.
ProcessingPlantTwin
dataclass
¶
Bases: Twin, ABC
A specific processing circuit's live state (throughput, recovery
inputs) -- narrower than PlantTwin's whole-plant framing.
ProductionTwin
dataclass
¶
Bases: Twin, ABC
A production system's aggregate live state, distinct from any
one EquipmentTwin/ConveyorTwin feeding it.
StockpileTwin
dataclass
¶
VentilationTwin
dataclass
¶
TwinNotFoundError ¶
Bases: NotFoundError
TwinRepository.get(twin_id) found no twin for that id, or
REGISTRY.get(code) found no registered Twin type for that
code (design spec §6).
TwinStateConflictError ¶
Bases: MineProductivityError
Two concurrent synchronize() calls for the same twin_id
raced past this package's per-id write serialization (design spec
§29) -- raised only if that serialization contract itself is
violated by a non-conforming TwinRepository implementation,
never under normal operation with the reference in-memory
repository.
TwinSyncError ¶
Bases: MineProductivityError
Twin._apply() raised for a batch of events that should have
been structurally valid -- distinct from a legitimately-empty-input
case (design spec §8's 'qualify, don't coerce' rule), which returns
a SyncResult carrying a warning instead of raising (§24).
TwinValidationError ¶
Bases: ValidationError
A TwinMetadata, TwinState, or TwinSnapshot failed
validation (design spec §26, §12, §13) -- e.g. an empty code, or
a TwinState with an empty attributes mapping where the twin
category requires at least one attribute.
TwinVersionConflictError ¶
Bases: RegistrationError
A plugin attempted to re-register an existing Twin type code
with materially different metadata without a version bump, mirroring
decision.DecisionVersionConflictError (spec 07 §6) and
analytics.AnalyticsVersionConflictError (spec 06 §6).
TwinStatus ¶
Bases: Enum
The twin INSTANCE's own operational lifecycle -- distinct from
TwinMetadata.version's type-level SemVer (design spec §21) and
from Policy-style governance lifecycles
(decision.DecisionStatus, spec 07 §12) that do not apply here at
all, since a Twin instance is not a governed business artifact.
RETIRED is terminal -- a retired twin's id is never reused
for a different real-world thing, mirroring the "a retired
identifier is never reused for a new meaning" rule kpis already
established for KPI codes (spec 05 §20). :class:~mineproductivity.digital_twin.synchronization.TwinSynchronizer
refuses to fold further events into a retired twin (§10, §24).
TwinCategory ¶
Bases: Enum
Closed enum -- adding a member is a governance-reviewed change,
mirroring decision.DecisionCategory's/analytics.AnalyticsCategory's
closed-enum rule (spec 07 §30, spec 06 §31).
TwinMetadata
dataclass
¶
TwinMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, version='1.0.0')
Bases: BaseMetadata
The minimal registration schema for a discoverable Twin type
(design spec §26). code names a type (e.g.
"CONVEYOR.Standard"), never an instance -- Twin.id (§8,
inherited from core.BaseEntity) identifies "which specific
conveyor," while code identifies "what kind of conveyor twin is
this." Two twin instances scoped to different assets share the same
code but never the same id.
BaseMetadata.name has no default upstream and a Twin 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 DecisionMetadata/
AnalyticsMetadata already established.
Examples:
>>> meta = TwinMetadata(
... code="CONVEYOR.Standard", category=TwinCategory.CONVEYOR,
... description="A standard conveyor system twin.",
... )
>>> meta.code
'CONVEYOR.Standard'
>>> meta.name
'CONVEYOR.Standard'
>>> meta.version
'1.0.0'
>>> TwinMetadata(code="", category=TwinCategory.CONVEYOR, description="x")
Traceback (most recent call last):
...
mineproductivity.digital_twin.exceptions.TwinValidationError: TwinMetadata.code must not be empty
SyncResult
dataclass
¶
SyncResult(previous_status, new_status, events_applied, *, twin_id='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: TwinResult
The outcome of one TwinSynchronizer.synchronize() call
(design spec §11, §25). warnings (inherited) is the primary
"why didn't this twin's state change the way I expected" signal
(§24) -- a legitimately-empty or legitimately-unchanged sync returns
a warning-carrying SyncResult, never raises.
Examples:
>>> result = SyncResult(
... twin_id="CONV-7",
... previous_status=TwinStatus.PROVISIONED,
... new_status=TwinStatus.SYNCHRONIZED,
... events_applied=3,
... )
>>> result.events_applied
3
TwinResult
dataclass
¶
Bases: BaseValueObject
The shared envelope every concrete digital-twin outcome composes
-- mirrors decision.DecisionResult's role (spec 07 §28), one
layer up.
Examples:
>>> TwinResult(twin_id="CONV-7").twin_id
'CONV-7'
>>> TwinResult(warnings=("no events to apply",)).warnings
('no events to apply',)
TwinSimulationResult
dataclass
¶
TwinSimulationResult(hypothesis, predicted_state, *, twin_id='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())
Bases: TwinResult
The outcome of one TwinSimulationModel._simulate() call -- a
stub-shaped result type defined now even though no producer of it
ships in this release (design spec §14, §25), exactly as
decision.WhatIfResult/analytics.ForecastResult were defined
ahead of any concrete implementation (spec 07 §19, spec 06 §16).
Examples:
>>> from datetime import datetime, timezone
>>> result = TwinSimulationResult(
... twin_id="CONV-7",
... hypothesis={"belt_speed_mps": 3.5},
... predicted_state=TwinState(
... attributes={"belt_speed_mps": 3.5},
... captured_at=datetime(2026, 7, 8, tzinfo=timezone.utc),
... ),
... )
>>> result.hypothesis["belt_speed_mps"]
3.5
TwinSimulationModel ¶
Bases: ABC
The contract a future simulation/optimization/agents
plugin implements to answer 'what would this twin look like under a
hypothetical change.' THIS MODULE SHIPS NO CONCRETE SUBCLASS
(design spec §14, §32's interface-purity proof).
This interface is the digital-twin-layer counterpart of
decision.WhatIfEngine (spec 07 §19), which was itself already
designed with digital_twin named as its most direct anticipated
implementer. A future concrete decision.WhatIfEngine
implementation is expected to be provided BY a
digital_twin-aware plugin that internally calls a concrete
TwinSimulationModel and translates its TwinSimulationResult
into a decision.WhatIfResult -- this package defines the
twin-state-level half of that bridge; spec 07 already defined the
business-decision-level half. TwinSimulationModel operates on
Twin/TwinState directly (a physical/operational-state
question), while decision.WhatIfEngine operates on
DecisionContext (a business-evidence question) -- related,
potentially composed together, never collapsed into one interface
(design spec §14).
TwinSnapshot
dataclass
¶
Bases: BaseValueObject
A point-in-time capture of one Twin, for audit, historical
query, and simulation-forking use -- distinct from TwinState
(design spec §12), which represents only the current condition.
Examples:
>>> from datetime import datetime, timezone
>>> snapshot = TwinSnapshot(
... twin_id="CONV-7",
... state=TwinState(
... attributes={"belt_speed_mps": 3.1},
... captured_at=datetime(2026, 7, 8, tzinfo=timezone.utc),
... ),
... status=TwinStatus.SYNCHRONIZED,
... as_of=AsOf(utc=datetime(2026, 7, 8, tzinfo=timezone.utc)),
... )
>>> snapshot.twin_id
'CONV-7'
>>> snapshot.status
<TwinStatus.SYNCHRONIZED: 'synchronized'>
TwinState
dataclass
¶
Bases: BaseValueObject
One twin's condition as of the moment it was last synchronized.
A frozen value object, not an entity -- the entity (continuous
identity over time) is Twin itself (design spec §8);
TwinState is what a Twin currently points to, exactly the
identity/value distinction core.entity/core.value_object
already draw platform-wide.
schema_version (§21's third versioning axis) is the
shape-version of attributes' contents, carried with the value
itself -- never inferred from TwinMetadata.version at read time
-- so a future _apply revision that changes which keys it
populates can detect and migrate an older TwinState it reads
back from a TwinRepository/TwinSnapshot.
Examples:
>>> from datetime import timezone
>>> state = TwinState(
... attributes={"belt_speed_mps": 3.1, "load_tph": 850.0},
... captured_at=datetime(2026, 7, 8, tzinfo=timezone.utc),
... )
>>> state.attributes["belt_speed_mps"]
3.1
>>> state.schema_version
'1.0.0'
>>> TwinState(attributes={}, captured_at=datetime(2026, 7, 8, tzinfo=timezone.utc))
Traceback (most recent call last):
...
mineproductivity.digital_twin.exceptions.TwinValidationError: TwinState.attributes must not be empty
SyncPolicy ¶
Declares how a Twin should be kept current: real-time
(subscribe to EventBus) vs. scheduled-pull (periodic
EventStore.query), and the EventFilter selecting which
events matter to this twin (design spec §11). The filter is
expected to narrow delivery to only the event types a given twin
category's _apply actually reads -- never a blanket
"every event" subscription (§33).
Examples:
>>> from mineproductivity.core import PredicateSpecification
>>> policy = SyncPolicy(
... mode="realtime",
... event_filter=PredicateSpecification(lambda envelope: True),
... )
>>> policy.mode
'realtime'
>>> policy.poll_interval is None
True
TwinSynchronizer ¶
Orchestrates one Twin's update: fetches the current instance
from a TwinRepository, folds new events through Twin._apply,
writes the resulting new instance back (the per-id swap and its
concurrency contract live in the repository, design spec §29), and
returns a SyncResult -- the digital-twin-layer counterpart of
decision.RealTimeDecisionSession/analytics.StreamingAnalyticsSession's
orchestration role (spec 07 §25, spec 06 §27), one layer up.
:meth:synchronize never mutates the Twin instance it reads --
it computes a replacement via with_state() and hands that
replacement to the repository (§11). The optional cache stores
the evidence inputs (TwinContext) per (twin_id, as_of) so
closely-spaced cycles can skip re-assembly; it is never treated as
authoritative for the twin's current state -- only the repository
is (§22, §31).
synchronize ¶
Fold events into the twin stored under twin_id.
A legitimately-empty events batch, or a batch whose fold
leaves the state value-unchanged, returns a warning-carrying
SyncResult -- never raises (design spec §24). A retired
twin is never folded further: Retired is terminal (§10).
Raises:
| Type | Description |
|---|---|
TwinNotFoundError
|
If no twin is stored under |
TwinSyncError
|
If |
TwinStateConflictError
|
If the repository's per-id write serialization contract (design spec §29) was violated by a concurrent writer mid-swap -- never under normal operation. |
TelemetryReading
dataclass
¶
Bases: BaseValueObject
The twin-local shape one already-ingested, already-event-sourced
telemetry observation takes once _apply reads it out of an
event's payload -- not a new ingestion contract (design spec §16).
Examples:
>>> from datetime import timezone
>>> reading = TelemetryReading(
... sensor_id="CONV-7.belt_speed", value=3.1, unit="m/s",
... observed_at=datetime(2026, 7, 8, tzinfo=timezone.utc),
... )
>>> reading.value, reading.unit
(3.1, 'm/s')
register ¶
Register cls into :data:REGISTRY, keyed by cls.meta.code.
Raises:
| Type | Description |
|---|---|
TwinValidationError
|
If |
TwinVersionConflictError
|
If |
by_category ¶
A specification satisfied by every twin whose type's
meta.category is category -- compose with
TwinRepository.list() and &/|/~ freely.
Examples:
by_scope ¶
A specification satisfied by every twin whose own scope
carries every key/value pair in scope (a subset match, mirroring
decision.DecisionAuditTrail.query()'s scope-filter semantics one
layer down).
Examples: