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.
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.
A 60-second mining primer
| Term | What it means (for an engineer) |
|---|---|
| Cycle | One round-trip of a haul truck: queue → spot → load → haul → dump → return. Emits a CycleEvent. |
| Payload | Tonnes carried in one cycle (a CAT 793 hauls ~220 t). |
| TPH | Tonnes 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. |
| Delay | Lost productive time (crusher down, queueing, refuelling). Emits a DelayEvent. |
| Match factor | Are there enough trucks for the shovel? ~1.0 is balanced. |
| Recovery | % of valuable metal the plant extracts from the ore. |
| Digital maturity | How 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.
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
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").
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
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]: ...
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')
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 (%)
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)
| Method | Bottleneck | Characteristic KPIs |
|---|---|---|
| Open pit | Haulage / congestion | cycle time, TKPH, rolling resistance, match factor |
| Underground | Access / hoisting | development 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.
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)
Choosing the right base class
| Category base | Use for | Example codes |
|---|---|---|
ProductionKPI | tonnes, plan attainment | PROD.TPH, PROD.PLAN_ATTAIN |
FleetKPI | availability, utilisation, match | FLEET.PA, FLEET.UA, FLEET.MATCH |
EquipmentKPI | per-machine metrics | EQUIP.TRUCK.CYCLE |
QualityKPI | grade, recovery, dilution | QUAL.RECOVERY |
DelayKPI | lost time / value | DELAY.HOURS |
CostKPI | unit cost, fuel, consumables | COST.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.
model/ and core/, never kpis/; nothing imports a vendor SDK into the core.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)
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
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)
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?”.
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
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.
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.
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()
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).
/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'
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.
| Notebook | You will learn | Level |
|---|---|---|
01_quickstart.ipynb | Ingest → compute → visualise a shift | Beginner |
02_custom_kpi.ipynb | Build, register, and golden-test a KPI | Beginner |
03_delay_pareto.ipynb | Lost-value Pareto & root-cause traversal | Intermediate |
04_fleet_sizing.ipynb | Match-factor optimisation with simulation | Intermediate |
05_digital_twin.ipynb | Live state, what-if forks, replay | Intermediate |
06_knowledge_graph.ipynb | Cross-domain “why” queries | Advanced |
07_benchmark_fleet.ipynb | Multi-site quartile benchmarking | Advanced |
08_oem_connector.ipynb | Write & contract-test a connector plugin | Advanced |
$ pip install "mineproductivity[notebooks]"
$ mineprod notebooks --open # copies the gallery into ./notebooks and launches Jupyter
Best Practices
- Extend, never edit. Add a KPI, connector, or commodity by subclassing and registering. If you find yourself editing an engine, stop — there is an extension point.
- Always populate complete metadata. It is what makes your KPI discoverable, documentable, and AI-usable. CI enforces it anyway.
- Let the framework aggregate. Declare
aggregation=RATIOand trust it; never hand-average percentages across periods or machines. - Keep connectors thin. Normalise only. No KPI logic, no analytics, no vendor SDK leaking upward.
- Ship a golden test with every KPI. A fixed input/output pins the metric’s meaning forever.
- Declare site conventions in config. Time-model choices belong in versioned configuration, not in code.
- Use codes, not strings, across boundaries. A KPI
code(PROD.TPH) is a stable public contract; a display name is not.
Common Mistakes
RATIO aggregation, which re-derives from sum(payload)/sum(hours).kpis/ breaks vendor neutrality and fails the dependency-direction CI check. Connectors only normalise._compute, @register. The engine executes your object unchanged.None as zeroNone KPI value means “could not be computed from valid inputs” (e.g. zero denominator, missing column) — check result.warnings rather than coercing to 0.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:
- Vectorise hot paths with the Polars backend:
mine.config.compute_backend = "polars". Identical results, far faster on millions of rows. - Compute incrementally. Stream mode updates only the affected window on each new event rather than recomputing the shift.
- Parallelise by site/asset. Per-unit work is embarrassingly parallel; enable Dask for fleet-wide runs (
mine.config.executor = "dask"). - Push down to DuckDB / Arrow. Aggregations over the event lake run in-engine; only results cross into Python.
- Cache aggregates. The OLAP layer caches shift/day rollups; the event lake remains the rebuildable source of truth.
| 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.
Contributing Guide
Contributions — from humans and AI coding agents alike — pass through the same gate. The workflow:
- Read the Module Brief for the area you are touching (objective, I/O, do-not-modify boundaries, required tests).
- Open an issue / RFC for anything architectural; trivial changes go straight to a PR.
- Implement by subclassing and registering — never by editing an engine.
- Ship tests: a golden test for any KPI, plus the edge cases listed in its metadata.
- 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