From cd2eb16b4f46db18b5e7434ed48ead80a8d477ca Mon Sep 17 00:00:00 2001 From: Toby Coleman Date: Tue, 10 Mar 2026 20:05:21 +0000 Subject: [PATCH 1/2] Add crusher circuit example --- .../002-crusher-circuit/.gitignore | 1 + .../002-crusher-circuit/_test_dynamics.py | 66 ++ .../002-crusher-circuit/crusher_circuit.ipynb | 1005 +++++++++++++++++ .../002-crusher-circuit/crusher_circuit.py | 457 ++++++++ 4 files changed, 1529 insertions(+) create mode 100644 examples/demos/physics-models/002-crusher-circuit/.gitignore create mode 100644 examples/demos/physics-models/002-crusher-circuit/_test_dynamics.py create mode 100644 examples/demos/physics-models/002-crusher-circuit/crusher_circuit.ipynb create mode 100644 examples/demos/physics-models/002-crusher-circuit/crusher_circuit.py diff --git a/examples/demos/physics-models/002-crusher-circuit/.gitignore b/examples/demos/physics-models/002-crusher-circuit/.gitignore new file mode 100644 index 00000000..afed0735 --- /dev/null +++ b/examples/demos/physics-models/002-crusher-circuit/.gitignore @@ -0,0 +1 @@ +*.csv diff --git a/examples/demos/physics-models/002-crusher-circuit/_test_dynamics.py b/examples/demos/physics-models/002-crusher-circuit/_test_dynamics.py new file mode 100644 index 00000000..0594a169 --- /dev/null +++ b/examples/demos/physics-models/002-crusher-circuit/_test_dynamics.py @@ -0,0 +1,66 @@ +"""Test the full circuit simulation with mass tracking + interpolated d80.""" + +import asyncio + +from crusher_circuit import Conveyor, Crusher, FeedSource, PSD, ProductCollector, Screen +from plugboard.connector import AsyncioConnector +from plugboard.process import LocalProcess +from plugboard.schemas import ConnectorSpec + +SIZE_CLASSES = [150.0, 106.0, 75.0, 53.0, 37.5, 26.5, 19.0, 13.2, 9.5, 6.7, 4.75] +FEED_FRACTIONS = [0.05, 0.10, 0.15, 0.20, 0.18, 0.12, 0.08, 0.05, 0.04, 0.02, 0.01] + +connect = lambda src, tgt: AsyncioConnector(spec=ConnectorSpec(source=src, target=tgt)) + +process = LocalProcess( + components=[ + FeedSource( + name="feed", + size_classes=SIZE_CLASSES, + feed_fractions=FEED_FRACTIONS, + feed_tonnes=100.0, + total_steps=50, + ), + Crusher( + name="crusher", + css=12.0, + oss=30.0, + k3=2.3, + n_stages=2, + initial_values={"recirc_psd": [None]}, + ), + Screen(name="screen", d50c=18.0, alpha=3.5, bypass=0.05), + Conveyor(name="conveyor", delay_steps=3), + ProductCollector(name="product", target_d80=20.0), + ], + connectors=[ + connect("feed.feed_psd", "crusher.feed_psd"), + connect("crusher.product_psd", "screen.crusher_product"), + connect("screen.undersize", "product.product_psd"), + connect("screen.oversize", "conveyor.input_psd"), + connect("conveyor.output_psd", "crusher.recirc_psd"), + ], +) + + +async def main() -> None: + async with process: + await process.run() + collector = process.components["product"] + d80s = [PSD(**h).d80 for h in collector.psd_history] + masses = [PSD(**h).mass_tonnes for h in collector.psd_history] + print(f"Steps: {len(d80s)}") + for i in [0, 1, 2, 3, 4, 5, 6, 7, 9, 14, 19, 29, 49]: + if i < len(d80s): + print(f" Step {i + 1:>2}: d80={d80s[i]:.2f} mm, mass={masses[i]:.1f} t") + unique_d80s = len(set(round(d, 4) for d in d80s)) + unique_masses = len(set(round(m, 4) for m in masses)) + print(f"\nUnique d80 values: {unique_d80s}") + print(f"Unique mass values: {unique_masses}") + print(f"d80 range: {min(d80s):.2f} - {max(d80s):.2f} mm") + print(f"Mass range: {min(masses):.1f} - {max(masses):.1f} t") + ok = unique_d80s > 1 and unique_masses > 1 + print("PASS" if ok else "FAIL - no dynamics") + + +asyncio.run(main()) diff --git a/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.ipynb b/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.ipynb new file mode 100644 index 00000000..10f80b4c --- /dev/null +++ b/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.ipynb @@ -0,0 +1,1005 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "423a3cf0", + "metadata": {}, + "source": [ + "# Crusher circuit simulation with Population Balance Modelling\n", + "\n", + "[![Open in Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/plugboard-dev/plugboard/blob/main/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.ipynb)\n", + "\n", + "[![Open in GitHub Codespaces](https://github.com/codespaces/badge.svg)](https://codespaces.new/plugboard-dev/plugboard)\n", + "\n", + "This example demonstrates how to build a dynamic **crusher circuit** simulation using Plugboard, based on the **Population Balance Model (PBM)** approach widely used in mineral processing.\n", + "\n", + "We model a closed-circuit crushing plant with five interconnected components:\n", + "\n", + "1. **Feed Source** — provides raw ore with an initial particle size distribution (PSD)\n", + "2. **Crusher** — breaks particles using the Whiten/King crusher model\n", + "3. **Screen** — classifies material into undersize (product) and oversize (recirculated)\n", + "4. **Conveyor** — introduces a transport delay on the recirculation loop (simulates a conveyor belt)\n", + "5. **Product Collector** — accumulates the final product and reports quality metrics\n", + "\n", + "The model tracks how the particle size distribution and **mass flow** evolve dynamically as the circulating load builds up over time.\n", + "\n", + "### Population Balance Model background\n", + "\n", + "The PBM is the standard mathematical framework for modelling comminution (size reduction) processes. For crushers, the Whiten model treats the crushing chamber as a series of classification and breakage stages:\n", + "\n", + "$$P = [B \\cdot S + (I - S)] \\cdot F$$\n", + "\n", + "where:\n", + "- $F$ is the feed size distribution vector\n", + "- $P$ is the product size distribution vector\n", + "- $S$ is the **selection (classification) matrix** — probability of a particle being selected for breakage based on crusher geometry\n", + "- $B$ is the **breakage matrix** — distribution of daughter fragments produced when a particle breaks\n", + "- $I$ is the identity matrix\n", + "\n", + "**Key references:**\n", + "- Whiten, W.J. (1974). *A matrix theory of comminution machines.* Chemical Engineering Science, 29(2), 589-599.\n", + "- King, R.P. (2001). *Modeling and Simulation of Mineral Processing Systems.* Butterworth-Heinemann.\n", + "- Austin, L.G., Klimpel, R.R., and Luckie, P.T. (1984). *Process Engineering of Size Reduction: Ball Milling.* SME." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ed5be457", + "metadata": {}, + "outputs": [], + "source": [ + "# Install plugboard and dependencies for Google Colab\n", + "!pip install -q plugboard plotly" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "104f591b", + "metadata": {}, + "outputs": [], + "source": [ + "import typing as _t\n", + "\n", + "from pydantic import BaseModel, field_validator\n", + "\n", + "from plugboard.component import Component, IOController as IO\n", + "from plugboard.connector import AsyncioConnector\n", + "from plugboard.process import LocalProcess\n", + "from plugboard.schemas import ComponentArgsDict, ConnectorSpec" + ] + }, + { + "cell_type": "markdown", + "id": "a4559c31", + "metadata": {}, + "source": [ + "## Particle Size Distribution data model\n", + "\n", + "We define a custom Pydantic model to represent particle size distributions (PSDs) throughout the circuit. This ensures data validation and provides a clean interface for working with size fractions.\n", + "\n", + "The `mass_tonnes` field tracks the **absolute mass flow** of each stream, which is critical for correctly modelling the mass-weighted mixing of fresh feed and recirculated material in the crusher." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2933c0b1", + "metadata": {}, + "outputs": [], + "source": [ + "class PSD(BaseModel):\n", + " \"\"\"Particle Size Distribution.\n", + "\n", + " Tracks the mass fraction retained in each discrete size class,\n", + " together with the absolute mass flow (tonnes) of the stream.\n", + "\n", + " Size classes are defined by their upper boundary (in mm),\n", + " ordered from coarsest to finest.\n", + " \"\"\"\n", + "\n", + " size_classes: list[float]\n", + " mass_fractions: list[float]\n", + " mass_tonnes: float = 1.0\n", + "\n", + " @field_validator(\"mass_fractions\")\n", + " @classmethod\n", + " def validate_fractions(cls, v: list[float]) -> list[float]:\n", + " total = sum(v)\n", + " if abs(total - 1.0) > 1e-6 and total > 0:\n", + " # Normalise to ensure mass balance\n", + " return [f / total for f in v]\n", + " return v\n", + "\n", + " def _percentile_size(self, target: float) -> float:\n", + " \"\"\"Interpolated percentile passing size (mm).\"\"\"\n", + " cumulative = 0.0\n", + " for i in range(len(self.size_classes) - 1, -1, -1):\n", + " prev_cum = cumulative\n", + " cumulative += self.mass_fractions[i]\n", + " if cumulative >= target:\n", + " if self.mass_fractions[i] > 0:\n", + " frac = (target - prev_cum) / self.mass_fractions[i]\n", + " else:\n", + " frac = 0.0\n", + " lower = (\n", + " self.size_classes[i + 1]\n", + " if i < len(self.size_classes) - 1\n", + " else 0.0\n", + " )\n", + " return lower + frac * (self.size_classes[i] - lower)\n", + " return self.size_classes[0]\n", + "\n", + " @property\n", + " def d80(self) -> float:\n", + " \"\"\"80th percentile passing size (mm), interpolated.\"\"\"\n", + " return self._percentile_size(0.80)\n", + "\n", + " @property\n", + " def d50(self) -> float:\n", + " \"\"\"50th percentile passing size (mm), interpolated.\"\"\"\n", + " return self._percentile_size(0.50)\n", + "\n", + " def to_cumulative_passing(self) -> list[float]:\n", + " \"\"\"Return cumulative % passing for each size class.\"\"\"\n", + " result: list[float] = []\n", + " cumulative = 0.0\n", + " for i in range(len(self.size_classes) - 1, -1, -1):\n", + " cumulative += self.mass_fractions[i]\n", + " result.insert(0, cumulative * 100)\n", + " return result" + ] + }, + { + "cell_type": "markdown", + "id": "ce410591", + "metadata": {}, + "source": [ + "## Crusher model functions\n", + "\n", + "These helper functions implement the Whiten crusher model. The **selection function** determines which particles are large enough to be broken based on the crusher's closed-side setting (CSS) and open-side setting (OSS). The **breakage function** describes how broken particles distribute across finer size classes." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "985e9906", + "metadata": {}, + "outputs": [], + "source": [ + "def build_selection_vector(\n", + " size_classes: list[float],\n", + " css: float,\n", + " oss: float,\n", + " k3: float = 2.3,\n", + ") -> list[float]:\n", + " \"\"\"Build the crusher selection (classification) vector.\n", + "\n", + " Implements the Whiten classification function:\n", + " - Particles smaller than CSS pass through unbroken (S=0)\n", + " - Particles larger than OSS are always broken (S=1)\n", + " - Particles between CSS and OSS have a fractional probability\n", + "\n", + " Args:\n", + " size_classes: Upper bounds of each size class (mm), coarsest first.\n", + " css: Closed-side setting (mm).\n", + " oss: Open-side setting (mm).\n", + " k3: Shape parameter controlling the classification curve steepness.\n", + "\n", + " Returns:\n", + " Selection probability for each size class.\n", + " \"\"\"\n", + " selection: list[float] = []\n", + " for d in size_classes:\n", + " if d <= css:\n", + " selection.append(0.0)\n", + " elif d >= oss:\n", + " selection.append(1.0)\n", + " else:\n", + " # Smooth transition between CSS and OSS\n", + " x = (d - css) / (oss - css)\n", + " selection.append(1.0 - (1.0 - x) ** k3)\n", + " return selection\n", + "\n", + "\n", + "def build_breakage_matrix(\n", + " size_classes: list[float],\n", + " phi: float = 0.4,\n", + " gamma: float = 1.5,\n", + " beta: float = 3.5,\n", + ") -> list[list[float]]:\n", + " \"\"\"Build the breakage distribution matrix (Austin & Luckie, 1972).\n", + "\n", + " B[i][j] is the fraction of material broken from size class j\n", + " that reports to size class i.\n", + "\n", + " The cumulative breakage function is:\n", + " B_cum(i,j) = phi * (x_i/x_j)^gamma + (1 - phi) * (x_i/x_j)^beta\n", + "\n", + " Args:\n", + " size_classes: Upper bounds of each size class (mm), coarsest first.\n", + " phi: Fraction of breakage due to impact (vs. abrasion).\n", + " gamma: Exponent for impact breakage.\n", + " beta: Exponent for abrasion breakage.\n", + "\n", + " Returns:\n", + " Lower-triangular breakage matrix B[i][j].\n", + " \"\"\"\n", + " n = len(size_classes)\n", + " B_cum = [[0.0] * n for _ in range(n)]\n", + "\n", + " for j in range(n):\n", + " for i in range(j, n):\n", + " if i == j:\n", + " B_cum[i][j] = 1.0\n", + " else:\n", + " ratio = size_classes[i] / size_classes[j]\n", + " B_cum[i][j] = phi * ratio**gamma + (1.0 - phi) * ratio**beta\n", + "\n", + " # Convert cumulative to fractional (incremental) breakage\n", + " B: list[list[float]] = [[0.0] * n for _ in range(n)]\n", + " for j in range(n):\n", + " for i in range(j + 1, n):\n", + " if i == n - 1:\n", + " B[i][j] = B_cum[i][j]\n", + " else:\n", + " B[i][j] = B_cum[i][j] - B_cum[i + 1][j]\n", + "\n", + " return B\n", + "\n", + "\n", + "def apply_crusher(feed: PSD, selection: list[float], B: list[list[float]]) -> PSD:\n", + " \"\"\"Apply one pass of the Whiten crusher model.\n", + "\n", + " P = [B·S + (I - S)] · F\n", + "\n", + " Args:\n", + " feed: Input particle size distribution.\n", + " selection: Selection vector (diagonal of S matrix).\n", + " B: Breakage matrix.\n", + "\n", + " Returns:\n", + " Product particle size distribution (mass_tonnes preserved).\n", + " \"\"\"\n", + " n = len(feed.size_classes)\n", + " f = feed.mass_fractions\n", + " product = [0.0] * n\n", + "\n", + " for i in range(n):\n", + " # Material that passes through unbroken\n", + " product[i] += (1.0 - selection[i]) * f[i]\n", + " # Material generated by breakage of coarser classes\n", + " for j in range(i):\n", + " product[i] += B[i][j] * selection[j] * f[j]\n", + "\n", + " # Normalise to maintain mass balance\n", + " total = sum(product)\n", + " if total > 0:\n", + " product = [p / total for p in product]\n", + "\n", + " return PSD(\n", + " size_classes=feed.size_classes,\n", + " mass_fractions=product,\n", + " mass_tonnes=feed.mass_tonnes,\n", + " )" + ] + }, + { + "cell_type": "markdown", + "id": "0c70fb1e", + "metadata": {}, + "source": [ + "## Screen model function\n", + "\n", + "The **screen partition curve** models the probability of a particle passing through the screen. Rather than a perfect cut, it uses a logistic model that accounts for the gradual transition around the cut-point size ($d_{50c}$) and a bypass fraction of fine material that misreports to the oversize stream.\n", + "\n", + "$$E_{\\text{undersize}}(d) = \\frac{1 - \\text{bypass}}{1 + (d / d_{50c})^{\\alpha}}$$\n", + "\n", + "This gives exactly 50% passing probability at the cut-point $d_{50c}$, with the sharpness parameter $\\alpha$ controlling how steep the transition is (King, 2001)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6682196d", + "metadata": {}, + "outputs": [], + "source": [ + "def screen_partition(\n", + " size_classes: list[float],\n", + " d50c: float,\n", + " alpha: float = 4.0,\n", + " bypass: float = 0.05,\n", + ") -> list[float]:\n", + " \"\"\"Calculate screen partition (efficiency) curve.\n", + "\n", + " Uses a logistic partition model for the screen. Returns the\n", + " fraction of material in each size class that reports to the\n", + " undersize (product) stream.\n", + "\n", + " Args:\n", + " size_classes: Upper bounds of each size class (mm), coarsest first.\n", + " d50c: Corrected cut-point size (mm) — 50% partition size.\n", + " alpha: Sharpness of separation (higher = sharper cut).\n", + " bypass: Fraction of fines that misreport to oversize.\n", + "\n", + " Returns:\n", + " Efficiency (fraction passing to undersize) for each size class.\n", + "\n", + " References:\n", + " King, R.P. (2001). Modeling and Simulation of Mineral Processing\n", + " Systems, Ch. 8. Butterworth-Heinemann.\n", + " \"\"\"\n", + " efficiency: list[float] = []\n", + " for d in size_classes:\n", + " e_undersize = (1.0 - bypass) / (1.0 + (d / d50c) ** alpha)\n", + " efficiency.append(e_undersize)\n", + " return efficiency" + ] + }, + { + "cell_type": "markdown", + "id": "4ad8c36d", + "metadata": {}, + "source": [ + "## Plugboard components\n", + "\n", + "Now we define the Plugboard components that make up our crusher circuit:\n", + "\n", + "1. **`FeedSource`** — generates a steady feed of ore with a given PSD and mass flow rate (tonnes per time step)\n", + "2. **`Crusher`** — applies the Whiten PBM model to produce a finer PSD, using mass-weighted mixing of feed and recirculated material\n", + "3. **`Screen`** — splits the crusher product into undersize (product) and oversize (recirculated), tracking mass flow through each stream\n", + "4. **`Conveyor`** — introduces a transport delay on the recirculation loop (simulates a conveyor belt carrying oversize material back to the crusher)\n", + "5. **`ProductCollector`** — accumulates the final product and reports quality metrics" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f4787aab", + "metadata": {}, + "outputs": [], + "source": [ + "class FeedSource(Component):\n", + " \"\"\"Generates a feed stream with a fixed particle size distribution.\n", + "\n", + " Runs for a set number of time steps, representing batches of ore\n", + " arriving at the crushing circuit.\n", + " \"\"\"\n", + "\n", + " io = IO(outputs=[\"feed_psd\"])\n", + "\n", + " def __init__(\n", + " self,\n", + " size_classes: list[float],\n", + " feed_fractions: list[float],\n", + " feed_tonnes: float = 100.0,\n", + " total_steps: int = 50,\n", + " **kwargs: _t.Unpack[ComponentArgsDict],\n", + " ) -> None:\n", + " super().__init__(**kwargs)\n", + " self._psd = PSD(\n", + " size_classes=size_classes,\n", + " mass_fractions=feed_fractions,\n", + " mass_tonnes=feed_tonnes,\n", + " )\n", + " self._total_steps = total_steps\n", + " self._step_count = 0\n", + "\n", + " async def step(self) -> None:\n", + " if self._step_count < self._total_steps:\n", + " self.feed_psd = self._psd.model_dump()\n", + " self._step_count += 1\n", + " else:\n", + " await self.io.close()\n", + "\n", + "\n", + "class Crusher(Component):\n", + " \"\"\"Whiten/King crusher model.\n", + "\n", + " Receives a feed PSD (fresh feed combined with recirculated oversize)\n", + " and applies the PBM to produce a crushed product PSD. Mass-weighted\n", + " mixing ensures the feed-to-recirculation ratio evolves dynamically.\n", + "\n", + " Parameters:\n", + " css: Closed-side setting (mm) — smallest gap in the crusher.\n", + " oss: Open-side setting (mm) — largest gap in the crusher.\n", + " k3: Classification curve shape parameter.\n", + " phi: Impact fraction in the breakage function.\n", + " gamma: Impact breakage exponent.\n", + " beta: Abrasion breakage exponent.\n", + " n_stages: Number of internal breakage stages (Whiten multi-stage model).\n", + " \"\"\"\n", + "\n", + " io = IO(inputs=[\"feed_psd\", \"recirc_psd\"], outputs=[\"product_psd\"])\n", + "\n", + " def __init__(\n", + " self,\n", + " css: float = 12.0,\n", + " oss: float = 25.0,\n", + " k3: float = 2.3,\n", + " phi: float = 0.4,\n", + " gamma: float = 1.5,\n", + " beta: float = 3.5,\n", + " n_stages: int = 3,\n", + " **kwargs: _t.Unpack[ComponentArgsDict],\n", + " ) -> None:\n", + " super().__init__(**kwargs)\n", + " self._css = css\n", + " self._oss = oss\n", + " self._k3 = k3\n", + " self._phi = phi\n", + " self._gamma = gamma\n", + " self._beta = beta\n", + " self._n_stages = n_stages\n", + " self._selection: list[float] | None = None\n", + " self._breakage: list[list[float]] | None = None\n", + "\n", + " async def step(self) -> None:\n", + " feed = PSD(**self.feed_psd)\n", + "\n", + " # Build matrices on first call (size classes are now known)\n", + " if self._selection is None:\n", + " self._selection = build_selection_vector(\n", + " feed.size_classes, self._css, self._oss, self._k3\n", + " )\n", + " self._breakage = build_breakage_matrix(\n", + " feed.size_classes, self._phi, self._gamma, self._beta\n", + " )\n", + "\n", + " # Mass-weighted mixing of fresh feed and recirculated oversize\n", + " recirc = self.recirc_psd\n", + " if recirc is not None:\n", + " recirc_psd = PSD(**recirc)\n", + " feed_mass = feed.mass_tonnes\n", + " recirc_mass = recirc_psd.mass_tonnes\n", + " total_mass = feed_mass + recirc_mass\n", + " w_f = feed_mass / total_mass\n", + " w_r = recirc_mass / total_mass\n", + " combined = [\n", + " w_f * f + w_r * r\n", + " for f, r in zip(feed.mass_fractions, recirc_psd.mass_fractions)\n", + " ]\n", + " feed = PSD(\n", + " size_classes=feed.size_classes,\n", + " mass_fractions=combined,\n", + " mass_tonnes=total_mass,\n", + " )\n", + "\n", + " # Apply multi-stage crushing\n", + " result = feed\n", + " if self._breakage is None:\n", + " raise RuntimeError(\"Breakage matrix not initialised\")\n", + " for _ in range(self._n_stages):\n", + " result = apply_crusher(result, self._selection, self._breakage)\n", + "\n", + " self.product_psd = result.model_dump()\n", + "\n", + "\n", + "class Screen(Component):\n", + " \"\"\"Vibrating screen separator.\n", + "\n", + " Splits the crusher product into undersize (product) and oversize\n", + " (recirculated back to the crusher). Tracks mass flow through each\n", + " output stream.\n", + "\n", + " Parameters:\n", + " d50c: Cut-point size (mm) — particles near this size have 50% chance of passing.\n", + " alpha: Sharpness of the separation curve.\n", + " bypass: Fraction of fines misreporting to oversize.\n", + " \"\"\"\n", + "\n", + " io = IO(inputs=[\"crusher_product\"], outputs=[\"undersize\", \"oversize\"])\n", + "\n", + " def __init__(\n", + " self,\n", + " d50c: float = 15.0,\n", + " alpha: float = 4.0,\n", + " bypass: float = 0.05,\n", + " **kwargs: _t.Unpack[ComponentArgsDict],\n", + " ) -> None:\n", + " super().__init__(**kwargs)\n", + " self._d50c = d50c\n", + " self._alpha = alpha\n", + " self._bypass = bypass\n", + "\n", + " async def step(self) -> None:\n", + " feed = PSD(**self.crusher_product)\n", + " efficiency = screen_partition(\n", + " feed.size_classes, self._d50c, self._alpha, self._bypass\n", + " )\n", + "\n", + " undersize_fracs: list[float] = []\n", + " oversize_fracs: list[float] = []\n", + " for i, f in enumerate(feed.mass_fractions):\n", + " undersize_fracs.append(f * efficiency[i])\n", + " oversize_fracs.append(f * (1.0 - efficiency[i]))\n", + "\n", + " # Mass split: proportion reporting to each stream\n", + " u_total = sum(undersize_fracs)\n", + " o_total = sum(oversize_fracs)\n", + "\n", + " undersize_mass = feed.mass_tonnes * u_total\n", + " oversize_mass = feed.mass_tonnes * o_total\n", + "\n", + " if u_total > 0:\n", + " undersize_fracs = [u / u_total for u in undersize_fracs]\n", + " if o_total > 0:\n", + " oversize_fracs = [o / o_total for o in oversize_fracs]\n", + "\n", + " self.undersize = PSD(\n", + " size_classes=feed.size_classes,\n", + " mass_fractions=undersize_fracs,\n", + " mass_tonnes=undersize_mass,\n", + " ).model_dump()\n", + " self.oversize = PSD(\n", + " size_classes=feed.size_classes,\n", + " mass_fractions=oversize_fracs,\n", + " mass_tonnes=oversize_mass,\n", + " ).model_dump()\n", + "\n", + "\n", + "class Conveyor(Component):\n", + " \"\"\"Transport delay for the recirculation loop.\n", + "\n", + " Simulates a conveyor belt by buffering PSDs in a FIFO queue.\n", + " The output is delayed by `delay_steps` time steps, creating\n", + " realistic transient dynamics in the circuit.\n", + "\n", + " Parameters:\n", + " delay_steps: Number of time steps to delay the material.\n", + " \"\"\"\n", + "\n", + " io = IO(inputs=[\"input_psd\"], outputs=[\"output_psd\"])\n", + "\n", + " def __init__(\n", + " self,\n", + " delay_steps: int = 3,\n", + " **kwargs: _t.Unpack[ComponentArgsDict],\n", + " ) -> None:\n", + " super().__init__(**kwargs)\n", + " self._delay_steps = delay_steps\n", + " self._buffer: list[dict[str, _t.Any] | None] = [None] * delay_steps\n", + "\n", + " async def step(self) -> None:\n", + " # Push new material onto the buffer, pop the oldest\n", + " self._buffer.append(self.input_psd)\n", + " delayed = self._buffer.pop(0)\n", + " self.output_psd = delayed\n", + "\n", + "\n", + "class ProductCollector(Component):\n", + " \"\"\"Collects the final product stream and tracks quality metrics.\n", + "\n", + " Records the PSD at each time step and computes running statistics\n", + " including the product d80 (80th percentile passing size).\n", + " \"\"\"\n", + "\n", + " io = IO(inputs=[\"product_psd\"], outputs=[\"d80\", \"d50\"])\n", + "\n", + " def __init__(\n", + " self,\n", + " target_d80: float = 20.0,\n", + " **kwargs: _t.Unpack[ComponentArgsDict],\n", + " ) -> None:\n", + " super().__init__(**kwargs)\n", + " self._target_d80 = target_d80\n", + " self.psd_history: list[dict[str, _t.Any]] = []\n", + "\n", + " async def step(self) -> None:\n", + " psd = PSD(**self.product_psd)\n", + " self.psd_history.append(psd.model_dump())\n", + " self.d80 = psd.d80\n", + " self.d50 = psd.d50" + ] + }, + { + "cell_type": "markdown", + "id": "743876a4", + "metadata": {}, + "source": [ + "## Define the size classes and initial feed\n", + "\n", + "We use a standard $\\sqrt{2}$ sieve series from 150 mm down to ~4.75 mm, giving 11 size classes. The fresh feed is a typical ROM (run-of-mine) ore distribution with most material in the coarser fractions." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "845fa3f5", + "metadata": {}, + "outputs": [], + "source": [ + "# Standard sqrt(2) sieve series from 150 mm down to ~4.75 mm\n", + "SIZE_CLASSES = [150.0, 106.0, 75.0, 53.0, 37.5, 26.5, 19.0, 13.2, 9.5, 6.7, 4.75]\n", + "\n", + "# ROM ore feed distribution (mass fractions, coarsest to finest)\n", + "FEED_FRACTIONS = [0.05, 0.10, 0.15, 0.20, 0.18, 0.12, 0.08, 0.05, 0.04, 0.02, 0.01]" + ] + }, + { + "cell_type": "markdown", + "id": "58ba9018", + "metadata": {}, + "source": [ + "## Assemble and run the process\n", + "\n", + "The circuit has a **recirculation loop**: screen oversize passes through a `Conveyor` (with a 3-step transport delay) before feeding back into the crusher. This delay simulates the time for material to travel on a conveyor belt, and creates realistic transient dynamics as the circulating load builds up.\n", + "\n", + "To break the circular dependency in the loop, we provide `initial_values` for the `recirc_psd` input on the Crusher component (set to `None` on the first step).\n", + "\n", + "Each stream carries its **absolute mass flow** (in tonnes), so the crusher correctly weights the mixing ratio of fresh feed vs. recirculated oversize." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a31fbc7f", + "metadata": {}, + "outputs": [], + "source": [ + "connect = lambda src, tgt: AsyncioConnector(spec=ConnectorSpec(source=src, target=tgt))\n", + "\n", + "process = LocalProcess(\n", + " components=[\n", + " FeedSource(\n", + " name=\"feed\",\n", + " size_classes=SIZE_CLASSES,\n", + " feed_fractions=FEED_FRACTIONS,\n", + " feed_tonnes=100.0,\n", + " total_steps=50,\n", + " ),\n", + " Crusher(\n", + " name=\"crusher\",\n", + " css=12.0,\n", + " oss=30.0,\n", + " k3=2.3,\n", + " n_stages=2,\n", + " initial_values={\"recirc_psd\": [None]},\n", + " ),\n", + " Screen(\n", + " name=\"screen\",\n", + " d50c=18.0,\n", + " alpha=3.5,\n", + " bypass=0.05,\n", + " ),\n", + " Conveyor(\n", + " name=\"conveyor\",\n", + " delay_steps=3,\n", + " ),\n", + " ProductCollector(\n", + " name=\"product\",\n", + " target_d80=20.0,\n", + " ),\n", + " ],\n", + " connectors=[\n", + " # Fresh feed → Crusher\n", + " connect(\"feed.feed_psd\", \"crusher.feed_psd\"),\n", + " # Crusher → Screen\n", + " connect(\"crusher.product_psd\", \"screen.crusher_product\"),\n", + " # Screen undersize → Product\n", + " connect(\"screen.undersize\", \"product.product_psd\"),\n", + " # Screen oversize → Conveyor → Crusher (recirculation with delay)\n", + " connect(\"screen.oversize\", \"conveyor.input_psd\"),\n", + " connect(\"conveyor.output_psd\", \"crusher.recirc_psd\"),\n", + " ],\n", + ")\n", + "\n", + "print(f\"Crusher circuit created: {len(process.components)} components, {len(process.connectors)} connectors\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a8539a20", + "metadata": {}, + "outputs": [], + "source": [ + "# Visualise the process as a diagram\n", + "from plugboard.diagram import MermaidDiagram\n", + "\n", + "diagram_url = MermaidDiagram.from_process(process).url\n", + "print(diagram_url)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f7b6d4da", + "metadata": {}, + "outputs": [], + "source": [ + "# Run the simulation\n", + "async with process:\n", + " await process.run()\n", + "\n", + "print(\"Simulation complete!\")\n", + "print(f\"Final product d80: {process.components['product'].d80:.2f} mm\")\n", + "print(f\"Final product d50: {process.components['product'].d50:.2f} mm\")" + ] + }, + { + "cell_type": "markdown", + "id": "4be6f6e0", + "metadata": {}, + "source": [ + "## Visualise results\n", + "\n", + "Let's plot the evolution of the product size metrics and mass flow over time. The **conveyor delay** creates distinct step-changes in the circulating load, and we can see the circuit gradually approach steady state." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d1612f5a", + "metadata": {}, + "outputs": [], + "source": [ + "import plotly.graph_objects as go\n", + "from plotly.subplots import make_subplots\n", + "\n", + "collector = process.components[\"product\"]\n", + "history = collector.psd_history\n", + "\n", + "# Extract metrics\n", + "d80_values = [PSD(**h).d80 for h in history]\n", + "d50_values = [PSD(**h).d50 for h in history]\n", + "mass_values = [PSD(**h).mass_tonnes for h in history]\n", + "steps = list(range(1, len(d80_values) + 1))\n", + "\n", + "fig = make_subplots(\n", + " rows=2, cols=2,\n", + " subplot_titles=(\n", + " \"Product size metrics over time\",\n", + " \"Product mass flow over time\",\n", + " \"Final cumulative PSD\",\n", + " \"Circulating load ratio\",\n", + " ),\n", + " vertical_spacing=0.15,\n", + " horizontal_spacing=0.12,\n", + ")\n", + "\n", + "# --- Plot 1: d80/d50 evolution ---\n", + "fig.add_trace(\n", + " go.Scatter(\n", + " x=steps, y=d80_values,\n", + " name=\"d80\",\n", + " line=dict(color=\"#2196F3\", width=2),\n", + " ),\n", + " row=1, col=1,\n", + ")\n", + "fig.add_trace(\n", + " go.Scatter(\n", + " x=steps, y=d50_values,\n", + " name=\"d50\",\n", + " line=dict(color=\"#FF9800\", width=2),\n", + " ),\n", + " row=1, col=1,\n", + ")\n", + "\n", + "# --- Plot 2: Mass flow ---\n", + "fig.add_trace(\n", + " go.Scatter(\n", + " x=steps, y=mass_values,\n", + " name=\"Product mass (t)\",\n", + " line=dict(color=\"#4CAF50\", width=2),\n", + " fill=\"tozeroy\",\n", + " fillcolor=\"rgba(76, 175, 80, 0.1)\",\n", + " ),\n", + " row=1, col=2,\n", + ")\n", + "fig.add_hline(\n", + " y=100.0, line_dash=\"dash\", line_color=\"gray\",\n", + " annotation_text=\"Feed rate (100 t)\",\n", + " row=1, col=2,\n", + ")\n", + "\n", + "# --- Plot 3: Final cumulative PSD ---\n", + "final_psd = PSD(**history[-1])\n", + "cum_passing = final_psd.to_cumulative_passing()\n", + "\n", + "fig.add_trace(\n", + " go.Scatter(\n", + " x=final_psd.size_classes, y=cum_passing,\n", + " name=\"Cumulative % passing\",\n", + " line=dict(color=\"#9C27B0\", width=2),\n", + " marker=dict(size=6),\n", + " ),\n", + " row=2, col=1,\n", + ")\n", + "\n", + "# --- Plot 4: Circulating load ratio ---\n", + "circ_load = [(100.0 - m) / 100.0 * 100 for m in mass_values]\n", + "fig.add_trace(\n", + " go.Scatter(\n", + " x=steps, y=circ_load,\n", + " name=\"Circulating load (%)\",\n", + " line=dict(color=\"#F44336\", width=2),\n", + " ),\n", + " row=2, col=2,\n", + ")\n", + "\n", + "fig.update_xaxes(title_text=\"Time step\", row=1, col=1)\n", + "fig.update_yaxes(title_text=\"Size (mm)\", row=1, col=1)\n", + "fig.update_xaxes(title_text=\"Time step\", row=1, col=2)\n", + "fig.update_yaxes(title_text=\"Mass (tonnes)\", row=1, col=2)\n", + "fig.update_xaxes(title_text=\"Particle size (mm)\", type=\"log\", row=2, col=1)\n", + "fig.update_yaxes(title_text=\"Cumulative % passing\", range=[0, 105], row=2, col=1)\n", + "fig.update_xaxes(title_text=\"Time step\", row=2, col=2)\n", + "fig.update_yaxes(title_text=\"Circulating load (%)\", row=2, col=2)\n", + "\n", + "fig.update_layout(\n", + " height=700,\n", + " width=1000,\n", + " title_text=\"Crusher Circuit — Dynamic Simulation Results\",\n", + " showlegend=True,\n", + ")\n", + "fig.show()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d47bae97", + "metadata": {}, + "outputs": [], + "source": [ + "# Display the final PSD as a table\n", + "print(f\"{'Size (mm)':>10} {'Mass %':>10} {'Cum. passing %':>15}\")\n", + "print(\"-\" * 37)\n", + "for size, frac, cum in zip(\n", + " final_psd.size_classes, final_psd.mass_fractions, cum_passing\n", + "):\n", + " print(f\"{size:10.1f} {frac * 100:10.2f} {cum:15.2f}\")" + ] + }, + { + "cell_type": "markdown", + "id": "ea439a25", + "metadata": {}, + "source": [ + "## Parameter optimisation with the Tuner\n", + "\n", + "Now let's use Plugboard's built-in `Tuner` to find the optimal crusher and screen settings. The Tuner uses [Ray Tune](https://docs.ray.io/en/latest/tune/index.html) with [Optuna](https://optuna.org/) to efficiently search the parameter space.\n", + "\n", + "**Objective:** Minimise the product d80 (we want a finer product).\n", + "\n", + "**Parameters to optimise:**\n", + "- **Crusher CSS** (6–20 mm): Controls the finest gap of the crusher\n", + "- **Crusher OSS** (20–50 mm): Controls the coarsest gap of the crusher\n", + "- **Screen cut-point** (8–25 mm): Controls where the screen separates undersize from oversize" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c40c9df0", + "metadata": {}, + "outputs": [], + "source": [ + "from plugboard.process import ProcessBuilder\n", + "from plugboard.schemas import (\n", + " FloatParameterSpec,\n", + " ObjectiveSpec,\n", + " ProcessArgsSpec,\n", + " ProcessSpec,\n", + ")\n", + "from plugboard.tune import Tuner\n", + "\n", + "# Define the process as a spec (required for the Tuner)\n", + "process_spec = ProcessSpec.model_validate(process.export())" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "cc504440", + "metadata": {}, + "outputs": [], + "source": [ + "tuner = Tuner(\n", + " objective=ObjectiveSpec(\n", + " object_name=\"product\",\n", + " field_type=\"field\",\n", + " field_name=\"d80\",\n", + " ),\n", + " parameters=[\n", + " FloatParameterSpec(\n", + " object_type=\"component\",\n", + " object_name=\"crusher\",\n", + " field_type=\"arg\",\n", + " field_name=\"css\",\n", + " lower=6.0,\n", + " upper=20.0,\n", + " ),\n", + " FloatParameterSpec(\n", + " object_type=\"component\",\n", + " object_name=\"crusher\",\n", + " field_type=\"arg\",\n", + " field_name=\"oss\",\n", + " lower=20.0,\n", + " upper=50.0,\n", + " ),\n", + " FloatParameterSpec(\n", + " object_type=\"component\",\n", + " object_name=\"screen\",\n", + " field_type=\"arg\",\n", + " field_name=\"d50c\",\n", + " lower=8.0,\n", + " upper=25.0,\n", + " ),\n", + " ],\n", + " num_samples=30,\n", + " max_concurrent=4,\n", + " mode=\"min\",\n", + ")\n", + "\n", + "result = tuner.run(spec=process_spec)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "545ab672", + "metadata": {}, + "outputs": [], + "source": [ + "# Display the optimal parameters found\n", + "best_css = result.config[\"component.crusher.arg.css\"]\n", + "best_oss = result.config[\"component.crusher.arg.oss\"]\n", + "best_d50c = result.config[\"component.screen.arg.d50c\"]\n", + "best_d80 = result.metrics[\"component.product.field.d80\"]\n", + "\n", + "print(\"=\" * 50)\n", + "print(\"OPTIMISATION RESULTS\")\n", + "print(\"=\" * 50)\n", + "print(f\" Crusher CSS: {best_css:.1f} mm\")\n", + "print(f\" Crusher OSS: {best_oss:.1f} mm\")\n", + "print(f\" Screen cut-point: {best_d50c:.1f} mm\")\n", + "print(f\" Product d80: {best_d80:.2f} mm\")\n", + "print(\"=\" * 50)" + ] + }, + { + "cell_type": "markdown", + "id": "1edd036e", + "metadata": {}, + "source": [ + "## Summary\n", + "\n", + "This example demonstrated how to:\n", + "\n", + "1. **Model a physical process** — using the Population Balance Model (Whiten/King approach) to simulate crushing and screening\n", + "2. **Use custom data structures** — a Pydantic `PSD` model to track particle size distributions and mass flow with built-in validation\n", + "3. **Build a recirculating circuit** — using `initial_values` to break the circular dependency in a closed-loop process\n", + "4. **Simulate transport delays** — using a `Conveyor` component with a FIFO buffer to model conveyor belt transport time\n", + "5. **Visualise dynamic behaviour** — tracking how the product quality and circulating load evolve over time\n", + "6. **Optimise process parameters** — using the `Tuner` to search for the crusher and screen settings that produce the finest product\n", + "\n", + "The Whiten crusher model and screen efficiency equations used here are standard tools in mineral processing engineering. For production use, these models would typically be calibrated against plant data from sampling surveys." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "plugboard (3.12.8)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.8" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.py b/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.py new file mode 100644 index 00000000..b5b31dde --- /dev/null +++ b/examples/demos/physics-models/002-crusher-circuit/crusher_circuit.py @@ -0,0 +1,457 @@ +"""Crusher circuit components for the Plugboard PBM comminution demo. + +Implements a closed-circuit crushing plant with: +- FeedSource: generates ore with an initial PSD +- Crusher: Whiten/King PBM model for size reduction +- Screen: Whiten efficiency curve for classification +- Conveyor: transport delay for the recirculation loop +- ProductCollector: tracks product quality metrics + +References: + Whiten, W.J. (1974). A matrix theory of comminution machines. + Chemical Engineering Science, 29(2), 589-599. + King, R.P. (2001). Modeling and Simulation of Mineral Processing Systems. + Butterworth-Heinemann. + Austin, L.G. & Luckie, P.T. (1972). Methods for determination of breakage + distribution parameters. Powder Technology, 5(4), 215-222. +""" + +import typing as _t + +from pydantic import BaseModel, field_validator + +from plugboard.component import Component, IOController as IO +from plugboard.schemas import ComponentArgsDict + + +# --------------------------------------------------------------------------- +# Data model +# --------------------------------------------------------------------------- + + +class PSD(BaseModel): + """Particle Size Distribution. + + Tracks the mass fraction retained in each discrete size class, + together with the absolute mass flow (tonnes) of the stream. + + Size classes are defined by their upper boundary (in mm), + ordered from coarsest to finest. + """ + + size_classes: list[float] + mass_fractions: list[float] + mass_tonnes: float = 1.0 + + @field_validator("mass_fractions") + @classmethod + def validate_fractions(cls, v: list[float]) -> list[float]: + """Normalise mass fractions so they sum to 1.""" + total = sum(v) + if abs(total - 1.0) > 1e-6 and total > 0: + return [f / total for f in v] + return v + + def _percentile_size(self, target: float) -> float: + """Interpolated percentile passing size (mm).""" + cumulative = 0.0 + for i in range(len(self.size_classes) - 1, -1, -1): + prev_cum = cumulative + cumulative += self.mass_fractions[i] + if cumulative >= target: + if self.mass_fractions[i] > 0: + frac = (target - prev_cum) / self.mass_fractions[i] + else: + frac = 0.0 + lower = self.size_classes[i + 1] if i < len(self.size_classes) - 1 else 0.0 + return lower + frac * (self.size_classes[i] - lower) + return self.size_classes[0] + + @property + def d80(self) -> float: + """80th percentile passing size (mm), interpolated.""" + return self._percentile_size(0.80) + + @property + def d50(self) -> float: + """50th percentile passing size (mm), interpolated.""" + return self._percentile_size(0.50) + + def to_cumulative_passing(self) -> list[float]: + """Return cumulative % passing for each size class.""" + result: list[float] = [] + cumulative = 0.0 + for i in range(len(self.size_classes) - 1, -1, -1): + cumulative += self.mass_fractions[i] + result.insert(0, cumulative * 100) + return result + + +# --------------------------------------------------------------------------- +# PBM helper functions +# --------------------------------------------------------------------------- + + +def build_selection_vector( + size_classes: list[float], + css: float, + oss: float, + k3: float = 2.3, +) -> list[float]: + """Build the crusher selection (classification) vector. + + Implements the Whiten classification function: + - Particles smaller than CSS pass through unbroken (S=0) + - Particles larger than OSS are always broken (S=1) + - Particles between CSS and OSS have a fractional probability + + Args: + size_classes: Upper bounds of each size class (mm), coarsest first. + css: Closed-side setting (mm). + oss: Open-side setting (mm). + k3: Shape parameter controlling the classification curve steepness. + + Returns: + Selection probability for each size class. + """ + selection: list[float] = [] + for d in size_classes: + if d <= css: + selection.append(0.0) + elif d >= oss: + selection.append(1.0) + else: + x = (d - css) / (oss - css) + selection.append(1.0 - (1.0 - x) ** k3) + return selection + + +def build_breakage_matrix( + size_classes: list[float], + phi: float = 0.4, + gamma: float = 1.5, + beta: float = 3.5, +) -> list[list[float]]: + """Build the breakage distribution matrix (Austin & Luckie, 1972). + + B[i][j] is the fraction of material broken from size class j + that reports to size class i. + + The cumulative breakage function is: + B_cum(i,j) = phi * (x_i/x_j)^gamma + (1 - phi) * (x_i/x_j)^beta + + Args: + size_classes: Upper bounds of each size class (mm), coarsest first. + phi: Fraction of breakage due to impact (vs. abrasion). + gamma: Exponent for impact breakage. + beta: Exponent for abrasion breakage. + + Returns: + Lower-triangular breakage matrix B[i][j]. + """ + n = len(size_classes) + b_cum = [[0.0] * n for _ in range(n)] + + for j in range(n): + for i in range(j, n): + if i == j: + b_cum[i][j] = 1.0 + else: + ratio = size_classes[i] / size_classes[j] + b_cum[i][j] = phi * ratio**gamma + (1.0 - phi) * ratio**beta + + # Convert cumulative to fractional (incremental) breakage + b_mat: list[list[float]] = [[0.0] * n for _ in range(n)] + for j in range(n): + for i in range(j + 1, n): + if i == n - 1: + b_mat[i][j] = b_cum[i][j] + else: + b_mat[i][j] = b_cum[i][j] - b_cum[i + 1][j] + + return b_mat + + +def apply_crusher(feed: PSD, selection: list[float], b_mat: list[list[float]]) -> PSD: + """Apply one pass of the Whiten crusher model. + + P = [B·S + (I - S)] · F + + Args: + feed: Input particle size distribution. + selection: Selection vector (diagonal of S matrix). + b_mat: Breakage matrix. + + Returns: + Product particle size distribution (mass_tonnes preserved). + """ + n = len(feed.size_classes) + f = feed.mass_fractions + product = [0.0] * n + + for i in range(n): + product[i] += (1.0 - selection[i]) * f[i] + for j in range(i): + product[i] += b_mat[i][j] * selection[j] * f[j] + + total = sum(product) + if total > 0: + product = [p / total for p in product] + + return PSD( + size_classes=feed.size_classes, + mass_fractions=product, + mass_tonnes=feed.mass_tonnes, + ) + + +def screen_partition( + size_classes: list[float], + d50c: float, + alpha: float = 4.0, + bypass: float = 0.05, +) -> list[float]: + """Calculate screen partition (efficiency) curve. + + Uses a logistic partition model for the screen. Returns the + fraction of material in each size class that reports to the + undersize (product) stream. + + The partition to undersize is: + E_undersize(d) = (1 - bypass) / (1 + (d / d50c)^alpha) + + This gives 50% efficiency at the cut-point size d50c, with + the sharpness parameter alpha controlling the steepness of + the transition. + + Args: + size_classes: Upper bounds of each size class (mm), coarsest first. + d50c: Corrected cut-point size (mm) — 50% partition size. + alpha: Sharpness of separation (higher = sharper cut). + bypass: Fraction of fines that misreport to oversize. + + Returns: + Efficiency (fraction passing to undersize) for each size class. + + References: + King, R.P. (2001). Modeling and Simulation of Mineral Processing + Systems, Ch. 8. Butterworth-Heinemann. + """ + efficiency: list[float] = [] + for d in size_classes: + e_undersize = (1.0 - bypass) / (1.0 + (d / d50c) ** alpha) + efficiency.append(e_undersize) + return efficiency + + +# --------------------------------------------------------------------------- +# Plugboard components +# --------------------------------------------------------------------------- + + +class FeedSource(Component): + """Generates a feed stream with a fixed particle size distribution.""" + + io = IO(outputs=["feed_psd"]) + + def __init__( + self, + size_classes: list[float], + feed_fractions: list[float], + feed_tonnes: float = 100.0, + total_steps: int = 50, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self._psd = PSD( + size_classes=size_classes, + mass_fractions=feed_fractions, + mass_tonnes=feed_tonnes, + ) + self._total_steps = total_steps + self._step_count = 0 + + async def step(self) -> None: + """Emit the feed PSD until the configured number of steps is reached.""" + if self._step_count < self._total_steps: + self.feed_psd = self._psd.model_dump() + self._step_count += 1 + else: + await self.io.close() + + +class Crusher(Component): + """Whiten/King crusher model. + + Receives a feed PSD (fresh feed combined with recirculated oversize) + and applies the PBM to produce a crushed product PSD. Mass-weighted + mixing ensures the feed-to-recirculation ratio evolves dynamically. + """ + + io = IO(inputs=["feed_psd", "recirc_psd"], outputs=["product_psd"]) + + def __init__( + self, + css: float = 12.0, + oss: float = 25.0, + k3: float = 2.3, + phi: float = 0.4, + gamma: float = 1.5, + beta: float = 3.5, + n_stages: int = 3, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self._css = css + self._oss = oss + self._k3 = k3 + self._phi = phi + self._gamma = gamma + self._beta = beta + self._n_stages = n_stages + self._selection: list[float] | None = None + self._breakage: list[list[float]] | None = None + + async def step(self) -> None: + """Combine fresh feed with recirculated oversize and crush.""" + feed = PSD(**self.feed_psd) + + if self._selection is None: + self._selection = build_selection_vector( + feed.size_classes, self._css, self._oss, self._k3 + ) + self._breakage = build_breakage_matrix( + feed.size_classes, self._phi, self._gamma, self._beta + ) + + # Mass-weighted mixing of fresh feed and recirculated oversize + recirc = self.recirc_psd + if recirc is not None: + recirc_psd = PSD(**recirc) + feed_mass = feed.mass_tonnes + recirc_mass = recirc_psd.mass_tonnes + total_mass = feed_mass + recirc_mass + w_f = feed_mass / total_mass + w_r = recirc_mass / total_mass + combined = [ + w_f * f + w_r * r for f, r in zip(feed.mass_fractions, recirc_psd.mass_fractions) + ] + feed = PSD( + size_classes=feed.size_classes, + mass_fractions=combined, + mass_tonnes=total_mass, + ) + + result = feed + if self._breakage is None: + msg = "Breakage matrix not initialised" + raise RuntimeError(msg) + for _ in range(self._n_stages): + result = apply_crusher(result, self._selection, self._breakage) + + self.product_psd = result.model_dump() + + +class Screen(Component): + """Vibrating screen separator using the Whiten efficiency model. + + Splits mass between undersize and oversize streams based on the + partition curve. Each output stream carries its share of the + total mass flow. + """ + + io = IO(inputs=["crusher_product"], outputs=["undersize", "oversize"]) + + def __init__( + self, + d50c: float = 15.0, + alpha: float = 4.0, + bypass: float = 0.05, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self._d50c = d50c + self._alpha = alpha + self._bypass = bypass + + async def step(self) -> None: + """Split crusher product into undersize and oversize streams.""" + feed = PSD(**self.crusher_product) + efficiency = screen_partition(feed.size_classes, self._d50c, self._alpha, self._bypass) + + undersize_fracs: list[float] = [] + oversize_fracs: list[float] = [] + for i, f in enumerate(feed.mass_fractions): + undersize_fracs.append(f * efficiency[i]) + oversize_fracs.append(f * (1.0 - efficiency[i])) + + u_total = sum(undersize_fracs) + o_total = sum(oversize_fracs) + + # Mass split: proportion reporting to each stream + undersize_mass = feed.mass_tonnes * u_total + oversize_mass = feed.mass_tonnes * o_total + + if u_total > 0: + undersize_fracs = [u / u_total for u in undersize_fracs] + if o_total > 0: + oversize_fracs = [o / o_total for o in oversize_fracs] + + self.undersize = PSD( + size_classes=feed.size_classes, + mass_fractions=undersize_fracs, + mass_tonnes=undersize_mass, + ).model_dump() + self.oversize = PSD( + size_classes=feed.size_classes, + mass_fractions=oversize_fracs, + mass_tonnes=oversize_mass, + ).model_dump() + + +class Conveyor(Component): + """Transport delay for the recirculation loop. + + Simulates a conveyor belt by buffering PSDs in a FIFO queue. + The output is delayed by ``delay_steps`` time steps, creating + realistic transient dynamics in the circuit. + """ + + io = IO(inputs=["input_psd"], outputs=["output_psd"]) + + def __init__( + self, + delay_steps: int = 3, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self._delay_steps = delay_steps + self._buffer: list[dict[str, _t.Any] | None] = [None] * delay_steps + + async def step(self) -> None: + """Push new material onto the buffer, pop the oldest.""" + self._buffer.append(self.input_psd) + delayed = self._buffer.pop(0) + self.output_psd = delayed + + +class ProductCollector(Component): + """Collects the final product stream and tracks quality metrics.""" + + io = IO(inputs=["product_psd"], outputs=["d80", "d50"]) + + def __init__( + self, + target_d80: float = 20.0, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self._target_d80 = target_d80 + self.psd_history: list[dict[str, _t.Any]] = [] + + async def step(self) -> None: + """Record the product PSD and emit quality metrics.""" + psd = PSD(**self.product_psd) + self.psd_history.append(psd.model_dump()) + self.d80 = psd.d80 + self.d50 = psd.d50 From 6e91047638dcc115e95a3065f4eeac6edf6e2765 Mon Sep 17 00:00:00 2001 From: Toby Coleman Date: Tue, 10 Mar 2026 20:07:45 +0000 Subject: [PATCH 2/2] Updated LLM context from feat/plugboard-go-app branch --- .github/agents/docs.agent.md | 1 + .github/agents/examples.agent.md | 42 ++++++++++++++++++++++++++++ .github/agents/lint.agent.md | 1 + .github/agents/researcher.agent.md | 20 ++++++++++++++ examples/AGENTS.md | 44 ++---------------------------- 5 files changed, 66 insertions(+), 42 deletions(-) create mode 100644 .github/agents/examples.agent.md create mode 100644 .github/agents/researcher.agent.md diff --git a/.github/agents/docs.agent.md b/.github/agents/docs.agent.md index 2fa4d8be..50c313c4 100644 --- a/.github/agents/docs.agent.md +++ b/.github/agents/docs.agent.md @@ -1,6 +1,7 @@ --- description: 'Maintain Plugboard documentation' tools: ['execute', 'read', 'edit', 'search', 'web', 'github.vscode-pull-request-github/activePullRequest', 'ms-python.python/getPythonEnvironmentInfo', 'ms-python.python/getPythonExecutableCommand', 'ms-python.python/installPythonPackage', 'ms-python.python/configurePythonEnvironment', 'todo'] +model: ['GPT-5 mini (copilot)', 'GPT-4.1 (copilot)'] --- You are an expert technical writer responsible for maintaining the documentation of the Plugboard project. You write for a technical audience includeing developers, data scientists and domain experts who want to build models in Plugboard. diff --git a/.github/agents/examples.agent.md b/.github/agents/examples.agent.md new file mode 100644 index 00000000..5aba458e --- /dev/null +++ b/.github/agents/examples.agent.md @@ -0,0 +1,42 @@ +--- +name: examples +description: Develops example Plugboard models to demonstrate the capabilities of the framework +argument-hint: A description of the example to generate, along with any specific requirements, ideas about structure, or constraints. +agents: ['researcher', 'docs', 'lint'] +--- + +You are responsible for building high quality tutorials and demo examples for the Plugboard framework. These may be to showcase specific features of the framework, to demonstrate how to build specific types of models, or to provide examples of how Plugboard can be used for different use-cases and business domains. + +## Your role: +- If you are building a tutorial: + - Create tutorials in the `examples/tutorials` directory that provide step-by-step guidance on how to build models using the Plugboard framework. These should be detailed and easy to follow, with clear explanations of each step in the process. + - Create markdown documentation alongside code. You can delegate to the `docs` subagent to make these updates. + - Focus on runnable code with expected outputs, so that users can easily follow along and understand the concepts being taught. +- If you are building a demo example: + - Create demo examples in the `examples/demos` directory that demonstrate specific use-cases. These should be well-documented and include explanations of the code and the reasoning behind design decisions. + - Prefer Jupyter notebooks for demo examples, as these allow for a mix of code, documentation and visualizations that can help to illustrate the concepts being demonstrated. +- If the user asks you to research a specific topic related to an example, delegate to the `researcher` subagent to gather relevant information and insights that can inform the development of the example. + + +## Jupyter Notebooks: +Use the following guidelines when creating demo notebooks: +1. **Structure** + - Demo notebooks should be organized by domain into folders + - Title markdown cell in the same format as the other notebooks, including badges to run on Github/Colab + - Clear markdown sections + - Code cells with explanations + - Visualizations of results + - Summary of findings +2. **Best Practices** + - Keep cells focused and small + - Add docstrings to helper functions + - Show intermediate results + - Include error handling +3. **Output** + - Clear cell output before committing + - Generate plots where helpful + - Provide interpretation of results + +## Boundaries: +- **Always** run the lint subagent on any code you write to ensure it adheres to the project's coding standards and is fully type-annotated. +- **Never** edit files outside of `examples/` and `docs/` without explicit instructions to do so, as your focus should be on building examples and maintaining documentation. \ No newline at end of file diff --git a/.github/agents/lint.agent.md b/.github/agents/lint.agent.md index 1998b07f..e30ee243 100644 --- a/.github/agents/lint.agent.md +++ b/.github/agents/lint.agent.md @@ -1,6 +1,7 @@ --- description: 'Maintain code quality by running linting tools and resolving issues' tools: ['execute', 'read', 'edit', 'search', 'ms-python.python/getPythonEnvironmentInfo', 'ms-python.python/getPythonExecutableCommand', 'ms-python.python/installPythonPackage', 'ms-python.python/configurePythonEnvironment'] +model: ['GPT-5 mini (copilot)', 'GPT-4.1 (copilot)'] --- You are responsible for maintaining code quality in the Plugboard project by running linting tools and resolving any issues that arise. diff --git a/.github/agents/researcher.agent.md b/.github/agents/researcher.agent.md new file mode 100644 index 00000000..27c1f45a --- /dev/null +++ b/.github/agents/researcher.agent.md @@ -0,0 +1,20 @@ +--- +name: researcher +description: Researches specific topics on the internet and gathers relevant information. +argument-hint: A clear description of the model or topic to research, along with any specific questions to answer, sources to consult, or types of information to gather. +tools: ['vscode', 'read', 'agent', 'search', 'web', 'todo'] +--- + +You are a subject-matter expert researcher responsible for gathering information on specific topics related to the Plugboard project. Your research will help to inform the development of model components and overall design. + +## Your role: +Focus on gathering information about: +- Approaches to modeling the specific process or system being researched, including any relevant theories, frameworks, or best practices +- How the model or simulation can be structured into components, and what the inputs and outputs of those components should be +- What the data flow between components should look like, and any data structures required +- Any specific algorithms or equations that need to be implemented inside the components + +## Boundaries: +- **Always** provide clear and concise summaries of the information you gather. +- Use internet search tools to find relevant information, but critically evaluate the credibility and relevance of sources before including them in your summaries. +- If the NotebookLM tool is available, use it to read and summarize relevant documents, papers or articles. Ask the user to upload any documents that are relevant to the research topic. \ No newline at end of file diff --git a/examples/AGENTS.md b/examples/AGENTS.md index 8917cfdd..a0464a60 100644 --- a/examples/AGENTS.md +++ b/examples/AGENTS.md @@ -1,24 +1,6 @@ -# AI Agent Instructions for Plugboard Examples +# AI Agent Instructions for Plugboard Models -This document provides guidelines for AI agents working with Plugboard example code, demonstrations, and tutorials. - -## Purpose - -These examples demonstrate how to use Plugboard to model and simulate complex processes. Help users build intuitive, well-documented examples that showcase Plugboard's capabilities. - -## Example Categories - -### Tutorials (`tutorials/`) - -Step-by-step learning materials for new users. Focus on: -- Clear explanations of concepts. -- Progressive complexity. -- Runnable code with expected outputs. -- Markdown documentation alongside code. You can delegate to the `docs` agent to make these updates. - -### Demos (`demos/`) - -Practical applications are organized by domain into folders. +This document provides guidelines for AI agents working with specific models implemented in Plugboard. ## Creating a Plugboard model @@ -244,28 +226,6 @@ Later, load and run via CLI plugboard process run my-model.yaml ``` -## Jupyter Notebooks - -Use the following guidelines when creating demo notebooks: - -1. **Structure** - - Title markdown cell in the same format as the other notebooks, including badges to run on Github/Colab - - Clear markdown sections - - Code cells with explanations - - Visualizations of results - - Summary of findings - -2. **Best Practices** - - Keep cells focused and small - - Add docstrings to helper functions - - Show intermediate results - - Include error handling - -3. **Output** - - Clear cell output before committing - - Generate plots where helpful - - Provide interpretation of results - ## Resources - **Library Components**: `plugboard.library`