Rangrez dashboard

.gitattributes

@@ -33,3 +33,16 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/crumpled_strip.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/daylight_window.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/evening_warm.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/kitchen_tube.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/kurta_target.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/lamp_desk.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/light_box.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/no_paper.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/phone_flash.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/shadowy.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/tilted_phone.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/vertical_strip.jpg filter=lfs diff=lfs merge=lfs -text
+src/rangrez/scenes/with_target.jpg filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,12 @@
+# Rangrez dashboard as a Hugging Face Space: the full dye-lab loop on the simulator,
+# every dip a real-looking phone photo. Meta models run only with the live passcode.
+FROM python:3.12-slim
+RUN pip install --no-cache-dir uv==0.9.5
+WORKDIR /app
+COPY pyproject.toml uv.lock README.md ./
+COPY src ./src
+RUN uv sync --frozen --no-dev && chmod -R a+rX /app
+ENV RANGREZ_HOST=0.0.0.0 RANGREZ_PORT_HTTP=7860 RANGREZ_RECORD=0 RANGREZ_STEP_DELAY=1.2 \
+    HOME=/tmp PYTHONUNBUFFERED=1
+EXPOSE 7860
+CMD ["/app/.venv/bin/rangrez-lab"]

README.md

@@ -1,10 +1,26 @@
 ---
 title: Rangrez
-emoji: 📉
-colorFrom: yellow
-colorTo: pink
+emoji: 🧵
+colorFrom: indigo
+colorTo: yellow
 sdk: docker
+app_port: 7860
 pinned: false
+license: apache-2.0
+short_description: A self-driving dye lab on Meta models
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Rangrez: a self-driving dye lab
+
+Pick a colour and press **Start campaign**. Rangrez mixes a recipe, dyes a cotton strip in its
+simulated lab (a hidden dye lot that differs from the supplier's model), photographs it like a
+phone would, measures it, and corrects itself until the cloth matches. Every dip photo is a
+real-looking scene: desk lamps, daylight, tube lights, shadows, paper of unknown whiteness.
+
+Without the live passcode it runs the offline policy and SAM masks cached for each scene.
+With it, hosted **SAM 3.1** measures every photo and **Muse Wonder** runs the campaign,
+and you can read a target colour from your own photo ("match this kurta").
+
+One lab, shared by everyone viewing: if a campaign is running, you are watching it.
+
+Code, hardware, firmware and evaluation: https://github.com/TusharTechs/rangrez

pyproject.toml

@@ -0,0 +1,38 @@
+[project]
+name = "rangrez"
+version = "0.1.0"
+description = "A self-driving dye lab: Meta models match any colour on real cotton"
+readme = "README.md"
+authors = [
+    { name = "Tushar Agarwal", email = "tusharaggarwal274@gmail.com" }
+]
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi>=0.162.0",
+    "meta-sam-parser>=0.1.0",
+    "numpy>=2.5.4",
+    "openai>=3.42.0",
+    "opencv-python-headless>=5.0.0.93",
+    "pillow>=13.0.0",
+    "pillow-heif>=1.8.1",
+    "pyserial>=3.5",
+    "python-dotenv>=1.2.4",
+    "python-multipart>=0.0.33",
+    "scipy>=1.18.1",
+    "uvicorn[standard]>=0.54.0",
+    "websockets>=17.4",
+]
+
+[project.scripts]
+rangrez-lab = "rangrez.server:main"
+rangrez-card = "rangrez.make_card:main"
+
+[build-system]
+requires = ["uv_build>=0.13.1,<0.14.0"]
+build-backend = "uv_build"
+
+[dependency-groups]
+dev = [
+    "pytest>=9.2.0",
+    "ruff>=0.17.3",
+]

src/rangrez/__init__.py

@@ -0,0 +1,2 @@
+def hello() -> str:
+    return "Hello from rangrez!"

src/rangrez/agent.py

@@ -0,0 +1,180 @@
+"""Muse Wonder runs the campaign; deterministic rules keep it honest.
+
+The model chooses which tool to call and explains why in the lab notebook.
+Code enforces the rest: the shot budget, the safety gate, and escalation to a
+human after three dips in a row that fail to improve on the best match.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from collections.abc import Callable
+
+import numpy as np
+
+from .campaign import Campaign
+from .profiles import PROFILE
+from .rig import DYES
+from .vision import BASE_URL, ApiCall, CallLog
+
+WONDER_MODEL = os.environ.get("RANGREZ_AGENT_MODEL", "muse-wonder-1.0")
+NO_PROGRESS_LIMIT = 3
+MIN_IMPROVEMENT_DE00 = 0.3  # smaller gains are within dip-to-dip noise
+CORRECTION_ZONE_DE00 = 6.0
+OUT_OF_GAMUT_STREAK = 2  # dips at the concentration cap that still miss
+
+SYSTEM_PROMPT = """You are the colourist running Rangrez, a small self-driving dye lab for cotton.
+Goal: match the target shade on a real cotton strip to CIEDE2000 <= the tolerance, in as few dips as possible.
+
+You control the lab only through tools. Rules:
+- Never invent a colour, concentration or measurement. Every number you state must come from a tool result.
+- Before every run_shot, call suggest_recipes. You may pick a candidate or blend candidates, but explain why.
+- After each shot with a usable measurement, call fit_model so the dye model learns this dye lot.
+- For the first dip, if a glimmer candidate is offered, use it: Glimmer learned this dye lot from the lab's own dips.
+- Far from the target, trust model_inverse. Once the best dip is within about dE00 6, prefer colourist_correction:
+  the model always predicts a perfect match but cannot capture everything about this dye lot (saturation, lot strength),
+  while the correction is learned from real dips.
+- If a shot is flagged REJECT, do not trust its colour; re-run or change something and say what.
+- Write one short notebook entry (write_note) per shot: hypothesis, what you changed, what you expect.
+- If dips at the concentration cap still miss, the shade is outside this dye set's gamut: say so and report the closest dip.
+- Stop with finish when the target is matched or the budget is spent. Use escalate if something looks physically wrong.
+Dyes: {dyes}. {description}."""
+
+TOOLS = [
+    {"type": "function", "function": {"name": "get_state", "description": "Target, shots so far, best match, model gains, stock levels.", "parameters": {"type": "object", "properties": {}}}},
+    {"type": "function", "function": {"name": "fit_model", "description": "Refit the Kubelka-Munk dye model to all valid shots.", "parameters": {"type": "object", "properties": {}}}},
+    {"type": "function", "function": {"name": "suggest_recipes", "description": "Candidate recipes from model inversion and colourist correction, each with predicted dE00 and gate status.", "parameters": {"type": "object", "properties": {}}}},
+    {"type": "function", "function": {"name": "run_shot", "description": "Dispense, dip a cotton strip, dry, photograph and measure. Gated by hard safety limits.", "parameters": {"type": "object", "properties": {
+        "recipe": {"type": "object", "properties": {d: {"type": "number"} for d in DYES}, "required": list(DYES)},
+        "dip_seconds": {"type": "number", "description": "30-300, default 90"},
+        "rationale": {"type": "string"}}, "required": ["recipe", "rationale"]}}},
+    {"type": "function", "function": {"name": "write_note", "description": "Append to the lab notebook.", "parameters": {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]}}},
+    {"type": "function", "function": {"name": "escalate", "description": "Stop and hand over to the human colourist.", "parameters": {"type": "object", "properties": {"reason": {"type": "string"}}, "required": ["reason"]}}},
+    {"type": "function", "function": {"name": "finish", "description": "End the campaign with a short summary.", "parameters": {"type": "object", "properties": {"summary": {"type": "string"}}, "required": ["summary"]}}},
+]
+
+
+def _no_progress(c: Campaign) -> int:
+    """Dips since the best match last improved by more than measurement noise."""
+    best_de, since = float("inf"), 0
+    for s in c.shots:
+        if s.measurement.ok and s.delta_e < best_de - MIN_IMPROVEMENT_DE00:
+            best_de, since = s.delta_e, 0
+        else:
+            since += 1
+    return since
+
+
+def _out_of_gamut(c: Campaign) -> bool:
+    """The last dips pushed a dye to its safety cap and still missed: unreachable."""
+    recent = [s for s in c.shots if s.measurement.ok][-OUT_OF_GAMUT_STREAK:]
+    if len(recent) < OUT_OF_GAMUT_STREAK:
+        return False
+    cap = 0.98 * c.rig.limits.max_conc_g_per_l
+    return all(max(s.recipe.conc) >= cap and s.delta_e > c.tolerance for s in recent)
+
+
+def dispatch(c: Campaign, name: str, args: dict) -> dict:
+    if name == "get_state":
+        return c.state()
+    if name == "fit_model":
+        return c.fit_model()
+    if name == "suggest_recipes":
+        return c.suggest_recipes()
+    if name == "run_shot":
+        out = c.run_shot(args["recipe"], float(args.get("dip_seconds", 90.0)), args.get("rationale", ""))
+        if c.status == "running" and _out_of_gamut(c):
+            c.declare_out_of_gamut()
+            out["out_of_gamut"] = True
+        elif c.status == "running" and _no_progress(c) >= NO_PROGRESS_LIMIT:
+            c.escalate(f"{NO_PROGRESS_LIMIT} dips in a row without improving on the best match")
+            out["escalated"] = True
+        return out
+    if name == "write_note":
+        c.note(args["text"])
+        return {"ok": True}
+    if name == "escalate":
+        return c.escalate(args["reason"])
+    if name == "finish":
+        if c.status == "running":
+            c.status = "finished"
+        c.note(args.get("summary", ""), who="agent")
+        c.emit("status", status=c.status)
+        return {"ok": True}
+    return {"error": f"unknown tool {name}"}
+
+
+def run_with_wonder(c: Campaign, log: CallLog | None = None, max_turns: int = 60) -> None:
+    """Drive the campaign with Muse Wonder over the Meta Model API."""
+    from openai import OpenAI
+
+    log = log or CallLog()
+    client = OpenAI(base_url=BASE_URL, api_key=os.environ["MODEL_API_KEY"], timeout=120, max_retries=2)
+    messages: list[dict] = [
+        {"role": "system", "content": SYSTEM_PROMPT.format(dyes=", ".join(DYES), description=PROFILE.description)},
+        {"role": "user", "content": "New campaign. Current state:\n" + json.dumps(c.state())},
+    ]
+    for _ in range(max_turns):
+        if c.status != "running":
+            return
+        started = time.monotonic()
+        try:
+            resp = client.chat.completions.create(model=WONDER_MODEL, messages=messages, tools=TOOLS)
+        except Exception as exc:
+            log.add(ApiCall(WONDER_MODEL, "plan next step", time.monotonic() - started, False, str(exc)))
+            raise
+        usage = getattr(resp, "usage", None)
+        log.add(ApiCall(WONDER_MODEL, "plan next step", time.monotonic() - started, True,
+                        f"{getattr(usage, 'total_tokens', '?')} tokens"))
+        msg = resp.choices[0].message
+        messages.append(msg.model_dump(exclude_none=True))
+        if msg.content:
+            c.emit("thought", text=msg.content)
+        if not msg.tool_calls:
+            messages.append({"role": "user", "content": "Continue with a tool call, or call finish."})
+            continue
+        for call in msg.tool_calls:
+            try:
+                args = json.loads(call.function.arguments or "{}")
+            except json.JSONDecodeError:
+                args = {}
+            c.emit("tool", name=call.function.name, args=args)
+            result = dispatch(c, call.function.name, args)
+            messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result, default=float)})
+    if c.status == "running":
+        c.escalate("agent turn budget exhausted")
+
+
+def run_deterministic(c: Campaign, on_step: Callable[[], None] | None = None) -> None:
+    """Offline policy with the same tools and rules: used in tests and replay mode."""
+    while c.status == "running" and len(c.shots) < c.max_shots:
+        if c.shots:
+            dispatch(c, "fit_model", {})
+        cands = [x for x in c.suggest_recipes()["candidates"] if x["gate"] == "pass"]
+        if not cands:
+            dispatch(c, "escalate", {"reason": "no candidate passes the safety gate"})
+            break
+        # A colourist starts from the model, then corrects from real dips once close.
+        near = c.best is not None and c.best.delta_e < CORRECTION_ZONE_DE00
+        correction = [x for x in cands if x["method"] == "colourist_correction"]
+        first = [x for x in cands if x["method"] == "glimmer"]
+        if first:
+            pick = first[0]
+        else:
+            pick = correction[0] if near and correction else min(cands, key=lambda x: x["predicted_delta_e00"])
+        dispatch(c, "write_note", {"text": f"Shot {len(c.shots) + 1}: {pick['method']} predicts dE00 {pick['predicted_delta_e00']}."})
+        dispatch(c, "run_shot", {"recipe": pick["recipe"], "rationale": pick["why"]})
+        if on_step:
+            on_step()
+    if c.status == "running":
+        dispatch(c, "finish", {"summary": "Shot budget spent."})
+
+
+def best_delta_e(c: Campaign) -> float:
+    b = c.best
+    return float("inf") if b is None else b.delta_e
+
+
+__all__ = ["run_with_wonder", "run_deterministic", "dispatch", "best_delta_e", "np"]

src/rangrez/assisted.py

@@ -0,0 +1,127 @@
+"""Assisted rig: a person is the pumps, the phone is the camera.
+
+The agent and the safety gate work exactly as with the robot rig. Instead of
+driving pumps, a dip becomes a short set of instructions on the dashboard
+(how many mL of each colour, how long to dip), and the lab waits for a photo
+of the dried strip lying next to the colour card.
+"""
+
+from __future__ import annotations
+
+import io
+import threading
+from collections.abc import Callable
+from dataclasses import dataclass, field
+
+import numpy as np
+from PIL import Image, ImageOps
+
+from .camera import grab
+from .camera import source as camera_source
+from .profiles import PROFILE
+from .rig import DYES, UNIT, Inventory, PumpPlan, Recipe, RigLimits, ShotResult, now
+
+STEP_ML = PROFILE.step_ml
+MAX_SIDE_PX = 1920  # SAM 3.1 accepts up to about 4 MP; phones shoot 12+
+
+
+class Cancelled(RuntimeError):
+    pass
+
+
+def normalise_photo(data: bytes) -> bytes:
+    """Any phone photo (JPEG, PNG, iPhone HEIC) -> upright RGB JPEG no larger than MAX_SIDE_PX."""
+    from pillow_heif import register_heif_opener
+
+    register_heif_opener()
+    try:
+        img = Image.open(io.BytesIO(data))
+        img = ImageOps.exif_transpose(img).convert("RGB")
+    except Exception as exc:
+        raise ValueError("could not read that image; please send a JPEG, PNG or HEIC photo") from exc
+    img.thumbnail((MAX_SIDE_PX, MAX_SIDE_PX))
+    buf = io.BytesIO()
+    img.save(buf, format="JPEG", quality=93)
+    return buf.getvalue()
+
+
+@dataclass
+class AssistedRig:
+    limits: RigLimits = field(default_factory=RigLimits)
+    inventory: Inventory = field(default_factory=Inventory)
+    on_instructions: Callable[[dict], None] = lambda _: None
+    timeout_s: float = 45 * 60
+    camera: int | str = field(default_factory=camera_source)
+
+    def __post_init__(self) -> None:
+        self._photo: bytes | None = None
+        self._ready = threading.Event()
+        self._cancelled = False
+        self.waiting: dict | None = None
+
+    # -- what a person can actually measure ------------------------------
+    def quantize(self, recipe: Recipe) -> Recipe:
+        ml = recipe.as_array() / self.limits.stock_g_per_l * self.limits.bath_ml
+        ml = np.round(ml / STEP_ML) * STEP_ML
+        conc = ml / self.limits.bath_ml * self.limits.stock_g_per_l
+        return Recipe(tuple(float(round(v, 4)) for v in conc), recipe.dip_seconds)
+
+    def instructions(self, recipe: Recipe, index: int) -> dict:
+        ml = recipe.as_array() / self.limits.stock_g_per_l * self.limits.bath_ml
+        water = self.limits.bath_ml - ml.sum()
+        return {
+            "index": index,
+            "dyes_ml": {d: round(float(v), 2) for d, v in zip(DYES, ml)},
+            "amounts": {d: (round(c) if UNIT == "drops" else round(float(v), 1)) for d, v, c in zip(DYES, ml, recipe.conc)},
+            "unit": "drops" if UNIT == "drops" else "mL",
+            "water_ml": round(float(water), 1),
+            "dip_seconds": int(recipe.dip_seconds),
+            "steps": [
+                "Rinse the glass and the dropper or syringes.",
+                *[self._amount(d, v, c) for d, v, c in zip(DYES, ml, recipe.conc) if v >= STEP_ML / 2],
+                (f"Add 4 teaspoons (20 mL) of water and stir." if UNIT == "drops" else f"Add {water:.1f} mL of water and stir for 10 seconds."),
+                f"Dip a fresh white cotton strip for {int(recipe.dip_seconds)} seconds.",
+                "Squeeze it gently and dry it fully with the hair dryer (about 2 minutes).",
+                "Lay it on a sheet of white paper (or next to the colour card) in the same spot and light as before, and send the photo.",
+                "If you have the customer's cloth, lay it beside the strip: Rangrez then compares them side by side under the same light.",
+            ],
+        }
+
+    @staticmethod
+    def _amount(dye: str, ml: float, conc: float) -> str:
+        if UNIT == "drops":
+            return f"Add {round(conc)} drops of {dye}."
+        return f"Add {ml:.1f} mL of {dye}."
+
+    # -- rig protocol ----------------------------------------------------------
+    def run_shot(self, recipe: Recipe) -> ShotResult:
+        started = now()
+        index = getattr(self, "_count", 0) + 1
+        self._count = index
+        self.waiting = self.instructions(recipe, index)
+        self._photo, self._cancelled = None, False
+        self._ready.clear()
+        self.on_instructions(self.waiting)
+        if not self._ready.wait(self.timeout_s):
+            self.waiting = None
+            raise Cancelled("no photo arrived in time")
+        self.waiting = None
+        if self._cancelled or self._photo is None:
+            raise Cancelled("campaign stopped")
+        ml = recipe.as_array() / self.limits.stock_g_per_l * self.limits.bath_ml
+        plan = PumpPlan(tuple(float(v) for v in ml), float(self.limits.bath_ml - ml.sum()))
+        return ShotResult(self._photo, plan, started, now(), {"assisted": True})
+
+    def provide(self, photo: bytes) -> None:
+        """Called when the person uploads a photo or presses 'use webcam'."""
+        if self.waiting is None:
+            raise RuntimeError("the lab is not waiting for a photo right now")
+        self._photo = normalise_photo(photo)
+        self._ready.set()
+
+    def cancel(self) -> None:
+        self._cancelled = True
+        self._ready.set()
+
+    def capture(self) -> bytes:
+        return grab(self.camera)

src/rangrez/camera.py

@@ -0,0 +1,41 @@
+"""Grab one frame from a camera: a webcam index, a phone streaming as an IP
+webcam (e.g. http://192.168.1.7:8080/video), or a video file.
+
+    RANGREZ_CAMERA=0                                # first webcam
+    RANGREZ_CAMERA=1                                # Continuity Camera on a Mac, often
+    RANGREZ_CAMERA=http://192.168.1.7:8080/video    # an Android phone running an IP-webcam app
+"""
+
+from __future__ import annotations
+
+import os
+
+
+def source() -> int | str:
+    raw = os.environ.get("RANGREZ_CAMERA", "0").strip()
+    return int(raw) if raw.isdigit() else raw
+
+
+def grab(src: int | str | None = None, warmup: int = 10, width: int = 1920, height: int = 1080) -> bytes:
+    """A JPEG of the current view. Warm-up frames let auto exposure settle."""
+    import cv2
+
+    src = source() if src is None else src
+    cam = cv2.VideoCapture(src)
+    try:
+        if not cam.isOpened():
+            raise RuntimeError(f"could not open camera {src!r}")
+        if isinstance(src, int):
+            cam.set(cv2.CAP_PROP_FRAME_WIDTH, width)
+            cam.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
+        frame = None
+        for _ in range(warmup + 1):
+            ok, f = cam.read()
+            if ok:
+                frame = f
+        if frame is None:
+            raise RuntimeError(f"camera {src!r} returned no frame")
+        ok, jpeg = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 94])
+        return jpeg.tobytes()
+    finally:
+        cam.release()

src/rangrez/campaign.py

@@ -0,0 +1,244 @@
+"""The colourist's toolbox: deterministic tools the agent calls.
+
+A shade-matching campaign is what a dye-house colourist does by hand: an
+initial recipe from the dye model, a lab dip, a measurement, a correction,
+again. Here every step is a tool with a checkable result. The agent decides
+which tool to call and why; it never computes a colour or a volume itself.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+
+import numpy as np
+
+from . import glimmer
+from .colour import delta_e_2000, lab_to_hex
+from .km import DyeModel, fit_gains, solve_recipe
+from .profiles import PROFILE
+from .rig import DYES, UNIT, GateError, Recipe, Rig, check_recipe, execute
+from .vision import Measurement, Segmenter, measure, overlay
+
+SIDE_BY_SIDE_MAX_DE00 = 12.0  # a cloth further than this from the target is not the customer's
+
+
+@dataclass
+class Shot:
+    index: int
+    recipe: Recipe
+    measurement: Measurement
+    delta_e: float
+    rationale: str
+    image_jpeg: bytes
+    overlay_jpeg: bytes
+    at: float = field(default_factory=time.time)
+
+
+@dataclass
+class Campaign:
+    target_lab: np.ndarray
+    rig: Rig
+    segmenter: Segmenter
+    tolerance: float = 1.5  # CIEDE2000 pass threshold
+    max_shots: int = 10
+    model: DyeModel = field(default_factory=PROFILE.nominal)
+    shots: list[Shot] = field(default_factory=list)
+    notebook: list[dict] = field(default_factory=list)
+    listeners: list = field(default_factory=list)
+    status: str = "running"
+    side_readings: list = field(default_factory=list)  # the customer's cloth, read beside each strip
+
+    # -- events ------------------------------------------------------------
+    def emit(self, kind: str, **data) -> None:
+        event = {"kind": kind, "at": time.time(), **data}
+        for fn in self.listeners:
+            fn(event)
+
+    def note(self, text: str, who: str = "agent") -> None:
+        entry = {"who": who, "text": text, "at": time.time()}
+        self.notebook.append(entry)
+        self.emit("note", **entry)
+
+    # -- state ---------------------------------------------------------------
+    @property
+    def best(self) -> Shot | None:
+        ok = [s for s in self.shots if s.measurement.ok]
+        return min(ok, key=lambda s: s.delta_e) if ok else None
+
+    def state(self) -> dict:
+        best = self.best
+        return {
+            "target_lab": [round(v, 2) for v in self.target_lab],
+            "target_hex": lab_to_hex(self.target_lab),
+            "tolerance_de00": self.tolerance,
+            "shots_used": len(self.shots),
+            "shots_left": self.max_shots - len(self.shots),
+            "best": None if best is None else self._shot_summary(best),
+            "history": [self._shot_summary(s) for s in self.shots],
+            "model_gains": dict(zip(DYES, np.round(self.model.gains, 3).tolist())),
+            "stock_ml": dict(zip(DYES, np.round(self.rig.inventory.stock_ml, 1).tolist())),
+            "status": self.status,
+        }
+
+    @staticmethod
+    def _shot_summary(s: Shot) -> dict:
+        m = s.measurement
+        return {
+            "shot": s.index,
+            "recipe": dict(zip(DYES, np.round(s.recipe.conc, 3).tolist())),
+            "measured_lab": [round(v, 2) for v in m.lab],
+            "lab_sd": [round(v, 2) for v in m.lab_sd],
+            "delta_e00": round(s.delta_e, 2),
+            "flags": m.flags,
+        }
+
+    # -- tools -----------------------------------------------------------------
+    def fit_model(self) -> dict:
+        """Refit per-dye uptake to every valid shot so far."""
+        data = [(s.recipe.as_array(), s.measurement.lab) for s in self.shots if s.measurement.ok]
+        before = self.model.gains.copy()
+        self.model = fit_gains(self.model, data)
+        errs = [float(delta_e_2000(self.model.predict_lab(c), lab)) for c, lab in data]
+        out = {
+            "gains_before": dict(zip(DYES, np.round(before, 3).tolist())),
+            "gains_after": dict(zip(DYES, np.round(self.model.gains, 3).tolist())),
+            "model_fit_error_de00": round(float(np.mean(errs)), 2) if errs else None,
+            "shots_used": len(data),
+        }
+        self.emit("model", **out)
+        return out
+
+    def suggest_recipes(self) -> dict:
+        """Candidate recipes with predicted CIEDE2000, from two independent methods."""
+        cands = []
+        conc, de = solve_recipe(self.model, self.target_lab, max_conc=self.rig.limits.max_conc_g_per_l)
+        cands.append(("model_inverse", conc, de, "Invert the refitted Kubelka-Munk model."))
+        if not self.shots and glimmer.configured():
+            g = glimmer.first_guess(self.target_lab, log=getattr(self.segmenter, "log", None))
+            if g is not None:
+                d = float(delta_e_2000(self.model.predict_lab(g), self.target_lab))
+                cands.append(("glimmer", g, d, "Muse Glimmer fine-tuned on this lab's own dips: its first recipes "
+                              "land closer than the supplier model's (the predicted dE00 shown is the supplier model's opinion)."))
+        best = self.best
+        if best is not None:
+            step = self._correction_step(best)
+            if step is not None:
+                pred = float(delta_e_2000(best.measurement.lab + self._jac(best.recipe.as_array()) @ (step - best.recipe.as_array()), self.target_lab))
+                cands.append(("colourist_correction", step, pred, "Correct the best real dip by its measured error, using the refitted model's sensitivity."))
+        out = []
+        for name, c, d, why in cands:
+            c = np.clip(c, 0, self.rig.limits.max_conc_g_per_l)
+            recipe = Recipe(tuple(float(round(v, 3)) for v in c))
+            try:
+                check_recipe(recipe, self.rig.limits, self.rig.inventory)
+                gate = "pass"
+            except GateError as exc:
+                gate = f"refused: {exc}"
+            out.append({
+                "method": name,
+                "recipe": dict(zip(DYES, recipe.conc)),
+                "predicted_delta_e00": round(float(d), 2),
+                "gate": gate,
+                "why": why,
+            })
+        return {"candidates": out}
+
+    def run_shot(self, recipe: dict, dip_seconds: float = 90.0, rationale: str = "") -> dict:
+        """Gate, dispense, dip, dry, photograph, measure."""
+        conc = tuple(float(recipe.get(d, 0.0)) for d in DYES)
+        recipe = Recipe(conc, dip_seconds)
+        # A rig that can only measure coarsely (a person with a syringe) reports what it will really dispense.
+        quantize = getattr(self.rig, "quantize", None)
+        if quantize:
+            recipe = quantize(recipe)
+        if len(self.shots) >= self.max_shots:
+            return {"refused": "shot budget exhausted; escalate or stop"}
+        try:
+            check_recipe(recipe, self.rig.limits, self.rig.inventory)
+        except GateError as exc:
+            self.emit("gate", recipe=recipe.describe(), refused=str(exc))
+            return {"refused": str(exc)}
+        self.emit("dispense", recipe=recipe.describe(), index=len(self.shots) + 1, rationale=rationale)
+        result = execute(self.rig, recipe)
+        m = measure(result.image_jpeg, self.segmenter, with_target=True, seed=len(self.shots))
+        if m.ok and m.target_lab is not None:
+            self._side_by_side(m)
+        de = float(delta_e_2000(m.lab, self.target_lab)) if m.ok else float("nan")
+        shot = Shot(len(self.shots) + 1, recipe, m, de, rationale, result.image_jpeg, overlay(result.image_jpeg, m))
+        self.shots.append(shot)
+        self.emit("shot", index=shot.index, recipe=recipe.describe(), lab=m.lab.tolist(),
+                  hex=lab_to_hex(m.lab), delta_e=de, flags=m.flags, rationale=rationale)
+        if m.ok and de <= self.tolerance:
+            self.status = "matched"
+            self.emit("status", status=self.status, shot=shot.index, delta_e=de)
+        return self._shot_summary(shot) | {"matched": m.ok and de <= self.tolerance}
+
+    def _side_by_side(self, m) -> None:
+        """The customer's cloth lies beside the strip: compare them under the same light.
+
+        A target read from a photo taken elsewhere carries that photo's light and camera
+        error; the cloth in the dip photo shares the strip's. Re-anchor to the running mean
+        of those side-by-side readings, unless the "cloth" is clearly something else.
+        """
+        seen = np.asarray(m.target_lab, dtype=float)
+        if float(delta_e_2000(seen, self.target_lab if not self.side_readings else np.mean(self.side_readings, axis=0))) > SIDE_BY_SIDE_MAX_DE00:
+            m.flags.append("WARN: another cloth is in the photo but it is not the target colour; ignored it")
+            return
+        first = not self.side_readings
+        self.side_readings.append(seen)
+        before = self.target_lab
+        self.target_lab = np.mean(self.side_readings, axis=0)
+        m.flags.append("NOTE: compared side by side with the customer's cloth in the same photo")
+        if first:
+            self.note(f"The customer's cloth is in the photo, so I compare the strip with it under the same light. "
+                      f"Target re-read as {lab_to_hex(self.target_lab)} (was {lab_to_hex(before)}, "
+                      f"dE00 {float(delta_e_2000(before, self.target_lab)):.1f} apart).", who="agent")
+            for s in self.shots:  # earlier dips are judged against the better target too
+                s.delta_e = float(delta_e_2000(s.measurement.lab, self.target_lab)) if s.measurement.ok else float("nan")
+
+    def declare_out_of_gamut(self) -> dict:
+        best = self.best
+        self.status = "out_of_gamut"
+        text = (
+            f"Target is outside what {', '.join(DYES)} can reach on cotton within the "
+            f"{self.rig.limits.max_conc_g_per_l:.0f} {UNIT} safety cap. Closest achievable: dip {best.index}, "
+            f"{lab_to_hex(best.measurement.lab)}, dE00 {best.delta_e:.2f}. A darker or different dye is needed."
+        )
+        self.note(text, who="agent")
+        self.emit("status", status=self.status, reason=text, shot=best.index, delta_e=best.delta_e)
+        return {"out_of_gamut": True, "closest_shot": best.index, "delta_e00": round(best.delta_e, 2)}
+
+    def escalate(self, reason: str) -> dict:
+        self.status = "escalated"
+        self.note(f"Escalated to the colourist: {reason}", who="agent")
+        self.emit("status", status=self.status, reason=reason)
+        return {"escalated": True}
+
+    # -- numerics ----------------------------------------------------------------
+    def _jac(self, conc: np.ndarray) -> np.ndarray:
+        # Sensitivity from the model refitted to every real dip. Learning it from the dips
+        # themselves (Broyden, or ridge regression on nearby dips) chases measurement noise
+        # when steps are whole drops: tried, and it matched fewer shades.
+        return self._model_jacobian(conc)
+
+    def _model_jacobian(self, conc: np.ndarray) -> np.ndarray:
+        h = 0.05
+        base = self.model.predict_lab(conc)
+        cols = []
+        for i in range(len(conc)):
+            c = conc.copy()
+            c[i] += h
+            cols.append((self.model.predict_lab(c) - base) / h)
+        return np.stack(cols, axis=1)
+
+    def _correction_step(self, best: Shot) -> np.ndarray | None:
+        c0 = best.recipe.as_array()
+        J = self._jac(c0)
+        err = self.target_lab - best.measurement.lab
+        # Damped least squares; weight lightness less, as CIEDE2000 does.
+        Wt = np.diag([0.6, 1.0, 1.0])
+        A = Wt @ J
+        lam = 0.05
+        dc = np.linalg.solve(A.T @ A + lam * np.eye(len(c0)), A.T @ (Wt @ err))
+        return np.clip(c0 + dc, 0, self.rig.limits.max_conc_g_per_l)

src/rangrez/card.py

@@ -0,0 +1,37 @@
+"""The printed reference card that sits in every frame.
+
+A 24-patch ColorChecker-style layout (4 rows x 6 columns). The reference
+values are the published sRGB approximations of the Classic chart; print the
+card in `hardware/colour_card.pdf` and every shot is colour-corrected against it,
+so lighting drift and camera auto-white-balance cancel out.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+
+from .colour import srgb_to_linear
+
+PATCH_NAMES = [
+    "dark skin", "light skin", "blue sky", "foliage", "blue flower", "bluish green",
+    "orange", "purplish blue", "moderate red", "purple", "yellow green", "orange yellow",
+    "blue", "green", "red", "yellow", "magenta", "cyan",
+    "white", "neutral 8", "neutral 6.5", "neutral 5", "neutral 3.5", "black",
+]
+
+PATCH_SRGB_8BIT = np.array(
+    [
+        [115, 82, 68], [194, 150, 130], [98, 122, 157], [87, 108, 67], [133, 128, 177], [103, 189, 170],
+        [214, 126, 44], [80, 91, 166], [193, 90, 99], [94, 60, 108], [157, 188, 64], [224, 163, 46],
+        [56, 61, 150], [70, 148, 73], [175, 54, 60], [231, 199, 31], [187, 86, 149], [8, 133, 161],
+        [243, 243, 242], [200, 200, 200], [160, 160, 160], [122, 122, 121], [85, 85, 85], [52, 52, 52],
+    ],
+    dtype=float,
+)
+
+ROWS, COLS = 4, 6
+
+
+def reference_linear() -> np.ndarray:
+    """(24, 3) linear sRGB reference values of the card patches."""
+    return srgb_to_linear(PATCH_SRGB_8BIT / 255.0)

src/rangrez/colour.py

@@ -0,0 +1,158 @@
+"""Colour science: sRGB <-> CIELAB, CIEDE2000, and camera colour correction.
+
+Everything colour-related that has a right answer lives here, so the agent
+never has to reason about it in prose.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+
+# D65 reference white for the sRGB path (CIE 1931 2 deg), Y normalised to 100.
+D65_WHITE = np.array([95.047, 100.0, 108.883])
+
+# Linear sRGB -> XYZ (D65), IEC 61966-2-1.
+_RGB_TO_XYZ = np.array(
+    [
+        [0.4124564, 0.3575761, 0.1804375],
+        [0.2126729, 0.7151522, 0.0721750],
+        [0.0193339, 0.1191920, 0.9503041],
+    ]
+)
+_XYZ_TO_RGB = np.linalg.inv(_RGB_TO_XYZ)
+
+
+def srgb_to_linear(rgb: np.ndarray) -> np.ndarray:
+    """sRGB in [0, 1] -> linear light in [0, 1]."""
+    rgb = np.asarray(rgb, dtype=float)
+    return np.where(rgb <= 0.04045, rgb / 12.92, ((rgb + 0.055) / 1.055) ** 2.4)
+
+
+def linear_to_srgb(lin: np.ndarray) -> np.ndarray:
+    lin = np.clip(np.asarray(lin, dtype=float), 0.0, 1.0)
+    return np.where(lin <= 0.0031308, lin * 12.92, 1.055 * lin ** (1 / 2.4) - 0.055)
+
+
+def linear_rgb_to_xyz(lin: np.ndarray) -> np.ndarray:
+    return np.asarray(lin, dtype=float) @ _RGB_TO_XYZ.T * 100.0
+
+
+def xyz_to_linear_rgb(xyz: np.ndarray) -> np.ndarray:
+    return (np.asarray(xyz, dtype=float) / 100.0) @ _XYZ_TO_RGB.T
+
+
+def xyz_to_lab(xyz: np.ndarray, white: np.ndarray = D65_WHITE) -> np.ndarray:
+    t = np.asarray(xyz, dtype=float) / white
+    delta = 6 / 29
+    f = np.where(t > delta**3, np.cbrt(t), t / (3 * delta**2) + 4 / 29)
+    L = 116 * f[..., 1] - 16
+    a = 500 * (f[..., 0] - f[..., 1])
+    b = 200 * (f[..., 1] - f[..., 2])
+    return np.stack([L, a, b], axis=-1)
+
+
+def lab_to_xyz(lab: np.ndarray, white: np.ndarray = D65_WHITE) -> np.ndarray:
+    lab = np.asarray(lab, dtype=float)
+    fy = (lab[..., 0] + 16) / 116
+    fx = fy + lab[..., 1] / 500
+    fz = fy - lab[..., 2] / 200
+    delta = 6 / 29
+    f = np.stack([fx, fy, fz], axis=-1)
+    t = np.where(f > delta, f**3, 3 * delta**2 * (f - 4 / 29))
+    return t * white
+
+
+def srgb_to_lab(rgb: np.ndarray) -> np.ndarray:
+    """sRGB in [0, 1] -> CIELAB (D65)."""
+    return xyz_to_lab(linear_rgb_to_xyz(srgb_to_linear(rgb)))
+
+
+def lab_to_srgb(lab: np.ndarray) -> np.ndarray:
+    return linear_to_srgb(xyz_to_linear_rgb(lab_to_xyz(lab)))
+
+
+def hex_to_lab(hex_colour: str) -> np.ndarray:
+    h = hex_colour.lstrip("#")
+    rgb = np.array([int(h[i : i + 2], 16) for i in (0, 2, 4)]) / 255.0
+    return srgb_to_lab(rgb)
+
+
+def lab_to_hex(lab: np.ndarray) -> str:
+    rgb = np.round(lab_to_srgb(lab) * 255).astype(int)
+    return "#" + "".join(f"{int(v):02x}" for v in np.clip(rgb, 0, 255))
+
+
+def delta_e_2000(lab1: np.ndarray, lab2: np.ndarray) -> np.ndarray:
+    """CIEDE2000 colour difference (Sharma, Wu & Dalal 2005), kL = kC = kH = 1."""
+    lab1 = np.asarray(lab1, dtype=float)
+    lab2 = np.asarray(lab2, dtype=float)
+    L1, a1, b1 = lab1[..., 0], lab1[..., 1], lab1[..., 2]
+    L2, a2, b2 = lab2[..., 0], lab2[..., 1], lab2[..., 2]
+
+    C1 = np.hypot(a1, b1)
+    C2 = np.hypot(a2, b2)
+    C_bar = (C1 + C2) / 2
+    G = 0.5 * (1 - np.sqrt(C_bar**7 / (C_bar**7 + 25**7)))
+    a1p = (1 + G) * a1
+    a2p = (1 + G) * a2
+    C1p = np.hypot(a1p, b1)
+    C2p = np.hypot(a2p, b2)
+    h1p = np.degrees(np.arctan2(b1, a1p)) % 360
+    h2p = np.degrees(np.arctan2(b2, a2p)) % 360
+
+    dLp = L2 - L1
+    dCp = C2p - C1p
+    dhp = h2p - h1p
+    dhp = np.where(dhp > 180, dhp - 360, dhp)
+    dhp = np.where(dhp < -180, dhp + 360, dhp)
+    dhp = np.where(C1p * C2p == 0, 0.0, dhp)
+    dHp = 2 * np.sqrt(C1p * C2p) * np.sin(np.radians(dhp / 2))
+
+    Lp_bar = (L1 + L2) / 2
+    Cp_bar = (C1p + C2p) / 2
+    hsum = h1p + h2p
+    hp_bar = np.where(
+        C1p * C2p == 0,
+        hsum,
+        np.where(
+            np.abs(h1p - h2p) <= 180,
+            hsum / 2,
+            np.where(hsum < 360, (hsum + 360) / 2, (hsum - 360) / 2),
+        ),
+    )
+    T = (
+        1
+        - 0.17 * np.cos(np.radians(hp_bar - 30))
+        + 0.24 * np.cos(np.radians(2 * hp_bar))
+        + 0.32 * np.cos(np.radians(3 * hp_bar + 6))
+        - 0.20 * np.cos(np.radians(4 * hp_bar - 63))
+    )
+    d_theta = 30 * np.exp(-(((hp_bar - 275) / 25) ** 2))
+    R_C = 2 * np.sqrt(Cp_bar**7 / (Cp_bar**7 + 25**7))
+    S_L = 1 + 0.015 * (Lp_bar - 50) ** 2 / np.sqrt(20 + (Lp_bar - 50) ** 2)
+    S_C = 1 + 0.045 * Cp_bar
+    S_H = 1 + 0.015 * Cp_bar * T
+    R_T = -np.sin(np.radians(2 * d_theta)) * R_C
+    return np.sqrt(
+        (dLp / S_L) ** 2
+        + (dCp / S_C) ** 2
+        + (dHp / S_H) ** 2
+        + R_T * (dCp / S_C) * (dHp / S_H)
+    )
+
+
+def fit_colour_correction(measured_lin: np.ndarray, reference_lin: np.ndarray) -> np.ndarray:
+    """Least-squares affine map (4x3) from camera linear RGB to reference linear sRGB.
+
+    measured_lin, reference_lin: (n_patches, 3). Returns M such that
+    [rgb, 1] @ M ~= reference.
+    """
+    X = np.hstack([measured_lin, np.ones((len(measured_lin), 1))])
+    M, *_ = np.linalg.lstsq(X, reference_lin, rcond=None)
+    return M
+
+
+def apply_colour_correction(M: np.ndarray, lin: np.ndarray) -> np.ndarray:
+    lin = np.asarray(lin, dtype=float)
+    X = np.concatenate([lin, np.ones(lin.shape[:-1] + (1,))], axis=-1)
+    return X @ M

src/rangrez/esp32_emulator.py

@@ -0,0 +1,213 @@
+"""A virtual Rangrez rig: the ESP32 firmware's protocol and hard limits on a pseudo serial port,
+with the simulator's hidden dye chemistry and a light-box camera that records video.
+
+The limits are read from firmware/rangrez_rig/rangrez_rig.ino itself, so the emulator
+refuses exactly what the board refuses. The real SerialRig driver talks to it over a
+real (pseudo) serial port and grabs frames from its camera video with OpenCV, so the
+whole robot path runs end to end without a single part bought.
+
+    RANGREZ_RIG=serial RANGREZ_PORT=virtual uv run rangrez-lab
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import tempfile
+import termios
+import threading
+import time
+import tty
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+from .km import reflectance_to_lab
+from .sim import VirtualRig
+
+FIRMWARE = Path(__file__).parents[2] / "firmware" / "rangrez_rig" / "rangrez_rig.ino"
+
+
+def firmware_limits(path: Path = FIRMWARE) -> dict[str, float]:
+    """The `const float MAX_*` limits and pump calibration, as compiled into the board."""
+    src = path.read_text()
+    limits = {k: float(v) for k, v in re.findall(r"const float (MAX_\w+) = ([\d.]+);", src)}
+    limits["msPerMl"] = [float(v) for v in re.search(r"msPerMl\[5\] = \{([^}]*)\}", src).group(1).split(",")]
+    return limits
+
+
+@dataclass
+class VirtualESP32:
+    """Speaks the firmware's line protocol. time_scale=0 skips the real waiting."""
+
+    camera_path: str = field(default_factory=lambda: tempfile.mkstemp(suffix=".mp4", prefix="rangrez-cam-")[1])
+    scene: str = "light_box"
+    card: bool = True
+    seed: int = 21
+    time_scale: float = 0.0
+    pump_rate_error: tuple[float, ...] = (0.97, 1.04, 1.01, 1.0)  # real pumps vs their calibration
+    pump_cv: float = 0.02
+
+    def __post_init__(self) -> None:
+        self.limits = firmware_limits()
+        self.rig = VirtualRig(seed=self.seed, scene=self.scene, card=self.card)
+        self.rng = np.random.default_rng(self.seed)
+        self.bath_ml = 0.0
+        self.dye_ml = np.zeros(3)
+        self.estop = False
+        self.light = False
+        self.log: list[dict] = []
+        self._write_video()  # the camera shows the empty light box until the first dip
+
+    # -- e-stop button ----------------------------------------------------------
+    def press_estop(self) -> None:
+        self.estop = True
+
+    def _wait(self, seconds: float) -> bool:
+        end = time.monotonic() + seconds * self.time_scale
+        while time.monotonic() < end:
+            if self.estop:
+                return False
+            time.sleep(0.005)
+        return not self.estop
+
+    # -- protocol ---------------------------------------------------------------
+    def handle(self, line: str) -> dict:
+        try:
+            doc = json.loads(line)
+        except json.JSONDecodeError:
+            return self._reply("?", "bad json")
+        cmd = str(doc.get("cmd", ""))
+        self.log.append(doc)
+        if self.estop and cmd not in ("reset", "ping"):
+            return self._reply(cmd, "e-stop latched; send reset")
+        L = self.limits
+        if cmd == "ping":
+            return {"fw": "rangrez-1", "estop": self.estop, "bath_ml": self.bath_ml}
+        if cmd == "reset":
+            self.estop = False
+            return self._reply(cmd)
+        if cmd == "pump":
+            pid, ml = int(doc.get("id", -1)), float(doc.get("ml", 0))
+            if not 0 <= pid <= 3:
+                return self._reply(cmd, "pump id must be 0-3 (use drain for 4)")
+            if ml <= 0 or ml > L["MAX_PUMP_ML"]:
+                return self._reply(cmd, "ml out of range")
+            if self.bath_ml + ml > L["MAX_BATH_ML"]:
+                return self._reply(cmd, "cup would overflow")
+            ok = self._wait(ml * L["msPerMl"][pid] / 1000)
+            delivered = ml * self.pump_rate_error[pid] * self.rng.normal(1, self.pump_cv)
+            if pid < 3:
+                self.dye_ml[pid] += delivered
+            self.bath_ml += ml  # worst case, for the overflow check
+            return self._reply(cmd) if ok else self._reply(cmd, "e-stop during pump")
+        if cmd == "drain":
+            s = float(doc.get("seconds", 0))
+            if s <= 0 or s > L["MAX_DRAIN_S"]:
+                return self._reply(cmd, "drain seconds out of range")
+            if not self._wait(s):
+                return self._reply(cmd, "e-stop during drain")
+            self.bath_ml, self.dye_ml = 0.0, np.zeros(3)
+            return self._reply(cmd)
+        if cmd == "stir":
+            s = float(doc.get("seconds", 0))
+            if s <= 0 or s > L["MAX_STIR_S"]:
+                return self._reply(cmd, "stir seconds out of range")
+            return self._reply(cmd) if self._wait(s) else self._reply(cmd, "e-stop during stir")
+        if cmd == "dip":
+            s = float(doc.get("seconds", 0))
+            if s < 10 or s > L["MAX_DIP_S"]:
+                return self._reply(cmd, "dip seconds out of range")
+            if self.bath_ml < 10:
+                return self._reply(cmd, "cup is nearly empty")
+            if not self._wait(s):
+                return self._reply(cmd, "e-stop during dip")
+            self._dye_segment(s)
+            return self._reply(cmd)
+        if cmd == "advance":
+            mm = float(doc.get("mm", 0))
+            if mm <= 0 or mm > L["MAX_ADVANCE_MM"]:
+                return self._reply(cmd, "advance mm out of range")
+            return self._reply(cmd) if self._wait(mm / 30) else self._reply(cmd, "e-stop")
+        if cmd == "dry":
+            s = float(doc.get("seconds", 0))
+            if s <= 0 or s > L["MAX_DRY_S"]:
+                return self._reply(cmd, "dry seconds out of range")
+            return self._reply(cmd) if self._wait(s) else self._reply(cmd, "e-stop during dry")
+        if cmd == "light":
+            self.light = bool(doc.get("on", False))
+            if self.light:
+                self._write_video()
+            return self._reply(cmd)
+        if cmd == "calibrate":
+            pid, ms = int(doc.get("id", -1)), float(doc.get("ms_per_ml", 0))
+            if not 0 <= pid <= 3 or not 100 <= ms <= 5000:
+                return self._reply(cmd, "bad calibration")
+            L["msPerMl"][pid] = ms
+            return self._reply(cmd)
+        return self._reply(cmd, "unknown command")
+
+    def _reply(self, cmd: str, error: str | None = None) -> dict:
+        out = {"error": error} if error else {"done": cmd}
+        return {**out, "bath_ml": self.bath_ml}
+
+    # -- the physical world -------------------------------------------------------
+    def _dye_segment(self, dip_seconds: float) -> None:
+        r = self.rig
+        dye_ml = self.dye_ml * r.limits.bath_ml / max(self.bath_ml, 1e-6)  # as if in the nominal bath
+        conc = dye_ml / r.limits.bath_ml * r.limits.stock_g_per_l
+        saturated = conc / (1 + conc / r.saturation_g_per_l)
+        time_factor = (1 - np.exp(-dip_seconds / 45)) / (1 - np.exp(-90 / 45))
+        r.last_swatch_lab = reflectance_to_lab(r.world.reflectance(saturated * time_factor))
+
+    def _write_video(self, frames: int = 14) -> None:
+        """What the light-box webcam sees: a short clip with a little sensor noise per frame."""
+        jpeg = self.rig.render()
+        img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR)
+        h, w = img.shape[:2]
+        tmp = self.camera_path + ".tmp.mp4"
+        out = cv2.VideoWriter(tmp, cv2.VideoWriter_fourcc(*"mp4v"), 10, (w, h))
+        for _ in range(frames):
+            noisy = np.clip(img.astype(np.int16) + self.rng.normal(0, 1.5, img.shape).astype(np.int16), 0, 255)
+            out.write(noisy.astype(np.uint8))
+        out.release()
+        os.replace(tmp, self.camera_path)
+
+    @property
+    def true_lab(self) -> np.ndarray | None:
+        return self.rig.last_swatch_lab
+
+    # -- serial port -------------------------------------------------------------
+    def serve(self) -> str:
+        """Open a pseudo serial port, answer on it from a thread, return its device path."""
+        master, slave = os.openpty()
+        tty.setraw(slave)
+        attrs = termios.tcgetattr(slave)
+        attrs[3] &= ~termios.ECHO
+        termios.tcsetattr(slave, termios.TCSANOW, attrs)
+        self._slave_fd = slave  # keep it open so the port stays alive between connections
+        threading.Thread(target=self._loop, args=(master,), daemon=True).start()
+        return os.ttyname(slave)
+
+    def _loop(self, fd: int) -> None:
+        buf = b""
+        while True:
+            try:
+                chunk = os.read(fd, 1024)
+            except OSError:
+                return
+            buf += chunk
+            while b"\n" in buf:
+                line, buf = buf.split(b"\n", 1)
+                if line.strip():
+                    os.write(fd, (json.dumps(self.handle(line.decode(errors="replace"))) + "\n").encode())
+
+
+def start(**kw) -> tuple[VirtualESP32, str]:
+    """A running virtual rig: (board, serial port path). Point the camera at board.camera_path."""
+    board = VirtualESP32(**kw)
+    return board, board.serve()
+

src/rangrez/glimmer.py

@@ -0,0 +1,73 @@
+"""Glimmer: Muse Glimmer 30B fine-tuned on this lab's own dips, proposing the first recipe.
+
+The supplier's model knows the dyes in general; Glimmer learned this dye lot through this
+camera, from every recipe the lab dipped and the colour SAM 3.1 measured on the cloth
+(eval/glimmer.py). The open LoRA weights run anywhere; on Fireworks set
+
+    GLIMMER_MODEL=accounts/<account>/models/<model>#accounts/<account>/deployments/<deployment>
+
+and the first dip of every campaign gets a "glimmer" candidate.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+
+import numpy as np
+
+from .colour import lab_to_hex
+from .profiles import PROFILE
+
+INFERENCE_URL = "https://api.fireworks.ai/inference/v1"
+SYSTEM = (
+    f"You are Glimmer, the colourist of a small cotton dye lab. Dyes: {', '.join(PROFILE.names)} "
+    f"({PROFILE.description}). Given a target colour measured in CIELAB (D65), reply with only the "
+    f"recipe for the first dip as JSON with one key per dye, in {PROFILE.unit}, each between 0 and {PROFILE.max_conc:g}."
+)
+
+
+def prompt(lab) -> str:
+    L, a, b = (float(v) for v in lab)
+    return f"Target L* {L:.1f}, a* {a:.1f}, b* {b:.1f} ({lab_to_hex(np.array([L, a, b]))}). First dip recipe?"
+
+
+def answer(conc) -> str:
+    return json.dumps({d: round(float(c), 2) for d, c in zip(PROFILE.names, conc)})
+
+
+def parse(text: str) -> np.ndarray | None:
+    try:
+        obj = json.loads(text[text.index("{"): text.rindex("}") + 1])
+        return np.clip([float(obj.get(d, 0.0)) for d in PROFILE.names], 0, PROFILE.max_conc)
+    except (ValueError, json.JSONDecodeError, TypeError, AttributeError):
+        return None
+
+
+def configured() -> bool:
+    return bool(os.environ.get("GLIMMER_MODEL") and os.environ.get("FIREWORKS_API_KEY"))
+
+
+def first_guess(target_lab, log=None, model: str | None = None) -> np.ndarray | None:
+    """Glimmer's first recipe for target_lab, or None if it is not configured or answers badly."""
+    from openai import OpenAI
+
+    from .vision import ApiCall
+
+    model = model or os.environ.get("GLIMMER_MODEL")
+    if not model:
+        return None
+    client = OpenAI(base_url=INFERENCE_URL, api_key=os.environ["FIREWORKS_API_KEY"].strip(), timeout=60, max_retries=1)
+    started = time.monotonic()
+    try:
+        r = client.chat.completions.create(model=model, temperature=0, max_tokens=400, messages=[
+            {"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt(target_lab)}])
+        conc = parse(r.choices[0].message.content or "")
+    except Exception as exc:  # the lab works without Glimmer: never fail a campaign over it
+        if log is not None:
+            log.add(ApiCall("muse-glimmer-30b (tuned)", "first dip recipe", time.monotonic() - started, False, str(exc)[:200]))
+        return None
+    if log is not None:
+        log.add(ApiCall("muse-glimmer-30b (tuned)", "first dip recipe", time.monotonic() - started, conc is not None))
+    return conc

src/rangrez/km.py

@@ -0,0 +1,152 @@
+"""Spectral Kubelka-Munk model of reactive dye on cotton.
+
+This is the colourist's model: each dye adds absorption (K/S) to the fibre in
+proportion to how much of it fixes, and reflectance follows from total K/S.
+The agent holds one of these as its hypothesis and refits it after every shot;
+the simulator holds a different, hidden one as "reality".
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+import numpy as np
+from scipy.optimize import least_squares
+
+from .colour import delta_e_2000, xyz_to_lab
+
+WAVELENGTHS = np.arange(400, 701, 20)  # 16 bands, nm
+
+# CIE 1931 2-degree colour matching functions at 20 nm.
+_CMF = np.array(
+    [
+        [0.0143, 0.0004, 0.0679], [0.1344, 0.0040, 0.6456], [0.3483, 0.0230, 1.7471],
+        [0.2908, 0.0600, 1.6692], [0.0956, 0.1390, 0.8130], [0.0049, 0.3230, 0.2720],
+        [0.0633, 0.7100, 0.0782], [0.2904, 0.9540, 0.0203], [0.5945, 0.9950, 0.0039],
+        [0.9163, 0.8700, 0.0017], [1.0622, 0.6310, 0.0008], [0.8544, 0.3810, 0.0002],
+        [0.4479, 0.1750, 0.0000], [0.1649, 0.0610, 0.0000], [0.0468, 0.0170, 0.0000],
+        [0.0114, 0.0041, 0.0000],
+    ]
+)
+
+# CIE D65 relative spectral power at 20 nm.
+_D65 = np.array(
+    [82.75, 93.43, 104.86, 117.81, 115.92, 109.35, 104.79, 104.41,
+     100.00, 95.79, 90.01, 87.70, 83.70, 80.21, 78.28, 71.61]
+)
+
+_WEIGHTS = _CMF * _D65[:, None]
+_NORM = 100.0 / _WEIGHTS[:, 1].sum()
+SPECTRAL_WHITE = _WEIGHTS.sum(axis=0) * _NORM  # XYZ of a perfect white
+
+
+def reflectance_to_xyz(R: np.ndarray) -> np.ndarray:
+    return (np.asarray(R, dtype=float) @ _WEIGHTS) * _NORM
+
+
+def reflectance_to_lab(R: np.ndarray) -> np.ndarray:
+    return xyz_to_lab(reflectance_to_xyz(R), white=SPECTRAL_WHITE)
+
+
+def ks_to_reflectance(ks: np.ndarray) -> np.ndarray:
+    ks = np.maximum(np.asarray(ks, dtype=float), 0.0)
+    return 1 + ks - np.sqrt(ks**2 + 2 * ks)
+
+
+def reflectance_to_ks(R: np.ndarray) -> np.ndarray:
+    R = np.clip(np.asarray(R, dtype=float), 1e-4, 1.0)
+    return (1 - R) ** 2 / (2 * R)
+
+
+def absorption_band(peak_nm: float, width_nm: float, strength: float) -> np.ndarray:
+    """A Gaussian absorption band, as K/S per g/L of fixed dye."""
+    return strength * np.exp(-0.5 * ((WAVELENGTHS - peak_nm) / width_nm) ** 2)
+
+
+# Bleached cotton: near white, slightly warm.
+COTTON_KS = reflectance_to_ks(np.linspace(0.80, 0.90, len(WAVELENGTHS)))
+
+
+@dataclass
+class DyeModel:
+    """K/S of cotton = substrate + sum(gain_i * c_i * unit_ks_i)."""
+
+    names: list[str]
+    unit_ks: np.ndarray  # (n_dyes, n_bands) K/S per g/L in the bath, at gain 1
+    substrate_ks: np.ndarray = field(default_factory=lambda: COTTON_KS.copy())
+    gains: np.ndarray | None = None  # per-dye uptake multipliers, fitted from shots
+
+    def __post_init__(self) -> None:
+        if self.gains is None:
+            self.gains = np.ones(len(self.names))
+
+    def ks(self, conc: np.ndarray) -> np.ndarray:
+        conc = np.maximum(np.asarray(conc, dtype=float), 0.0)
+        return self.substrate_ks + (conc * self.gains) @ self.unit_ks
+
+    def reflectance(self, conc: np.ndarray) -> np.ndarray:
+        return ks_to_reflectance(self.ks(conc))
+
+    def predict_lab(self, conc: np.ndarray) -> np.ndarray:
+        return reflectance_to_lab(self.reflectance(conc))
+
+    def copy(self) -> DyeModel:
+        return DyeModel(list(self.names), self.unit_ks.copy(), self.substrate_ks.copy(), self.gains.copy())
+
+
+def nominal_procion_cmy() -> DyeModel:
+    """The supplier's nominal reactive set; kept for tests and the simulator."""
+    from .profiles import nominal_reactive
+
+    return nominal_reactive()
+
+
+def fit_gains(model: DyeModel, shots: list[tuple[np.ndarray, np.ndarray]]) -> DyeModel:
+    """Refit per-dye uptake gains to measured (concentration, Lab) shots.
+
+    Minimises CIEDE2000 between predicted and measured shades; returns a new model.
+    """
+    if not shots:
+        return model.copy()
+    concs = np.array([c for c, _ in shots])
+    labs = np.array([lab for _, lab in shots])
+
+    def residual(log_gains: np.ndarray) -> np.ndarray:
+        trial = model.copy()
+        trial.gains = np.exp(log_gains)
+        pred = np.array([trial.predict_lab(c) for c in concs])
+        # Mild prior pulling gains towards 1 keeps the fit sane with few shots.
+        return np.concatenate([delta_e_2000(pred, labs), 0.2 * log_gains])
+
+    # Wide bounds: household inks can be far stronger or weaker than any nominal guess.
+    result = least_squares(residual, np.log(model.gains), bounds=(np.log(0.05), np.log(20.0)))
+    fitted = model.copy()
+    fitted.gains = np.exp(result.x)
+    return fitted
+
+
+def solve_recipe(
+    model: DyeModel,
+    target_lab: np.ndarray,
+    max_conc: float = 20.0,
+    start: np.ndarray | None = None,
+) -> tuple[np.ndarray, float]:
+    """Concentrations (g/L) that the model predicts will match target_lab.
+
+    Returns (concentrations, predicted CIEDE2000).
+    """
+    n = len(model.names)
+    starts = [start] if start is not None else []
+    starts += [np.full(n, v) for v in (0.5, 2.0, 6.0)]
+    best = None
+    for x0 in starts:
+        x0 = np.clip(np.asarray(x0, dtype=float), 0.0, max_conc)
+        r = least_squares(
+            lambda c: np.atleast_1d(delta_e_2000(model.predict_lab(c), target_lab)),
+            x0,
+            bounds=(0.0, max_conc),
+        )
+        de = float(delta_e_2000(model.predict_lab(r.x), target_lab))
+        if best is None or de < best[1]:
+            best = (r.x, de)
+    return best

src/rangrez/make_card.py

@@ -0,0 +1,40 @@
+"""Render the printable 24-patch colour card used in every frame.
+
+    uv run rangrez-card            # writes hardware/colour_card.png (300 dpi, 112 x 76 mm)
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from PIL import Image, ImageDraw
+
+from .card import COLS, PATCH_SRGB_8BIT, ROWS
+
+DPI = 300
+PATCH_MM, GAP_MM, BORDER_MM = 16, 2, 3
+
+
+def render() -> Image.Image:
+    mm = DPI / 25.4
+    patch, gap, border = round(PATCH_MM * mm), round(GAP_MM * mm), round(BORDER_MM * mm)
+    w = 2 * border + COLS * patch + (COLS - 1) * gap
+    h = 2 * border + ROWS * patch + (ROWS - 1) * gap
+    img = Image.new("RGB", (w, h), (8, 8, 8))
+    draw = ImageDraw.Draw(img)
+    for i, rgb in enumerate(PATCH_SRGB_8BIT.astype(int)):
+        r, c = divmod(i, COLS)
+        x, y = border + c * (patch + gap), border + r * (patch + gap)
+        draw.rectangle([x, y, x + patch - 1, y + patch - 1], fill=tuple(int(v) for v in rgb))
+    return img
+
+
+def main() -> None:
+    out = Path("hardware/colour_card.png")
+    out.parent.mkdir(exist_ok=True)
+    render().save(out, dpi=(DPI, DPI))
+    print(f"wrote {out} — print at 100% scale on matte photo paper")
+
+
+if __name__ == "__main__":
+    main()

src/rangrez/photoreal.py

@@ -0,0 +1,240 @@
+"""A camera that takes realistic phone photos of the simulator's cloth.
+
+Scenes are real-looking photos (made with Muse Image) of undyed white cotton on
+printer paper, under desk lamps, daylight, tube lights, phone flash and more.
+SAM 3.1 masked each scene once (eval/scene_masks.py) and the masks are cached
+in scenes/masks. To photograph a dyed strip, the strip's pixels are recoloured
+to its true colour under the scene's own light, keeping the cloth's real folds,
+weave and shadows. The paper's true whiteness and tint are randomised and NOT
+told to the pipeline, the way real printer paper is never exactly white.
+"""
+
+from __future__ import annotations
+
+import io
+from dataclasses import dataclass
+from functools import cache, cached_property
+from pathlib import Path
+
+import cv2
+import numpy as np
+from PIL import Image
+from scipy.ndimage import gaussian_filter
+
+from .colour import lab_to_srgb, linear_to_srgb, srgb_to_linear
+from .vision import CARD_PROMPT, PAPER_PROMPTS, TARGET_PROMPT, Instance
+
+SCENE_DIR = Path(__file__).parent / "scenes"
+COTTON_LIN = 0.80  # linear reflectance of undyed cotton, to infer the light when there is no paper
+
+
+def scene_names(kind: str = "strip") -> list[str]:
+    """Scenes with a test strip ("strip"), or with only a garment to read ("target")."""
+    masks = SCENE_DIR / "masks"
+    strips = sorted(p.name[: -len("_strip.png")] for p in masks.glob("*_strip.png"))
+    if kind == "strip":
+        return strips
+    return sorted(n for n in (p.name[: -len("_target.png")] for p in masks.glob("*_target.png")) if n not in strips)
+
+
+@cache
+def _card_lin(printed: bool = False) -> np.ndarray:
+    from .make_card import render as card_image  # the same card people print
+
+    lin = srgb_to_linear(np.asarray(card_image(), dtype=float) / 255)
+    return _home_print(lin) if printed else lin
+
+
+def _home_print(lin: np.ndarray) -> np.ndarray:
+    """What a home inkjet on plain paper makes of the card: the printer can't reach saturated
+    colours, blacks print as dark grey, and every patch is a few dE00 off its reference."""
+    from .card import COLS, PATCH_SRGB_8BIT, ROWS
+    from .colour import lab_to_srgb as to_srgb
+    from .colour import srgb_to_lab
+
+    rng = np.random.default_rng(7)
+    out = lin.copy()
+    flat = out.reshape(-1, 3)
+    for ref in PATCH_SRGB_8BIT:
+        lab = srgb_to_lab(ref / 255.0)
+        L, a, b = lab
+        c = np.hypot(a, b)
+        squash = 0.8 if c > 45 else 0.92  # out-of-gamut colours compress
+        printed = np.array([max(L, 18.0) * 0.97 + 2.0, a * squash, b * squash]) + rng.normal(0, 1.2, 3)
+        sel = np.all(np.abs(flat - srgb_to_linear(ref / 255.0)) < 1e-6, axis=1)
+        flat[sel] = srgb_to_linear(np.clip(to_srgb(printed), 0, 1))
+    body = np.all(flat < 0.01, axis=1)
+    flat[body] = srgb_to_linear(np.full(3, 0.12))  # black body prints as dark grey
+    assert ROWS * COLS == len(PATCH_SRGB_8BIT)
+    return out
+
+
+def _mask(name: str, kind: str) -> np.ndarray | None:
+    p = SCENE_DIR / "masks" / f"{name}_{kind}.png"
+    return np.asarray(Image.open(p)) > 127 if p.exists() else None
+
+
+def _instance(mask: np.ndarray) -> Instance:
+    ys, xs = np.nonzero(mask)
+    return Instance(mask, (int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1))
+
+
+@dataclass(frozen=True, eq=False)
+class Scene:
+    name: str
+    lin: np.ndarray
+    strip: np.ndarray | None
+    paper: np.ndarray | None
+    target: np.ndarray | None
+    background: np.ndarray | None  # bedsheet or tablecloth SAM also calls "fabric"
+
+    @staticmethod
+    @cache
+    def load(name: str) -> Scene:
+        rgb = np.asarray(Image.open(SCENE_DIR / f"{name}.jpg").convert("RGB"), dtype=float) / 255
+        return Scene(name, srgb_to_linear(rgb), _mask(name, "strip"), _mask(name, "paper"), _mask(name, "target"),
+                     _mask(name, "background"))
+
+    @property
+    def cloth(self) -> np.ndarray:
+        out = np.zeros(self.lin.shape[:2], bool)
+        for m in (self.strip, self.target):
+            if m is not None:
+                out |= m
+        return out
+
+    @property
+    def bare(self) -> np.ndarray:
+        """Where a card can lie: the paper, or the table when there is no paper."""
+        base = self.paper if self.paper is not None else np.ones(self.lin.shape[:2], bool)
+        return base & ~self.cloth
+
+    def light(self) -> np.ndarray:
+        """Linear RGB of white under this scene's light, assuming nominal paper / cotton."""
+        if self.paper is not None:
+            return np.median(self.lin[self.bare], axis=0)
+        return np.median(self.lin[self.cloth], axis=0) / COTTON_LIN
+
+    @cached_property
+    def illumination(self) -> np.ndarray:
+        """Relative light across the frame (1 = median over the paper), read off the bare paper.
+
+        This is how the scene was lit: lamp falloff, window light, flash hotspot, shadows.
+        Where paper is visible the light is the paper's own (lightly smoothed) brightness.
+        Under the cloth it is filled in from the lit paper around it, leaving out cast
+        shadows beside the cloth so they do not darken the cloth itself.
+        """
+        h, w = self.lin.shape[:2]
+        if self.paper is None:
+            return np.ones((h, w, 3))
+        small, bare = self.lin[::4, ::4], self.bare[::4, ::4].astype(float)
+        weight = bare
+        for _ in range(2):
+            num = gaussian_filter(small * weight[..., None], (25, 25, 0))
+            den = gaussian_filter(weight, 25)[..., None]
+            under = num / np.maximum(den, 1e-9)
+            weight = bare * (small.mean(axis=2) > 0.8 * under.mean(axis=2))  # drop shadows, refit
+        den = gaussian_filter(bare, 3)
+        seen = gaussian_filter(small * bare[..., None], (3, 3, 0)) / np.maximum(den, 1e-9)[..., None]
+        trust = np.clip(den / 0.6, 0, 1)[..., None]
+        field = trust * seen + (1 - trust) * under
+        field /= np.median(field[bare > 0], axis=0)
+        return cv2.resize(field.astype(np.float32), (w, h), interpolation=cv2.INTER_LINEAR).astype(float)
+
+    def card_box(self, overlap: bool = False) -> tuple[int, int, int, int]:
+        """Card placement: fully on bare paper, or (overlap=True) partly over the cloth's edge."""
+        return _card_box(self.name, overlap)
+
+
+@cache
+def _card_box(name: str, overlap: bool) -> tuple[int, int, int, int]:
+    scene = Scene.load(name)
+    bare = scene.bare
+    ii = np.pad(bare.astype(np.int32).cumsum(0).cumsum(1), ((1, 0), (1, 0)))
+    for width in (0.22, 0.18, 0.15):  # a smaller print when the paper margin is narrow
+        cw = int(width * scene.lin.shape[1])
+        ch = int(cw * _card_lin().shape[0] / _card_lin().shape[1])
+        best, spot = -np.inf, (0, 0)
+        for y in range(0, bare.shape[0] - ch, 16):
+            for x in range(0, bare.shape[1] - cw, 16):
+                frac = (ii[y + ch, x + cw] - ii[y, x + cw] - ii[y + ch, x] + ii[y, x]) / (ch * cw)
+                score = -abs(frac - 0.7) if overlap else frac
+                if score > best:
+                    best, spot = score, (y, x)
+        if overlap or best > 0.97:
+            break
+    y, x = spot
+    return (x, y, x + cw, y + ch)
+
+
+def render(scene: Scene, strip_lab=None, target_lab=None, card: bool = False, seed: int = 0,
+           paper_seed: int | None = None, overlap: bool = False, printed_card: bool = False,
+           ) -> tuple[bytes, dict[str, np.ndarray]]:
+    """One phone photo and the masks of what is in it. Same paper_seed: same sheet of paper.
+
+    strip_lab=None leaves the strip undyed. Returns the JPEG and {"strip", "target",
+    "paper", "card"} masks (absent keys: not in the photo).
+    """
+    rng = np.random.default_rng(seed)
+    prng = np.random.default_rng(seed if paper_seed is None else paper_seed)
+    lin = scene.lin.copy()
+    # True paper (or cotton) whiteness, unknown to the pipeline: 0.86-0.96, slightly bluish (optical brighteners).
+    tint = prng.uniform(0.86, 0.96) * np.array([0.985, 0.995, 1.02]) * prng.normal(1, 0.01, 3)
+    light = scene.light() / tint if scene.paper is not None else scene.light()  # true illuminant
+    masks: dict[str, np.ndarray] = {}
+    for kind, lab in (("strip", strip_lab), ("target", target_lab)):
+        mask = getattr(scene, kind)
+        if mask is None:
+            continue
+        masks[kind] = mask
+        if lab is not None:
+            refl = srgb_to_linear(lab_to_srgb(np.asarray(lab, dtype=float)))
+            lit = scene.illumination[mask]
+            texture = scene.lin[mask] / lit  # folds, weave and shading of the white cloth
+            texture /= np.median(texture, axis=0)
+            lin[mask] = texture * lit * light * refl
+    if scene.paper is not None:
+        masks["paper"] = scene.paper
+    if scene.background is not None:
+        masks["background"] = scene.background
+    if card:
+        l, t, r, b = scene.card_box(overlap)
+        img = Image.fromarray((linear_to_srgb(_card_lin(printed_card)) * 255).astype(np.uint8)).resize((r - l, b - t))
+        # The card is lit by the light where it lies, like everything else in the scene.
+        card_lin = srgb_to_linear(np.asarray(img, dtype=float) / 255)
+        lin[t:b, l:r] = card_lin * light * scene.illumination[t:b, l:r]
+        cm = np.zeros(lin.shape[:2], bool)
+        cm[t:b, l:r] = True
+        masks["card"] = cm
+    # Camera: auto exposure and white balance wobble, sensor noise, JPEG.
+    lin = lin * rng.uniform(0.9, 1.1) * rng.normal(1, 0.02, 3) + rng.normal(0, 0.004, lin.shape)
+    out = (linear_to_srgb(lin) * 255).round().clip(0, 255).astype(np.uint8)
+    buf = io.BytesIO()
+    Image.fromarray(out).save(buf, format="JPEG", quality=int(rng.integers(82, 95)))
+    return buf.getvalue(), masks
+
+
+class SceneSegmenter:
+    """Offline stand-in for hosted SAM 3.1: answers prompts from the scene's cached SAM masks."""
+
+    def __init__(self, masks: dict[str, np.ndarray]):
+        self.masks = masks
+
+    def segment(self, image: Image.Image, jpeg: bytes, concept: str) -> list[Instance]:
+        if concept == CARD_PROMPT:
+            kinds = ["card"]
+        elif concept in PAPER_PROMPTS:
+            kinds = ["paper"]
+        elif concept == TARGET_PROMPT:
+            kinds = ["target"]
+        else:  # "fabric", "cloth", "kurta"...: SAM returns every piece of cloth, bedsheet included
+            kinds = ["strip", "target", "background"]
+        size = (image.height, image.width)
+        out = []
+        for k in kinds:
+            if k in self.masks:
+                m = self.masks[k]
+                if m.shape != size:  # e.g. a video codec trimmed a row
+                    m = cv2.resize(m.astype(np.uint8), size[::-1], interpolation=cv2.INTER_NEAREST) > 0
+                out.append(_instance(m))
+        return out

src/rangrez/profiles.py

@@ -0,0 +1,89 @@
+"""Dye sets the lab can run. Choose with RANGREZ_DYES=reactive|food.
+
+reactive: cold fibre-reactive dyes on soda-ash cotton, as a dye house uses.
+food:     supermarket food colours on plain cotton; a stand-in anyone can buy
+          tonight. Recipes are in % of the bath that is food colour.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import numpy as np
+
+from .km import DyeModel, absorption_band
+
+
+@dataclass(frozen=True)
+class Profile:
+    key: str
+    names: tuple[str, ...]
+    unit: str  # unit of a recipe entry, as shown to people and to the agent
+    stock_per_l: float  # stock concentration in the same unit, so mL = c / stock * bath
+    max_conc: float
+    description: str
+    nominal: Callable[[], DyeModel]
+    bottle_ml: float = 250.0  # stock on hand per dye
+    min_stock_ml: float = 5.0
+    step_ml: float = 0.1  # smallest amount a person can measure in assisted mode
+
+
+def nominal_reactive() -> DyeModel:
+    """Supplier's nominal turquoise, fuchsia, lemon yellow (cold reactive)."""
+    unit_ks = np.stack(
+        [
+            absorption_band(640, 55, 0.55),  # turquoise absorbs red
+            absorption_band(540, 40, 0.60),  # fuchsia absorbs green
+            absorption_band(430, 35, 0.50),  # lemon yellow absorbs blue
+        ]
+    )
+    return DyeModel(["turquoise", "fuchsia", "lemon yellow"], unit_ks)
+
+
+def nominal_food() -> DyeModel:
+    """Brilliant blue, allura red and tartrazine food colours staining cotton."""
+    unit_ks = np.stack(
+        [
+            absorption_band(630, 45, 0.30),  # blue (brilliant blue FCF peaks near 630 nm)
+            absorption_band(505, 45, 0.30),  # red (allura red peaks near 504 nm)
+            absorption_band(427, 35, 0.28),  # yellow (tartrazine peaks near 427 nm)
+        ]
+    )
+    return DyeModel(["blue", "red", "yellow"], unit_ks)
+
+
+def nominal_home() -> DyeModel:
+    """Household blue, red and yellow inks, per drop in a 20 mL bath."""
+    unit_ks = np.stack(
+        [
+            absorption_band(620, 55, 0.10),
+            absorption_band(515, 50, 0.10),
+            absorption_band(430, 40, 0.09),
+        ]
+    )
+    return DyeModel(["blue", "red", "yellow"], unit_ks)
+
+
+PROFILES = {
+    "reactive": Profile(
+        "reactive", ("turquoise", "fuchsia", "lemon yellow"), "g/L", 20.0, 10.0,
+        "cold fibre-reactive dyes on soda-ash-treated cotton; concentrations are g/L of dye in a 20 mL bath",
+        nominal_reactive,
+    ),
+    "food": Profile(
+        "food", ("blue", "red", "yellow"), "%", 100.0, 15.0,
+        "liquid food colours on white cotton; amounts are % of the 20 mL bath that is food colour (the rest is water); a person measures them with syringes to 0.1 mL",
+        nominal_food, bottle_ml=30.0, min_stock_ml=2.0,
+    ),
+    # Household inks (sketch-pen ink, food colour, haldi...) counted in drops of about 0.05 mL
+    # into a 20 mL (4 teaspoon) bath: c drops -> c * 0.05 mL, so stock = 20 / 0.05 = 400.
+    "home": Profile(
+        "home", ("blue", "red", "yellow"), "drops", 400.0, 40.0,
+        "household colours (sketch-pen ink, food colour) on white cotton; amounts are drops of each colour in 4 teaspoons (20 mL) of water, counted by a person",
+        nominal_home, bottle_ml=20.0, min_stock_ml=0.5, step_ml=0.05,
+    ),
+}
+
+PROFILE = PROFILES[os.environ.get("RANGREZ_DYES", "reactive")]

src/rangrez/recorder.py

@@ -0,0 +1,65 @@
+"""Every campaign, recorded to disk for review and for the demo video.
+
+    recordings/2026-11-02_101500_SerialRig/
+        meta.json          rig, dyes, agent, scene, tolerance
+        target.jpg         the customer's photo, if the target came from one (+ target.json)
+        events.jsonl       every event: Meta API calls with timings, Wonder's notes and tool calls,
+                           dispenses, gate refusals, measurements, status
+        dip01_photo.jpg    each dip exactly as the camera or phone sent it
+        dip01_sam.jpg      the same photo with SAM 3.1's card and strip boxes drawn on
+        state.json         the final campaign state
+
+Turn one into a video with `uv run python eval/demo_reel.py recordings/<run>`.
+Set RANGREZ_RECORD=0 to switch it off.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+import time
+from pathlib import Path
+
+
+class Recorder:
+    def __init__(self, root: str | None = None) -> None:
+        self.root = Path(root or os.environ.get("RANGREZ_RECORD_DIR", "recordings"))
+        self.enabled = os.environ.get("RANGREZ_RECORD", "1") != "0"
+        self.dir: Path | None = None
+        self.pending: list[tuple[str, bytes]] = []  # made before the campaign starts (the target photo)
+        self.lock = threading.Lock()
+
+    def stash(self, name: str, data: bytes | str | dict) -> None:
+        """Keep something for the next campaign's folder."""
+        self.pending.append((name, self._bytes(data)))
+
+    def begin(self, label: str, meta: dict) -> Path | None:
+        if not self.enabled:
+            return None
+        with self.lock:
+            self.dir = self.root / f"{time.strftime('%Y-%m-%d_%H%M%S')}_{label}"
+            self.dir.mkdir(parents=True, exist_ok=True)
+            (self.dir / "meta.json").write_text(json.dumps(meta, indent=2, default=str))
+            for name, data in self.pending:
+                (self.dir / name).write_bytes(data)
+            self.pending = []
+        return self.dir
+
+    def event(self, event: dict) -> None:
+        if self.dir is None:
+            return
+        with self.lock, open(self.dir / "events.jsonl", "a") as f:
+            f.write(json.dumps(event, default=str) + "\n")
+
+    def save(self, name: str, data: bytes | str | dict) -> None:
+        if self.dir is not None:
+            (self.dir / name).write_bytes(self._bytes(data))
+
+    @staticmethod
+    def _bytes(data: bytes | str | dict) -> bytes:
+        if isinstance(data, bytes):
+            return data
+        if isinstance(data, dict):
+            data = json.dumps(data, indent=2, default=str)
+        return data.encode()

src/rangrez/rig.py

@@ -0,0 +1,116 @@
+"""Recipes, the rig interface, and the deterministic safety gate.
+
+Model output never reaches a pump directly. The agent proposes a Recipe; the
+gate checks it against hard limits; only then does the executor turn it into
+pump volumes. The firmware enforces the same limits a second time.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Protocol
+
+import numpy as np
+
+from .profiles import PROFILE
+
+DYES = list(PROFILE.names)
+UNIT = PROFILE.unit
+
+
+@dataclass(frozen=True)
+class RigLimits:
+    bath_ml: float = 20.0  # liquor per shot
+    stock_g_per_l: float = PROFILE.stock_per_l  # stock concentration, in UNIT
+    max_conc_g_per_l: float = PROFILE.max_conc  # per dye, in the bath, in UNIT
+    max_dye_fraction: float = 0.9  # dye stock may fill at most this share of the bath
+    dip_seconds: tuple[float, float] = (30.0, 300.0)
+    min_stock_ml: float = PROFILE.min_stock_ml  # refuse to run a bottle dry
+
+
+@dataclass(frozen=True)
+class Recipe:
+    conc: tuple[float, ...]  # g/L in the bath, one per dye in DYES order
+    dip_seconds: float = 90.0
+
+    def as_array(self) -> np.ndarray:
+        return np.array(self.conc, dtype=float)
+
+    def describe(self) -> str:
+        parts = [f"{name} {c:.2f} {UNIT}" for name, c in zip(DYES, self.conc)]
+        return ", ".join(parts) + f", dip {self.dip_seconds:.0f} s"
+
+
+@dataclass
+class PumpPlan:
+    dye_ml: tuple[float, ...]
+    water_ml: float
+
+
+class GateError(ValueError):
+    """A recipe the gate refused. The message says why, for the agent to read."""
+
+
+@dataclass
+class Inventory:
+    stock_ml: list[float] = field(default_factory=lambda: [PROFILE.bottle_ml] * len(PROFILE.names))
+    strips: int = 60
+
+
+def check_recipe(recipe: Recipe, limits: RigLimits, inventory: Inventory) -> PumpPlan:
+    """Deterministic gate. Returns the pump plan or raises GateError."""
+    conc = recipe.as_array()
+    if conc.shape != (len(DYES),):
+        raise GateError(f"recipe needs {len(DYES)} concentrations, got {conc.shape[0]}")
+    if not np.all(np.isfinite(conc)):
+        raise GateError("recipe has a non-finite concentration")
+    if np.any(conc < 0):
+        raise GateError("concentrations must be >= 0")
+    if np.any(conc > limits.max_conc_g_per_l):
+        raise GateError(f"each dye is capped at {limits.max_conc_g_per_l} {UNIT}")
+    lo, hi = limits.dip_seconds
+    if not lo <= recipe.dip_seconds <= hi:
+        raise GateError(f"dip time must be between {lo:.0f} and {hi:.0f} s")
+    dye_ml = conc / limits.stock_g_per_l * limits.bath_ml
+    if dye_ml.sum() > limits.max_dye_fraction * limits.bath_ml:
+        raise GateError("total dye stock exceeds the bath volume; lower the concentrations")
+    for name, need, have in zip(DYES, dye_ml, inventory.stock_ml):
+        if have - need < limits.min_stock_ml:
+            raise GateError(f"{name} stock is too low ({have:.0f} mL); refill before continuing")
+    if inventory.strips <= 0:
+        raise GateError("out of cotton strips")
+    return PumpPlan(tuple(float(v) for v in dye_ml), float(limits.bath_ml - dye_ml.sum()))
+
+
+@dataclass
+class ShotResult:
+    image_jpeg: bytes
+    plan: PumpPlan
+    started_at: float
+    finished_at: float
+    extra: dict = field(default_factory=dict)
+
+
+class Rig(Protocol):
+    limits: RigLimits
+    inventory: Inventory
+
+    def run_shot(self, recipe: Recipe) -> ShotResult: ...
+
+    def capture(self) -> bytes: ...
+
+
+def execute(rig: Rig, recipe: Recipe) -> ShotResult:
+    """Gate, then run. The only path from a recipe to the pumps."""
+    plan = check_recipe(recipe, rig.limits, rig.inventory)
+    result = rig.run_shot(recipe)
+    for i, ml in enumerate(plan.dye_ml):
+        rig.inventory.stock_ml[i] -= ml
+    rig.inventory.strips -= 1
+    result.plan = plan
+    return result
+
+
+def now() -> float:
+    return time.time()