"""CellTriage — QC operator console. An inference-only interface over the artifacts built by ``src.pipelines.build_app_artifacts``. It loads fitted models from ``outputs/models/`` and a small demo subset; **no raw cycling data is shipped**. DESIGN PRIORITIES, in order: 1. **The budget advisor**, because Phase 8's flat cost frontier and 99.17% chamber-time release is the most actionable result in the project, and a slider the user moves themselves is more convincing than a sentence. 2. **The campaign-shift alarm as a STATE CHANGE**, not a disclaimer. Phase 10 found the model becomes *confidently wrong* under campaign shift — accuracy halves while intervals narrow — so a grey footnote is not an adequate response to that failure mode. Unfamiliar data changes what the result panel looks like. 3. **Decisions rendered visually.** Whether the confidence interval crosses a grade boundary is a spatial question; asking a reader to infer it from two numbers wastes the result. 4. **Limitations one click away**, never buried. Run locally: ``python -m app.app`` """ from __future__ import annotations import json import os import sys from pathlib import Path from typing import Any # HuggingFace Spaces launches `app_file` AS A SCRIPT -- `python app/app.py` -- # which puts app/ on sys.path instead of the repository root. `import app` then # resolves to this very file and `from app import panels` fails with a circular # import, while `python -m app.app` works fine. Local module-mode runs therefore # cannot detect the one failure mode that matters for deployment, so the repo # root is put first on the path before any first-party import. if __package__ in (None, ""): # pragma: no cover - only on script launch sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import gradio as gr import numpy as np import pandas as pd from scipy import stats from app import panels, theme # --------------------------------------------------------------------------- # ZeroGPU: why NOTHING here is decorated with @spaces.GPU # # This Space runs on ZeroGPU because it is the free tier available for new # Spaces, not because the workload needs a GPU. Every model is a CPU-bound # scikit-learn pipeline -- no neural networks anywhere in the project, by hard # constraint -- and a single prediction takes ~45 ms on CPU. # # AN EARLIER VERSION DECORATED THE INFERENCE PATH WITH @spaces.GPU AND THAT WAS # A REAL BUG, caught only on the deployed Space. `@spaces.GPU` REQUESTS a GPU # slot per call and bills the declared duration against a daily quota that is # 5 minutes on a free account. The campaign-shift check alone calls predict() # once per cell to build the reference distribution -- 22 calls -- so at the # declared 15 s per call a SINGLE page load reserved 330 s and exhausted the # entire day's quota. The deployed console failed with "You have exceeded your # ZeroGPU runs limit" while every local test passed, because locally the # decorator is documented as effect-free. # # SO THE INFERENCE PATH IS NOT DECORATED. What the platform still needs is a # GPU entry point to exist at all: with no `@spaces.GPU` anywhere the Space sat # at `hardware: None` against `requested_hardware: zero-a10g`, bound its port, # and served 503 on every request -- the ZeroGPU container was never # provisioned. HuggingFace's own instructions list importing `spaces` and # decorating a GPU function as the steps that make a ZeroGPU Space work. # # The resolution is to declare ONE minimal GPU entry point that the console # never calls on any user path. It satisfies provisioning; it consumes no quota # because it is never invoked. This is a platform accommodation, stated plainly # rather than dressed up as a computational need: nothing in this project uses # a GPU, and no model was changed to pretend otherwise. try: # pragma: no cover - the Hub runtime always provides this import spaces except ImportError: # local development, CI, the test suite spaces = None if spaces is not None: # pragma: no cover - only meaningful on ZeroGPU @spaces.GPU(duration=1) def _zerogpu_provisioning_probe() -> str: """Declared so ZeroGPU provisions the Space. Never called by the UI. Every user-facing path -- prediction, grading, the shift check -- deliberately avoids this function. See the note above on why decorating the real inference path exhausted a 5-minute daily quota in one page load. """ return "cpu-bound console; no GPU work is performed" ROOT = Path(__file__).resolve().parent.parent ASSETS = ROOT / "app" / "assets" MODELS = ROOT / "outputs" / "models" FIGURES = ROOT / "outputs" / "figures" #: A KS statistic above this against the qualification cohort puts the console #: into its alarm state. Phase 10 measured D = 0.651 for the shifted campaign #: and D near zero in-distribution. This threshold is DIRECTIONAL, not #: calibrated: one shifted campaign can demonstrate the association but cannot #: establish a decision threshold, and the interface says so when it fires. KS_ALARM_THRESHOLD = 0.30 ALPHA = 0.10 #: Labels for the incoming-lot selector. The shifted option is real batch-3 #: data, not a perturbation -- the alarm has to be demonstrable on the campaign #: that actually broke the guarantee, or the console is asserting a safety #: property it never exercises. QUALIFICATION_LOT = "Qualification campaigns (batches 1–2)" SHIFTED_LOT = "New production campaign (batch 3)" #: The one capacity at which the allocation policies were actually compared #: (held_fraction = 0.2017 ± 0.0034 over 50 folds, allocation_summary.csv). EVALUATED_HELD_PERCENT = 20.2 # --------------------------------------------------------------------------- # Bundle loading (cold start) # --------------------------------------------------------------------------- class Bundle: """Everything the console needs, loaded once at import.""" def __init__(self) -> None: self.manifest = json.loads((ASSETS / "manifest.json").read_text(encoding="utf-8")) self.results = json.loads((ASSETS / "results.json").read_text(encoding="utf-8")) self.demo = pd.read_parquet(ASSETS / "demo_cells.parquet") self.budgets = [int(b) for b in self.manifest["budgets"]] self.grades = self.manifest["grades"] self.warranty = int(self.manifest["warranty_target_cycles"]) self.cost_matrix = self.manifest["cost_matrix"] self.grade_order = list(self.grades) self.boundaries = {g: float(self.grades[g]["min_cycles"]) for g in self.grade_order} # Models are loaded LAZILY and cached. ZeroGPU Spaces spin down # aggressively, so cold start matters: the constructor touches only the # small metadata bundle (~0.2 MB) and the first prediction for a budget # pays for that one model. Loading all five eagerly would add ~6 MB of # joblib deserialisation to every cold start for models a given session # may never use. self._models: dict[int, Any] = {} self.features: dict[int, list[str]] = {} self.half_widths: dict[int, float] = {} for budget in self.budgets: entry = self.manifest["models"].get(str(budget)) if entry and (MODELS / f"triage_extra_trees__budget{budget:03d}.joblib").exists(): self.features[budget] = entry["feature_names"] self.half_widths[budget] = float(entry["conformal_half_width_90"]) # Campaign membership drives the shift check. Batches 1-2 are the # qualification cohort the model and its conformal quantiles were # calibrated on; batch 3 is the genuinely shifted production campaign # (Phase 3: KS D = 0.651, p = 1.7e-11). at_reference = self.demo[self.demo["budget"] == max(self.budgets)] self.qualification_ids = sorted( at_reference.loc[at_reference["batch"] != "batch3", "cell_id"] ) self.shifted_ids = sorted( at_reference.loc[at_reference["batch"] == "batch3", "cell_id"] ) self._reference_cache: dict[int, np.ndarray] = {} self.cell_ids = sorted(self.demo["cell_id"].unique()) self.build_time = self.manifest.get("provenance", {}).get("generated_at_utc", "unknown") def model(self, budget: int): """Load and cache one budget's model on first use.""" if budget not in self._models: import joblib self._models[budget] = joblib.load( MODELS / f"triage_extra_trees__budget{budget:03d}.joblib" ) return self._models[budget] @property def models(self) -> dict[int, Any]: """Budgets with a model available (not necessarily loaded yet).""" return {b: None for b in self.features} def row(self, cell_id: str, budget: int) -> pd.Series: subset = self.demo[(self.demo["cell_id"] == cell_id) & (self.demo["budget"] == budget)] return subset.iloc[0] def predict(self, cell_id: str, budget: int) -> tuple[float, float, float]: """Point prediction and conformal interval, in log10 cycle life.""" row = self.row(cell_id, budget) columns = self.features[budget] X = pd.DataFrame([[row.get(c, np.nan) for c in columns]], columns=columns) centre = float(_infer(self.model(budget), X)) half = self.half_widths[budget] return centre, centre - half, centre + half def grade_probabilities(self, lower: float, upper: float) -> dict[str, float]: samples = np.linspace(lower, upper, 512) cycles = 10.0 ** samples ordered = sorted(self.boundaries.items(), key=lambda kv: -kv[1]) out: dict[str, float] = {} previous = np.inf for grade, floor in ordered: out[grade] = float(((cycles >= floor) & (cycles < previous)).mean()) previous = floor total = sum(out.values()) or 1.0 return {g: v / total for g, v in out.items()} def action_costs(self, probabilities: dict[str, float]) -> dict[str, float]: out: dict[str, float] = {} for j, assigned in enumerate(self.grade_order): out[assigned] = float(sum( probabilities[true] * self.cost_matrix[f"true_{true}"][j] for true in self.grade_order )) return out def lot_ids(self, campaign: str) -> list[str]: """Cell IDs making up an incoming lot from the named campaign.""" return self.shifted_ids if campaign == SHIFTED_LOT else self.qualification_ids def campaign_of(self, cell_id: str) -> str: """Which campaign a cell actually came from. The lot is a PROPERTY OF THE CELL, never an operator choice. An earlier version offered the campaign as a separate control, which let the interface show a batch-1 cell under a batch-3 lot -- an incoherent state -- and, more seriously, made the shift check something an operator could decline to apply. A safety check that the person it protects can switch off is not a safety check. """ return SHIFTED_LOT if cell_id in self.shifted_ids else QUALIFICATION_LOT def cell_choices(self) -> list[tuple[str, str]]: """Dropdown entries labelled with the campaign each cell came from.""" return [ (f"{c} · {'batch 3 — new campaign' if c in self.shifted_ids else 'qualification'}", c) for c in self.cell_ids ] def reference_predictions(self, budget: int) -> np.ndarray: """Model predictions on the qualification cohort, cached per budget. The shift check compares PREDICTIONS on both sides, never true lives. A production console has no labels for an incoming lot, so a check that needed them would be undeployable -- and comparing true reference lives against predicted incoming ones would conflate model bias with distribution shift. Both sides go through the same model, so the only thing the statistic can move on is the input distribution. """ if budget not in self._reference_cache: self._reference_cache[budget] = np.array( [self.predict(c, budget)[0] for c in self.qualification_ids] ) return self._reference_cache[budget] def grade_for(self, cycles: float) -> str: for grade in self.grade_order: if cycles >= self.boundaries[grade]: return grade return self.grade_order[-1] def _infer(model, X: pd.DataFrame) -> float: """The heaviest inference path. Runs on CPU, deliberately undecorated. See the ZeroGPU note at the top of this module: decorating this with @spaces.GPU exhausted the daily quota on one page load and broke the deployed console, because the shift check calls it once per reference cell. """ return float(model.predict(X)[0]) BUNDLE = Bundle() # --------------------------------------------------------------------------- # Campaign-shift check # --------------------------------------------------------------------------- def campaign_shift_check(campaign: str, budget: int) -> tuple[bool, float, str]: """Two-sample KS of the incoming LOT against the qualification cohort. Returns (alarm, statistic, html). The alarm is a STATE, not a message: the result panel renders differently when it fires, because Phase 10 showed the failure mode is a model that becomes confidently wrong, and a footnote is not a proportionate response to that. THIS IS A LOT-LEVEL PROPERTY AND CANNOT BE OTHERWISE. A single cell carries no distribution to test, so an earlier version that passed one prediction to a two-sample test could only ever report "not run". Shift is a statement about the population a cell arrived in, and the operator screens a cell in the context of its lot. WHAT THIS CHECK CANNOT DETECT. A shift that leaves the predicted-life distribution unchanged while moving the feature-to-life mapping -- same marginal, different conditional. That is precisely the regime where the model is wrong and the check is silent, and no univariate two-sample test on the output can see it. """ lot = BUNDLE.lot_ids(campaign) if len(lot) < 3: return False, 0.0, _nominal_html(0.0, insufficient=True) # Identity is decided on cell IDs, not predicted values. Comparing the float # arrays looked equivalent and was not: tree-ensemble reductions differ at # ~1e-15 between calls, which is enough for ks_2samp to see two distinct # samples and report D = 1/n for a cohort compared against itself. if lot == BUNDLE.qualification_ids: return False, 0.0, _nominal_html(0.0, is_reference=True, n_lot=len(lot)) values = np.array([BUNDLE.predict(c, budget)[0] for c in lot]) reference = BUNDLE.reference_predictions(budget) result = stats.ks_2samp(reference, values) statistic = float(result.statistic) if statistic >= KS_ALARM_THRESHOLD: return True, statistic, _alarm_html(statistic, float(result.pvalue), len(lot)) return False, statistic, _nominal_html(statistic, n_lot=len(lot)) def _alarm_html(statistic: float, pvalue: float, n_lot: int) -> str: return f"""
This lot of {n_lot} cells does not look like the qualification cohort
(KS D = {statistic:.3f}, p = {pvalue:.2e}, alarm at
D ≥ {KS_ALARM_THRESHOLD}).
Conformal validity assumes calibration and production data are exchangeable. That assumption is not met here, so the escape-rate bound shown below is not guaranteed.
Measured on the one shifted campaign in this dataset: coverage fell from
90.7% to 42.5% while the prediction interval got
31.5% NARROWER and error nearly doubled. The model becomes
confidently wrong — its own confidence signal moves in the
reassuring direction exactly as it stops being trustworthy.
Required action: recalibrate the conformal quantiles on labelled cells from this campaign before relying on any decision here. Minimum calibration set: 19 cells for α=0.05, 99 for α=0.01.
The D ≥ {KS_ALARM_THRESHOLD} threshold is directional, not calibrated — one shifted campaign can demonstrate the association but cannot set a decision boundary.
Fewer than 3 cells in this lot; a two-sample test has nothing to compare.
This cell came from the qualification cohort, so the comparison is against its own campaign and passes by construction. That is shown to make the contrast legible, not as evidence the check works — select a cell marked batch 3 — new campaign to see it fire on real out-of-distribution data.
This lot of {n_lot} cells is consistent with the qualification cohort (KS D = {statistic:.3f}, alarm at D ≥ {KS_ALARM_THRESHOLD}). The conformal guarantee's exchangeability assumption is not contradicted.
The policy comparison below was measured at a held fraction of
20.2% ± 0.3 across 50 folds and
has not been recomputed at {slot_percent:.0f}%. Only the
slot count above responds to this control. Treat the ranking as evidence at
the evaluated capacity, not as a capacity sweep.
Cost-optimal, risk-controlled end-of-line screening for lithium-ion cells — grade assignment with a conformal bound on the escape rate.
model build {BUNDLE.build_time} · {len(BUNDLE.models)} budgets · recipe descriptors excluded per Phase 10 · inference only, CPU
CellTriage · inference-only console · no raw cycling data is shipped · demo subset of {len(BUNDLE.cell_ids)} cells
""") return demo def main() -> None: """Launch the console. Theme and CSS are passed to launch() rather than to Blocks: Gradio 6 moved them there, and `show_api` was removed in the same release. DELIBERATELY PLAIN. Successive attempts to out-guess the Spaces runner -- exposing a module-level `demo`, wrapping `launch` to re-inject styling, ceding the port, forcing `block_thread()` -- each fixed an imagined problem and left the Space unprovisioned at `hardware: None`. The structure below is the one that actually reached RUNNING on ZeroGPU with the theme intact. """ build_interface().launch( server_name="0.0.0.0", server_port=7860, theme=theme.build_theme(), css=theme.CSS, ) if __name__ == "__main__": main()