Domain Model Reference¶

Machina's domain model is the backbone of the framework. Every connector normalizes external data into these entities, and every agent reasons in these terms. The model is aligned with ISO 14224 — the international standard for reliability and maintenance data collection — so teams already using ISO-aligned CMMS tooling can interoperate without custom field mapping.

The entities are pydantic v2 models with validators. Fields carrying ISO 14224 codes (iso_14224_code, equipment_class_code, failure_impact) are optional and live alongside Machina's own catalog identifiers — you can adopt the ISO alignment incrementally, one field at a time.

Core Entities¶

Asset¶

Asset ¶

Bases: BaseModel

A physical piece of equipment in the maintenance hierarchy.

Assets represent any maintainable item — from an entire plant down to an individual bearing. They carry metadata for identification, location, and maintenance strategy.

AssetType ¶

Bases: StrEnum

Pragmatic equipment classification inspired by ISO 14224 and common industry usage.

This enum uses Machina-specific buckets that blend ISO 14224 equipment categories (rotating, electrical, safety) with common industry terms (static, instrument, piping, structural, HVAC). For strict ISO 14224 alignment, use Asset.equipment_class_code to carry the Annex A Table A.4 code (e.g. 'PU', 'CO', 'HE', 'EM').

Criticality ¶

Bases: StrEnum

Asset criticality classification for maintenance prioritisation.

A = Critical (production-stopping), B = Important, C = Standard.

WorkOrder¶

WorkOrder ¶

Bases: BaseModel

A maintenance work order with lifecycle tracking.

Supports status transitions, SLA tracking, and links to the asset, failure mode, and required spare parts.

transition_to ¶

transition_to(new_status: WorkOrderStatus) -> None

Transition the work order to a new status.

Raises:

Type	Description
`ValueError`	If the transition is not allowed.

WorkOrderType ¶

Bases: StrEnum

Classification of the maintenance activity.

WorkOrderStatus ¶

Bases: StrEnum

Lifecycle state of a work order.

Priority ¶

Bases: StrEnum

Work order urgency level.

FailureImpact ¶

Bases: StrEnum

Failure impact on equipment function (ISO 14224 Table 6).

CRITICAL: loss of equipment function (production-stopping)
DEGRADED: equipment still functioning but below specification
INCIPIENT: imminent failure detected before functional loss

SparePartRequirement ¶

Bases: BaseModel

A spare part needed for a work order.

FailureMode¶

FailureMode ¶

Bases: BaseModel

A known failure mode for a class of equipment.

Encodes domain knowledge about how equipment fails, how to detect the failure, and what to do about it.

SparePart¶

SparePart ¶

Bases: BaseModel

A spare part with inventory and reorder tracking.

Links to compatible assets for automatic part lookup during work order creation.

needs_reorder `property` ¶

needs_reorder: bool

Whether current stock is at or below the reorder point.

Alarm¶

Alarm ¶

Bases: BaseModel

A sensor alarm or event from an industrial data source.

Represents a threshold exceedance or anomalous condition detected by the IoT / SCADA layer.

Example

from machina.domain.alarm import Alarm, Severity

alarm = Alarm(
    id="ALM-001",
    asset_id="P-201",
    severity=Severity.WARNING,
    parameter="vibration",
    value=12.5,
    threshold=10.0,
    unit="mm/s",
)

is_above_threshold `property` ¶

is_above_threshold: bool

Whether the measured value exceeds the threshold.

Severity ¶

Bases: StrEnum

Alarm severity level.

MaintenancePlan¶

MaintenancePlan ¶

Bases: BaseModel

A preventive maintenance plan defining recurring tasks for an asset.

Links an asset to a set of inspection or service tasks performed at a fixed calendar or operating-hours interval.

Example

from machina.domain.maintenance_plan import Interval, MaintenancePlan

plan = MaintenancePlan(
    id="MP-P201-Q",
    asset_id="P-201",
    name="Quarterly Bearing Inspection",
    interval=Interval(months=3),
    tasks=["Check vibration levels", "Inspect seal condition"],
)

Interval ¶

Bases: BaseModel

A maintenance interval specification.

At least one of the duration fields should be set.

total_days `property` ¶

total_days: int

Approximate interval in calendar days (months ≈ 30 days).

Plant¶

Plant ¶

Bases: BaseModel

Top-level container representing a physical plant or site.

Acts as the root of the asset hierarchy and provides lookup methods for navigating the equipment tree.

register_asset ¶

register_asset(asset: Asset) -> None

Add an asset to the plant registry.

get_asset ¶

get_asset(asset_id: str) -> Asset

Look up an asset by ID.

Raises:

Type	Description
`AssetNotFoundError`	If no asset with the given ID exists.

list_assets ¶

list_assets() -> list[Asset]

Return all registered assets.

load_assets_from `async` ¶

load_assets_from(source: Any) -> int

Load assets into the plant from a connector or a file path.

Parameters:

Name	Type	Description	Default
`source`	`Any`	Either a connector exposing an async `read_assets()` method, or a path (`str` or `Path`) to a JSON / YAML file containing a list of asset dicts.	required

Returns:

Type	Description
`int`	The number of assets loaded.

Raises:

Type	Description
`TypeError`	If `source` is neither a connector nor a path.
`FileNotFoundError`	If the file path does not exist.
`ValueError`	If the file extension is unsupported or the file does not contain a list of asset dicts.

Domain Services¶

FailureAnalyzer¶

FailureAnalyzer ¶

FailureAnalyzer(failure_modes: list[FailureMode] | None = None)

Service that matches symptoms to known failure modes.

This is a rule-based baseline; future versions will integrate LLM-assisted reasoning and historical failure data.

register_failure_mode ¶

register_failure_mode(fm: FailureMode) -> None

Add a failure mode to the knowledge base.

diagnose ¶

diagnose(alarms: list[Alarm], **kwargs: Any) -> list[FailureMode]

diagnose(alarms: None = None, **kwargs: Any) -> DiagnosisResult

diagnose(alarms: list[Alarm] | None = None, **kwargs: Any) -> list[FailureMode] | DiagnosisResult

Return failure modes whose indicators match the alarm parameters.

Can be called two ways:

Programmatic: diagnose(alarms=[alarm1, alarm2]) — returns a plain list[FailureMode] for Python callers.
From workflow engine: diagnose(parameter="vibration_velocity_mm_s", value="7.8", asset_id="P-201", severity="warning") — returns a :class:DiagnosisResult so downstream workflow steps can extract a primary_code (str) for WorkOrder.failure_mode via the template {analyze_alarm.primary_code}.

Parameters:

Name	Type	Description	Default
`alarms`	`list[Alarm] \| None`	Active alarms to match against failure modes.	`None`
`**kwargs`	`Any`	Workflow trigger fields (parameter, value, asset_id, severity).	`{}`

Returns:

Type	Description
`list[FailureMode] \| DiagnosisResult`	Either a list of matching :class:`FailureMode` objects
`list[FailureMode] \| DiagnosisResult`	(programmatic path) or a :class:`DiagnosisResult` (workflow
`list[FailureMode] \| DiagnosisResult`	path) — the workflow wrapper exposes the same ranked dicts
`list[FailureMode] \| DiagnosisResult`	via iteration plus convenience attributes for templating.

WorkOrderFactory¶

WorkOrderFactory ¶

Factory for creating pre-filled work orders from alarms or plans.

Future versions will use the failure analyzer and spare-part inventory to auto-populate fields.

create ¶

create(*, id: str = '', asset_id: str = '', type: WorkOrderType = WorkOrderType.CORRECTIVE, priority: Priority = Priority.MEDIUM, description: str = '', failure_mode: str | None = None, **kwargs: Any) -> WorkOrder

Create a work order with sensible defaults.

Parameters:

Name	Type	Description	Default
`id`	`str`	Work order identifier. Auto-generated as `WO-AUTO-<hex>` when not supplied — required because `WorkOrder.id` is validated non-empty.	`''`
`asset_id`	`str`	Target asset.	`''`
`type`	`WorkOrderType`	Maintenance type.	`CORRECTIVE`
`priority`	`Priority`	Urgency level.	`MEDIUM`
`description`	`str`	Free-text summary.	`''`
`failure_mode`	`str \| None`	Optional failure mode code.	`None`

Returns:

Type	Description
`WorkOrder`	A new `WorkOrder` instance.

create_batch ¶

create_batch(*, work_orders: list[dict[str, Any]] | None = None, **kwargs: Any) -> list[WorkOrder]

Create multiple work orders in one call.

Each item in work_orders is a dict of keyword arguments forwarded to :meth:create.

Parameters:

Name	Type	Description	Default
`work_orders`	`list[dict[str, Any]] \| None`	List of per-WO keyword-argument dicts.	`None`

Returns:

Type	Description
`list[WorkOrder]`	List of created `WorkOrder` instances.

MaintenanceScheduler¶

MaintenanceScheduler ¶

MaintenanceScheduler(*, plans: list[MaintenancePlan] | None = None, last_executed: dict[str, date] | None = None)

Service for computing maintenance due dates.

This is a calendar-based baseline; future versions will account for operating hours and condition-based triggers.

Parameters:

Name	Type	Description	Default
`plans`	`list[MaintenancePlan] \| None`	Optional list of maintenance plans to manage.	`None`
`last_executed`	`dict[str, date] \| None`	Optional mapping of plan ID to last execution date.	`None`

next_due_date ¶

next_due_date(plan: MaintenancePlan, last_executed: date) -> date

Calculate the next due date for a maintenance plan.

Parameters:

Name	Type	Description	Default
`plan`	`MaintenancePlan`	The maintenance plan.	required
`last_executed`	`date`	Date the plan was last executed.	required

Returns:

Type	Description
`date`	The next calendar date the maintenance is due.

is_overdue ¶

is_overdue(plan: MaintenancePlan, last_executed: date, today: date | None = None) -> bool

Check whether a maintenance plan is overdue.

Parameters:

Name	Type	Description	Default
`plan`	`MaintenancePlan`	The maintenance plan.	required
`last_executed`	`date`	Date the plan was last executed.	required
`today`	`date \| None`	Reference date (defaults to today).	`None`

Returns:

Type	Description
`bool`	True if the plan is past its due date.

find_window ¶

find_window(*, asset_id: str = '', **kwargs: Any) -> dict[str, Any]

Find the next available maintenance window for an asset.

Returns a suggested window based on the asset's maintenance plans and last execution dates. In future versions this will integrate with calendar connectors to check production schedules.

Parameters:

Name	Type	Description	Default
`asset_id`	`str`	Target asset identifier.	`''`

Returns:

Type	Description
`dict[str, Any]`	A dict with `start`, `end`, `plan_id` and `asset_id`.

scan_due_plans ¶

scan_due_plans(*, horizon_days: int | str = 14, **kwargs: Any) -> list[dict[str, Any]]

Return maintenance plans due within the given horizon.

Parameters:

Name	Type	Description	Default
`horizon_days`	`int \| str`	Number of days to look ahead (accepts str for workflow template compatibility).	`14`

Returns:

Type	Description
`list[dict[str, Any]]`	A list of dicts with plan details and due dates.

Domain Model Reference¶

Core Entities¶

Asset¶

Asset ¶

AssetType ¶

Criticality ¶

WorkOrder¶

WorkOrder ¶

transition_to ¶

WorkOrderType ¶

WorkOrderStatus ¶

Priority ¶

FailureImpact ¶

SparePartRequirement ¶

FailureMode¶

FailureMode ¶

SparePart¶

SparePart ¶

needs_reorder property ¶

Alarm¶

Alarm ¶

is_above_threshold property ¶

Severity ¶

MaintenancePlan¶

MaintenancePlan ¶

Interval ¶

total_days property ¶

Plant¶

Plant ¶

register_asset ¶

get_asset ¶

list_assets ¶

load_assets_from async ¶

Domain Services¶

FailureAnalyzer¶

FailureAnalyzer ¶

register_failure_mode ¶

diagnose ¶

WorkOrderFactory¶

WorkOrderFactory ¶

create ¶

create_batch ¶

MaintenanceScheduler¶

MaintenanceScheduler ¶

next_due_date ¶

is_overdue ¶

find_window ¶

scan_due_plans ¶

needs_reorder `property` ¶

is_above_threshold `property` ¶

total_days `property` ¶

load_assets_from `async` ¶