Architecture Overview

MineProductivity is an open-source Python framework for mine-productivity analytics — think scikit-learn, but for how a mine produces. You do not need a mining background to use it: the domain is modelled for you as typed Python objects, and you work with a familiar, discoverable API.

Who this is for
A software engineer who has never set foot on a mine. Every mining term below is defined the first time it appears, and the running examples use realistic data from a fictional copper open-pit, “Bingham West”.

The one mental model

Everything in MineProductivity is one closed loop. A truck completes a haul; that becomes an event; events feed KPIs (the metrics that matter); analytics and AI turn KPIs into a recommendation; a human or agent acts; the action changes the mine, producing new events. Learn this picture and the rest of the docs are just detail.

Canonical end-to-end data flow
The canonical data flow. Every stage maps to a part of the API you will use.

A 60-second mining primer

TermWhat it means (for an engineer)
CycleOne round-trip of a haul truck: queue → spot → load → haul → dump → return. Emits a CycleEvent.
PayloadTonnes carried in one cycle (a CAT 793 hauls ~220 t).
TPHTonnes per hour — the headline production rate.
Availability (PA)Fraction of calendar time a machine was not broken/in-maintenance.
Utilisation (UA)Fraction of available time it was actually working.
DelayLost productive time (crusher down, queueing, refuelling). Emits a DelayEvent.
Match factorAre there enough trucks for the shovel? ~1.0 is balanced.
Recovery% of valuable metal the plant extracts from the ore.
Digital maturityHow instrumented a site is: L1 (paper) → L2 (fleet system) → L3 (analytics) → L4 (autonomous).

Two ideas that explain every design choice

1. Ontology-first. The mining domain is modelled as a hierarchy of typed objects (the ontology) before any computation. 2. KPI-as-object. Every metric is a class that knows its own formula, units, and data needs — TonnesPerHour(), not calculate_tph(df). Adding a metric means adding an object, never editing an engine.

The MineProductivity ontology stack
The ontology stack. The KPI ontology is the backbone; everything else feeds it or is gated by it.
Where to go next
New here? Read InstallationQuick StartYour First KPI. Coming from pandas/scikit-learn? The Python SDK will feel immediately familiar.

Installation

MineProductivity requires Python 3.11+. Install from PyPI:

$ pip install mineproductivity

Optional extras pull in heavier dependencies only when you need them:

$ pip install "mineproductivity[analytics]"   # pandas, polars, forecasting
$ pip install "mineproductivity[graph]"       # knowledge-graph + embeddings
$ pip install "mineproductivity[api]"         # REST/GraphQL server
$ pip install "mineproductivity[all]"         # everything

OEM connectors are separate plugins, so the core stays vendor-neutral and lightweight:

$ pip install mineproductivity-dispatch     # a fleet-management-system adapter

Verify the install:

$ mineprod --version
mineproductivity 1.0.0
$ python -c "import mineproductivity as mp; print(len(mp.kpis.REGISTRY), 'KPIs registered')"
10000 KPIs registered
No mining data yet?
Every example in these docs runs on the bundled sample dataset (mineproductivity.datasets.bingham_west()), so you can follow along with nothing but a fresh install.

Quick Start

From zero to your first productivity number in ten lines.

The Mine object is your entry point — analogous to a scikit-learn estimator or a pandas DataFrame. You describe the site, load some data, and ask for KPIs by their stable code.

from mineproductivity import Mine

# 1. Describe the operation (copper, open-pit, fleet-management-system maturity)
mine = Mine(site="Bingham West", commodity="copper",
            method="open_pit", maturity=2)

# 2. Ingest a shift of data from a connector (here, the bundled CSV sample)
mine.ingest(connector="csv", path="bingham_west_shift_A.csv")

# 3. Ask for a KPI by its code — returned as an object, not a bare number
tph = mine.kpi("PROD.TPH").by("shift")
print(tph.value, tph.unit)        # 1212.1 t/h

# 4. Discover what else this site can compute (maturity/commodity-aware)
for cls in mine.available_kpis():
    print(cls.meta.code, "—", cls.meta.name)
# PROD.TPH  — Tonnes per Hour
# FLEET.PA  — Physical Availability
# FLEET.UA  — Use of Availability
# EQUIP.TRUCK.CYCLE — Truck Cycle Time  …

That is the whole loop in miniature: events in, KPI objects out, everything discoverable. The sequence below shows what happened under the hood when you called .by("shift").

KPI compute sequence
What a KPI computation does internally — the engine executes the KPI object over stored events and returns a traceable result.
Key idea
A KPI result is an object (KPIResult) that carries its value, unit, the number of source records used (n), and any warnings. It is always traceable back to the events that produced it.

Your First KPI

KPIs are the heart of MineProductivity. Every metric is a class that inherits a category, carries complete metadata, and registers itself so the whole system (API, CLI, docs, dashboards, AI) can discover it. Let’s read a built-in one, then compute it by hand.

Inspecting a built-in KPI

from mineproductivity.kpis import describe, REGISTRY

meta = describe("PROD.TPH")
print(meta.formula)        # sum(payload_t) / sum(operating_h)
print(meta.unit)           # t/h
print(meta.required_columns) # ('payload_t', 'operating_h')
print(meta.direction)      # Direction.HIGHER_IS_BETTER

Computing it directly

Every KPI exposes .compute(rows), where rows is a sequence of mappings containing the columns the KPI declares it needs. The result is a KPIResult.

from mineproductivity.kpis import REGISTRY

rows = [{"payload_t": 36000, "operating_h": 29.7}]   # one shift, fleet total
result = REGISTRY["PROD.TPH"]().compute(rows, window="shift")
print(result.value, result.unit, "| n =", result.n)
# 1212.121... t/h | n = 1
Why ratios are re-derived, not averaged
PROD.TPH declares aggregation=RATIO, so over multiple periods it is computed as sum(payload)/sum(hours) — never the mean of per-period rates. Averaging percentages is the single most common KPI error in mining; the metadata prevents it for you.

Filtering KPIs to your site

Not every KPI applies to every site. available_kpis() filters by digital maturity, commodity, and mining method:

from mineproductivity.kpis import available_kpis, DigitalMaturity, Commodity, MiningMethod

kpis = available_kpis(DigitalMaturity.L1_MANUAL, Commodity.GOLD, MiningMethod.OPEN_PIT)
print(sorted(c.meta.code for c in kpis))
# ['DELAY.HOURS', 'FLEET.PA', 'PROD.TPH', 'QUAL.RECOVERY']
# Cycle time & match factor need L2 telemetry, so they are correctly withheld.

Ready to build your own? Jump to Writing Custom KPIs.

Your First Connector

A connector is an adapter that reads a data source (a fleet-management system, a CSV export, a database) and yields canonical Event objects. Connectors are the only place vendor-specific knowledge lives — the rest of the framework never sees a proprietary code.

Using the bundled CSV connector

from mineproductivity.connectors import get_connector

Conn = get_connector("csv")
conn = Conn(path="bingham_west_shift_A.csv", shift_id="A-2026-06-25")
cycles = list(conn.get_cycle_data())
print(len(cycles), "cycle events", "| first payload =", cycles[0].payload_t)
# 842 cycle events | first payload = 220.0

What a connector must implement

Every connector implements the FMSConnector abstract base class. That is the entire contract:

from abc import ABC, abstractmethod
from typing import Iterable
from mineproductivity.core.events import CycleEvent, DelayEvent

class FMSConnector(ABC):
    name: str = "abstract"
    @abstractmethod
    def get_cycle_data(self, since: int, until: int) -> Iterable[CycleEvent]: ...
    @abstractmethod
    def get_delay_data(self, since: int, until: int) -> Iterable[DelayEvent]: ...
Connectors never compute
A connector’s only job is to normalise — map the source’s fields and status codes onto canonical events. It must never import the kpis/ package. This keeps the dependency graph clean and the core vendor-neutral.

To build a real OEM adapter and ship it as an installable plugin, see Plugin Development.

Your First Ontology

The ontology is the typed model of the mining domain: sites, pits, equipment, shifts, plans, and so on. You rarely build one from scratch — the framework ships a complete ontology — but understanding it unlocks everything else.

Walking the entity tree

from mineproductivity import Mine

mine = Mine.load("Bingham West")
print(mine.enterprise)                 # Enterprise(id='ent-001', name='Rio Example Corp')
print([p.id for p in mine.pits])         # ['pit-west', 'pit-south']
print(mine.pits[0].commodity)         # Commodity.COPPER
print([e.id for e in mine.equipment][:3]) # ['HT-201', 'HT-202', 'SH-04']

Every entity is metadata-rich

Like KPIs, ontology entities are first-class objects with a stable id, a parent, and a JSON-Schema expression. This is why dashboards, validation, and AI can all read the same model.

truck = mine.equipment_by_id("HT-201")
print(truck.type)                # RigidHaulTruck
print(truck.rated_capacity)      # 220.0  (tonnes)
print(truck.supported_kpis)      # ('EQUIP.TRUCK.CYCLE', 'FLEET.PA', 'HAUL.TKPH', ...)
print(truck.to_schema()["required"]) # ['equipment_id', 'type', 'fleet_id']

The ontology is organised into ~35 sub-ontologies (enterprise, geology, planning, dispatch, …). To extend it, see Creating Equipment Types, Commodities, and Mining Methods.

Creating Equipment Types

An equipment type declares what a class of machine knows about itself: which KPIs apply, what states it can be in, what sensors it emits, and its failure modes. Define one by subclassing EquipmentType.

from mineproductivity.ontology.assets import EquipmentType, register_equipment

@register_equipment
class WaterTruck(EquipmentType):
    code = "WATER_TRUCK"
    supported_kpis = ("FLEET.PA", "FLEET.UA", "COST.FUEL_PER_T")
    operational_states = ("operating", "standby", "maint", "refill")
    sensors = ("gps", "tank_level", "fuel_rate")
    failure_modes = ("pump_failure", "spray_bar_block")
    rated_capacity = 90_000   # litres

Once registered, the type is available everywhere — instantiate equipment of it, and the framework automatically offers only the KPIs the type supports.

wt = mine.add_equipment(id="WT-09", type="WATER_TRUCK", fleet_id="anc-1")
print(wt.type.supported_kpis)   # ('FLEET.PA', 'FLEET.UA', 'COST.FUEL_PER_T')
Pattern
Equipment types use the same register-by-subclassing pattern as KPIs and connectors. If you have learned one extension point in MineProductivity, you have learned them all.

Creating Commodities

A commodity (copper, gold, coal…) activates additional KPIs and quality parameters. Adding one is declarative: register the commodity and tag the KPIs that should switch on for it.

from mineproductivity.ontology.commodity import register_commodity
from mineproductivity.kpis import register, KPIMetadata, QualityKPI, Direction, Aggregation, DigitalMaturity

register_commodity("lithium", quality_params=("li2o_grade", "fe2o3", "mica"))

@register
class SpodumeneRecovery(QualityKPI):
    meta = KPIMetadata(
        code="QUAL.LI.RECOVERY", name="Spodumene Recovery", category="Quality",
        description="Li2O recovery to concentrate (two-product formula).",
        formula="c*(f-t) / (f*(c-t)) * 100", unit="%",
        direction=Direction.HIGHER_IS_BETTER, aggregation=Aggregation.WEIGHTED_MEAN,
        required_columns=("feed_grade", "conc_grade", "tail_grade"),
        commodities=("lithium",), min_maturity=DigitalMaturity.L1_MANUAL)
    def _compute(self, rows):
        r = rows[-1]; f, c, t = r["feed_grade"], r["conc_grade"], r["tail_grade"]
        d = f * (c - t)
        return None if d == 0 else c * (f - t) / d * 100

A site created with Mine(commodity="lithium") now sees QUAL.LI.RECOVERY in its catalogue. Verify the maths on a real feed:

li = [{"feed_grade": 1.2, "conc_grade": 6.0, "tail_grade": 0.4}]   # %Li2O
print(SpodumeneRecovery().compute(li).value)   # 71.4 (%)
No engine changes
Notice you never touched the KPI engine or a switch statement. Commodity activation is pure metadata — the discovery API reads the commodities field.

Creating Mining Methods

A mining method (open-pit, underground, longwall…) changes the relevant equipment, events, and KPIs — because the bottleneck differs. Open-pit is haulage-bound; underground is access/hoisting-bound. Register a method and the KPIs it activates.

from mineproductivity.ontology.method import register_method

register_method(
    "block_cave",
    family="underground",
    bottleneck="drawpoint_availability",
    method_kpis=("UG.DRAWPOINT.UTIL", "UG.CAVE.PROPAGATION", "UG.ORE.HOISTING_RATE"),
)

KPIs tagged for a method are gated by it. A surface-only metric like truck cycle time is correctly withheld from an underground site, and vice-versa:

mine = Mine(site="Cadia East", commodity="gold", method="block_cave", maturity=3)
print("EQUIP.TRUCK.CYCLE" in [c.meta.code for c in mine.available_kpis()])
# False  (open-pit only)
MethodBottleneckCharacteristic KPIs
Open pitHaulage / congestioncycle time, TKPH, rolling resistance, match factor
UndergroundAccess / hoistingdevelopment advance (m/day), stope turnaround, hoisting rate

Writing Custom KPIs

This is the most common thing you will do. Adding a KPI is adding one object; the engine, API, CLI, docs, and dashboards pick it up automatically.

KPI class hierarchy (UML)
The KPI class hierarchy. Your KPI subclasses a category, sets meta, implements _compute, and is registered.

The recipe

Four steps, always the same: (1) pick a category base, (2) fill in complete metadata, (3) implement _compute(rows), (4) decorate with @register. Here is fuel intensity, a cost KPI that depends on production:

from mineproductivity.kpis import (register, KPIMetadata, CostKPI,
    Direction, Aggregation, DigitalMaturity)

@register
class FuelPerTonne(CostKPI):
    meta = KPIMetadata(
        code="COST.FUEL_PER_T", name="Fuel per Tonne", category="Cost",
        description="Litres of diesel burned per tonne moved.",
        formula="sum(fuel_l) / sum(payload_t)", unit="L/t",
        direction=Direction.LOWER_IS_BETTER, aggregation=Aggregation.RATIO,
        required_columns=("fuel_l", "payload_t"),
        dependencies=("PROD.TPH",), min_maturity=DigitalMaturity.L2_FMS,
        target="< 0.6 L/t", edge_cases=("payload_t == 0 -> None",))

    def _compute(self, rows):
        t = sum(r["payload_t"] for r in rows)
        return None if t == 0 else sum(r["fuel_l"] for r in rows) / t

On import it self-registers. It is now a first-class metric:

rows = [{"fuel_l": 5200, "payload_t": 9000}, {"fuel_l": 5400, "payload_t": 9300}]
print(REGISTRY["COST.FUEL_PER_T"]().compute(rows).value)   # 0.579 (L/t)
Metadata is mandatory
A KPI missing a required metadata field (formula, unit, direction, aggregation, required_columns…) fails CI. Complete metadata is what makes the metric discoverable, documentable, and AI-navigable.

Choosing the right base class

Category baseUse forExample codes
ProductionKPItonnes, plan attainmentPROD.TPH, PROD.PLAN_ATTAIN
FleetKPIavailability, utilisation, matchFLEET.PA, FLEET.UA, FLEET.MATCH
EquipmentKPIper-machine metricsEQUIP.TRUCK.CYCLE
QualityKPIgrade, recovery, dilutionQUAL.RECOVERY
DelayKPIlost time / valueDELAY.HOURS
CostKPIunit cost, fuel, consumablesCOST.FUEL_PER_T

Plugin Development

Connectors, commodity packs, solvers, visualisations, and agents all ship as plugins — separate installable packages that register themselves via Python entry-points. This is how the ecosystem grows without anyone forking the core.

Anatomy of a connector plugin

# mineproductivity_modular/connector.py
from mineproductivity.connectors import FMSConnector
from mineproductivity.core.events import CycleEvent, DelayEvent

class ModularConnector(FMSConnector):
    name = "modular"
    def get_cycle_data(self, since, until):
        for raw in self._client.cycles(since, until):   # vendor SDK, isolated here
            yield CycleEvent(event_id=raw["id"], equipment_id=raw["unit"],
                shift_id=raw["shift"], start_ts=raw["t0"], end_ts=raw["t1"],
                payload_t=raw["tonnes"], haul_min=raw["haul"] / 60)   # map codes
    def get_delay_data(self, since, until):
        ...   # map Modular reason codes -> canonical DelayEvent

Register it in the plugin’s pyproject.toml:

# pyproject.toml
[project.entry-points."mineproductivity.connectors"]
modular = "mineproductivity_modular.connector:ModularConnector"

After pip install mineproductivity-modular, the connector is selectable everywhere — mine.ingest(connector="modular") — with no change to the core.

Package dependency graph
The dependency rule plugins must respect: connectors may import model/ and core/, never kpis/; nothing imports a vendor SDK into the core.
Common pitfall
Do not put any KPI or analytics logic in a connector. Its single job is to normalise the source into canonical events. A connector that imports kpis/ will fail the dependency-direction CI check.

Validation

MineProductivity validates data without hiding problems. The principle is Traceability by Design: a result is either reproducible from named events or explicitly flagged — never silently wrong.

KPI-level validation

Every KPI’s .compute() runs .validate() first. Missing a required column yields value=None and a warning, not an exception or a wrong number:

bad = REGISTRY["FLEET.PA"]().compute([{"calendar_h": 24}])   # 'available_h' missing
print(bad.value, bad.warnings)
# None ("missing required columns: ['available_h']",)

Event-level validation

Events are validated against their JSON Schema at ingest (e.g. end_ts > start_ts, sub-durations sum within tolerance, payload within a plausibility band). A confidence score is attached so downstream KPIs know how complete their inputs were.

from mineproductivity.core.validation import validate_event
warnings = validate_event(cycle)
if warnings:
    log.warning("event %s: %s", cycle.event_id, warnings)
Design rule
Data problems become warnings on results, structural problems become typed exceptions. You can always trust that a non-None KPI value came from validated inputs.

Testing

MineProductivity is test-first. Every KPI ships a golden test — a fixed input with a known output — so a refactor can never silently change a published metric’s meaning.

import pytest
from mineproductivity.kpis import REGISTRY

def test_tph_golden():
    rows = [{"payload_t": 36000, "operating_h": 29.7}]
    assert round(REGISTRY["PROD.TPH"]().compute(rows).value, 1) == 1212.1

def test_pa_bounds():                       # property test
    rows = [{"available_h": 146, "calendar_h": 168}]
    pa = REGISTRY["FLEET.PA"]().compute(rows).value
    assert 0 <= pa <= 100

def test_tph_zero_hours_is_none():            # edge case from meta.edge_cases
    assert REGISTRY["PROD.TPH"]().compute([{"payload_t": 1, "operating_h": 0}]).value is None

Run the suite:

$ pytest -q
....................................  100%
312 passed in 1.84s
Contract tests for connectors
Every connector is automatically run through a shared contract suite that asserts it satisfies FMSConnector against recorded fixtures — no live OEM system required in CI.

Benchmarking

Benchmarking compares a KPI against a target, a peer group, or an industry quartile — fairly. Because ratio KPIs are re-derived from sums and normalised to a common basis, comparisons are like-for-like, not measurement artefacts.

from mineproductivity.analytics import benchmark

# Compare two sister mines in the same business unit against a 1,300 t/h plan
a = mine_a.kpi("PROD.TPH").by("shift")
b = mine_b.kpi("PROD.TPH").by("shift")

for name, r in [("Mine A", a), ("Mine B", b)]:
    bm = benchmark(r, scope="business_unit", target=1300)
    print(f"{name}: {r.value:.0f} t/h  ({bm.vs_target:+.1f}% vs target, Q{bm.quartile})")
# Mine A: 1212 t/h  (-6.8% vs target, Q3)
# Mine B: 1357 t/h  (+4.4% vs target, Q2)
Why this is trustworthy
Mine A’s −6.8% gap is a real signal because both numbers were computed identically and normalised to the same haul basis. A naive average of shift percentages would have hidden it.

Digital Twin

The Digital Twin is one living model of your mine kept in sync with reality. It exposes four queryable state layers from a single model, so you can ask “what is happening now?”, “what will happen?”, “what if?”, and “what happened last Tuesday?”.

Digital Twin state lifecycle
The twin synchronises from events and exposes live, predicted, simulated, and scenario/replay states.
from mineproductivity.twin import DigitalTwin

twin = DigitalTwin(mine)
twin.sync("PROD.TPH", shift="A-2026-06-25")        # live state
print(twin.live_state["PROD.TPH"].value)        # 1212.1

# What-if: fork the simulated state and add a truck to the loader on Bench 7
scenario = twin.fork()
scenario.set("bench7.trucks", 7)
print(scenario.predict("PROD.TPH").value)        # 1291.0  (closer to plan)

# Time-travel: reconstruct any past state from the immutable event lake
past = twin.replay(at="2026-06-18T06:00")
print(past.live_state["FLEET.PA"].value)         # 84.3
Event sourcing makes replay free
Because raw events are immutable truth and all KPIs are derived, the twin can rebuild any historical state exactly. Nothing special is stored for replay — it is recomputation.

Knowledge Graph

The Knowledge Graph is a traversable projection of the ontology: entities and KPIs are nodes; relationships and KPI dependencies are edges. It is what lets you ask cross-domain “why” questions without writing bespoke SQL joins.

Ontology concept map
The graph is a projection of the whole ontology around the KPI backbone — the substrate the AI agents reason over.
from mineproductivity.graph import KnowledgeGraph

kg = KnowledgeGraph.build(mine)

# Why might TPH be low? Walk its dependency chain.
for node in kg.dependencies("PROD.TPH"):
    print(node)
# operating_hours -> DelayEvent -> reason_code

# Find the path connecting fragmentation to crusher throughput
path = kg.path("DRILL_BLAST.P80", "CRUSH.TPH")
print(" -> ".join(path))
# DRILL_BLAST.P80 -> fragmentation -> diggability -> CRUSH.TPH

# Semantic search over KPI definitions
print([h.code for h in kg.search("tyre thermal limit")][:2])
# ['HAUL.TKPH', 'EQUIP.TYRE.LIFE']

Decision Intelligence

Decision Intelligence closes the loop: it turns KPI results into ranked, risk-weighted, explainable recommendations — and, where authorised, actions. A recommendation is an object that carries its evidence and expected impact.

from mineproductivity.decision import DecisionEngine

engine = DecisionEngine(mine, plan_tph=1300)
rec = engine.evaluate(mine.kpi("PROD.TPH").by("shift"))

print(rec.message)            # "Bench 7 loader under-trucked (match 0.92); assign one truck."
print(rec.expected_impact)    # "+88 t/h toward plan"
print(rec.evidence)          # "RCA: crusher_down 4.2h = 40% of lost time"
print(rec.risk, rec.authority) # "high (mill feed)" "advisory"

Recommendations are produced by AI agents that traverse the knowledge graph for evidence. The full reasoning chain — observation → hypothesis → evidence → root cause → recommendation → expected impact — is recorded and auditable.

Explainability is first-class
Every recommendation can show why. Operators act on what they understand; a black-box score with no provenance is rejected by design.

Python SDK

The SDK is the primary interface and is designed to feel like pandas or scikit-learn: typed, discoverable, object-first. Mine is the facade; everything hangs off it.

from mineproductivity import Mine

mine = Mine(site="Bingham West", commodity="copper", method="open_pit", maturity=2)
mine.ingest(connector="dispatch", since="2026-06-01")

# KPIs by object, sliced by any grouping the ontology supports
mine.kpi("PROD.TPH").by("shift")          # KPIResult series per shift
mine.kpi("FLEET.PA").by("equipment")      # per machine
mine.kpi("DELAY.HOURS").by("pit")         # per pit

# Discovery & introspection
mine.available_kpis()                       # list[type[KPI]] for this site profile
mine.describe("PROD.TPH").formula          # 'sum(payload_t) / sum(operating_h)'

# Export to a DataFrame for your own analysis
df = mine.kpi("PROD.TPH").by("shift").to_frame()
Familiar by design
If you know df.groupby(...).agg(...), .kpi(code).by(grouping) will feel natural — but it is correct-by-construction (right aggregation, maturity-gated, traceable).

REST API

For language-neutral integration, MineProductivity ships a FastAPI-based server. The schema is generated from the KPI registry, so the API never drifts from the model. Start it with the CLI:

$ mineprod serve --site "Bingham West" --port 8000
# List computable KPIs for a site
$ curl localhost:8000/v1/sites/bingham-west/kpis
[{"code":"PROD.TPH","name":"Tonnes per Hour","unit":"t/h"}, ...]

# Compute a KPI for a window
$ curl "localhost:8000/v1/sites/bingham-west/kpis/PROD.TPH?window=shift&date=2026-06-25"
{"code":"PROD.TPH","value":1212.1,"unit":"t/h","window":"shift","n":842,"warnings":[]}

A GraphQL endpoint at /graphql is ideal for traversing the ontology in one round-trip, and an Event API (webhooks / Kafka) publishes KPI-breach events. All three share one JSON Schema and one auth model (bearer tokens scoped to the enterprise tree).

OpenAPI included
Interactive docs are served at /docs (Swagger UI) and /redoc, exactly as a FastAPI user expects.

CLI

The mineprod command mirrors the SDK for automation and shift-handover use:

$ mineprod ingest  --site "Bingham West" --connector dispatch --since 2026-06-01
Ingested 18,420 cycle events, 312 delay events.

$ mineprod kpis    --site "Bingham West" --date 2026-06-25 --window shift
PROD.TPH    1212.1 t/h   (plan 1300, -6.8%)
FLEET.PA      86.9 %
FLEET.UA      89.0 %
FLEET.MATCH   0.923        (under-trucked)

$ mineprod report  --site "Bingham West" --date 2026-06-25 --format pdf -o shift.pdf
Wrote shift.pdf (4 pages).

Configuration

Site-specific choices are versioned configuration, not code. Configuration is layered: defaults → enterprise → region → site, each overriding the last. The KPI engine reads conventions like the shift calendar; it never assumes them.

# config/bingham_west.yaml
site: Bingham West
units: metric
timezone: America/Denver
shift_calendar:
  pattern: 2x12          # two 12-hour shifts
  start: "06:00"
conventions:
  operating_delay_in_operating: true   # queue/spot counted as operating
  standby_reduces_mechanical_availability: false
delay_codes: standard_v3
from mineproductivity.core.config import load_config
cfg = load_config(site="Bingham West")
print(cfg.shift_calendar.pattern)   # '2x12'
Comparability depends on config
Two sites are only comparable when their time-model conventions match. Declaring them explicitly (rather than hard-coding a midnight boundary) is what makes cross-site benchmarking valid.

Tutorials

End-to-end, story-driven walkthroughs. Each builds on the last; allow 15–30 minutes.

1 · From CSV to your first dashboard

Ingest a shift of Bingham West cycle data, compute the core fleet KPIs, and render a shift scorecard. Teaches: Mine, connectors, .kpi().by().

2 · Diagnosing a production shortfall

TPH is below plan. Use the knowledge graph and a delay Pareto to find the root cause and quantify the lost tonnes. Teaches: graph traversal, RCA, Decision Intelligence.

3 · Sizing a haul fleet

Drive match factor toward 1.0 by simulating 5–8 trucks against one shovel, then read P10/P50/P90 throughput. Teaches: dispatch, optimization, simulation.

4 · Onboarding a new commodity

Add lithium end-to-end: commodity, quality KPI, golden test, and a benchmark against a peer site. Teaches: domain extension, testing.

5 · Writing an OEM connector

Build, register, and contract-test a connector plugin for a new fleet-management system. Teaches: plugins, entry-points, contract tests.

6 · Standing up the REST API

Serve a site over REST + GraphQL, secure it with scoped tokens, and subscribe to KPI-breach webhooks. Teaches: API layer, security.

Tutorial 2 in miniature: diagnosing a shortfall

mine = Mine.load("Bingham West")
tph = mine.kpi("PROD.TPH").by("shift")
if tph.value < mine.plan("PROD.TPH"):
    cause = mine.rca("PROD.TPH")                 # traverses the knowledge graph
    print(cause.root, cause.share, cause.lost_tonnes)
    # 'crusher_down' 0.40 5091
    print(mine.recommend(cause).message)
    # 'Inspect crusher liner schedule; rebalance Bench 7 fleet.'

Examples

Short, copy-paste recipes for common tasks. Every example runs on the bundled bingham_west() dataset.

Compute the daily OEE-style fleet summary

summary = mine.summary(date="2026-06-25")
print(summary[["PROD.TPH", "FLEET.PA", "FLEET.UA", "FLEET.MATCH"]])
# PROD.TPH  FLEET.PA  FLEET.UA  FLEET.MATCH
# 1212.1     86.9      89.0      0.923

Rank delays by lost value, not lost hours

pareto = mine.kpi("DELAY.HOURS").by("reason").pareto(value=True)
for reason, lost_t in pareto.top(3):
    print(f"{reason:14s} {lost_t:>7,.0f} t")
# crusher_down     5,091 t
# queue            2,546 t
# no_operator      2,182 t

Export a copper recovery trend to Excel

mine.kpi("QUAL.RECOVERY").by("day").to_excel("recovery_june.xlsx")

Compare two shifts side by side

mine.compare(kpi="PROD.TPH", by="shift", of=["A-2026-06-25", "B-2026-06-25"])
# A-2026-06-25  1212.1 t/h
# B-2026-06-25  1357.0 t/h   (+12.0%)

Notebook Gallery

Runnable Jupyter notebooks, rendered in the docs and launchable on Binder/Colab. Each is self-contained and uses realistic data.

NotebookYou will learnLevel
01_quickstart.ipynbIngest → compute → visualise a shiftBeginner
02_custom_kpi.ipynbBuild, register, and golden-test a KPIBeginner
03_delay_pareto.ipynbLost-value Pareto & root-cause traversalIntermediate
04_fleet_sizing.ipynbMatch-factor optimisation with simulationIntermediate
05_digital_twin.ipynbLive state, what-if forks, replayIntermediate
06_knowledge_graph.ipynbCross-domain “why” queriesAdvanced
07_benchmark_fleet.ipynbMulti-site quartile benchmarkingAdvanced
08_oem_connector.ipynbWrite & contract-test a connector pluginAdvanced
$ pip install "mineproductivity[notebooks]"
$ mineprod notebooks --open       # copies the gallery into ./notebooks and launches Jupyter

Best Practices

Common Mistakes

Averaging ratios
Computing TPH per shift and then averaging the shift values. Use the declared RATIO aggregation, which re-derives from sum(payload)/sum(hours).
Putting logic in a connector
A connector that computes or imports kpis/ breaks vendor neutrality and fails the dependency-direction CI check. Connectors only normalise.
Editing the KPI engine to add a metric
Never necessary. Subclass a category, set metadata, implement _compute, @register. The engine executes your object unchanged.
Treating None as zero
A None KPI value means “could not be computed from valid inputs” (e.g. zero denominator, missing column) — check result.warnings rather than coercing to 0.
Comparing sites with mismatched conventions
Benchmarking two sites whose time-model conventions differ yields nonsense. Align their config first.

Performance Tips

The reference compute() is row-based and storage-agnostic, so it runs unchanged from a laptop CSV to a distributed lake. Scale with these levers:

Target (v1.0)Value
Event ingestion≥ 100M events/day/cluster
Incremental KPI latency< 2 s from event to updated shift KPI
API P95 (cached read)< 100 ms

FAQ

Do I need to know mining to use this?

No. The domain is modelled for you; the 60-second primer covers every term in the examples. You work with a normal Python API.

How is this different from just using pandas?

pandas gives you groupby; MineProductivity gives you correct, named, discoverable mining metrics with the right aggregation, maturity gating, and traceability built in — plus connectors, a Digital Twin, and AI. You can always drop to a DataFrame with .to_frame().

Is it tied to a particular fleet-management vendor?

No. The core is vendor-neutral; OEM systems are pluggable connectors. Swapping vendors changes one plugin.

Can AI agents safely extend it?

Yes — that is a core design goal. Agents read the same metadata and pass the same Definition of Done as humans. See the Machine Contributor model in the Architecture Handbook.

What Python versions are supported?

Python 3.11 and newer.

Is there a hosted version?

The open-source library is self-hostable today; a managed cloud is on the roadmap (see the Handbook’s Strategic Outlook).

Migration Guide

MineProductivity follows Semantic Versioning. A KPI code and its published formula semantics are public contracts — their meaning never changes silently.

Upgrading across minor versions (e.g. 1.2 → 1.3)

Backward compatible. New KPIs and connectors may appear; nothing existing changes meaning. Just upgrade:

$ pip install -U mineproductivity

Deprecations

A deprecated KPI or API is marked (not removed) for at least one minor version with a documented replacement, and emits a DeprecationWarning:

DeprecationWarning: 'FLEET.UTIL' is deprecated since 1.3; use 'FLEET.UA'. Removed in 2.0.

Major versions (e.g. 1.x → 2.0)

May change a formula’s meaning or remove deprecated symbols. Every breaking change ships with a migration note, a schema upgrader, and (where possible) a code alias. Check CHANGELOG.md and run mineprod migrate --check before upgrading.

Versioned artefacts
Package, ontology, schema, API, KPI, and benchmark versions are tracked independently so you always know exactly what changed.

Contributing Guide

Contributions — from humans and AI coding agents alike — pass through the same gate. The workflow:

  1. Read the Module Brief for the area you are touching (objective, I/O, do-not-modify boundaries, required tests).
  2. Open an issue / RFC for anything architectural; trivial changes go straight to a PR.
  3. Implement by subclassing and registering — never by editing an engine.
  4. Ship tests: a golden test for any KPI, plus the edge cases listed in its metadata.
  5. Pass the Definition of Done (below) — the CI gate.

Definition of Done (the merge gate)

[ ] Ontology/metadata entry exists first; code conforms to it
[ ] Full type hints; mypy --strict clean
[ ] Golden-value test + all edge_cases covered; suite green
[ ] KPI metadata complete; code unique
[ ] No vendor import in core/ontology/kpis; no upward-layer imports
[ ] Docstring + Module Brief updated
[ ] Changelog entry; SemVer impact assessed
$ git clone https://github.com/mineproductivity/mineproductivity
$ pip install -e ".[dev]"
$ pre-commit install        # lint + mypy + tests on every commit
$ pytest -q                  # run the suite
One pattern to rule them all
KPIs, connectors, commodities, methods, equipment types, solvers, agents — all extend the framework the same way: subclass, set metadata, register. Learn it once.