# Project export: Phagentic
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: The OS for autonomous biotech.
- Devpost: https://devpost.com/software/phagentic
- GitHub: https://github.com/nirpechuk/phagentic
- Video: https://www.youtube.com/embed/z2aTEHjq1O8?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 4 GitHub contributor(s) — Nir Pechuk (17 commits), Shira (17 commits), Claude Opus 4.8 (1M context) (6 commits), Melissa Jin (5 commits)
## Devpost submission (written by the team)
### Overview
By 2050, antibiotic resistance will have killed more people than cancer. Phagentic is the solution. Phagentic is a physical bioreactor with an operating system built to autonomously drive sensitive, hidden-state reactions. Its namesake - the bacteriophage - is biotech's most promising antibiotic alternative: a virus engineered to destroy otherwise-resistant bacterial strains. Because phages co-evolve with their targets, resistance never reoccurs But manufacturing is the bottleneck: it's a complex, multi-stage reaction that behaves differently every run. Phagentic uses a chemistry-informed model that continuously learns the reaction's kinetics while driving it. About Us We’re a team of three CS and one ECE student from Carnegie Mellon University. None of us had any previous biochem experience, but the ambitious nature of the challenge inspired us to give it a try. Learned wayyyy too much chemistry for this :) Trials and Tribulations The surface area of our project is massive: we were running a chemistry experiment, building the hardware for a bioreactor, writing the firmware, 3D printing part of the rig, creating the OS frontend/backend, and training the controller model all at once. We hit many challenges along the way, including: 3D printing challenges. Despite our print theoretically taking 2 hours, technical issues (and a quick visit from the fire marshal) led us to get our first working prototype at 2am. Along the way, we learned about tuning the slicer, print speed, and filament type to ensure that the print actually finished. Tuning the actual reaction. Before any controller could work, we had to make the reaction controllable. That meant real wet-chemistry iteration - adjusting the concentrations of glucose, NaOH, and dye to land the oscillation in a usable window: fast enough to demo, with a swing big enough for the sensor to read and slow enough for the controller to act on. Dozens of trial batches were spent dialing those ratios in. Moving from sim to real. We developed and tested the controllers in simulation against our ODE model first, where iteration is cheap — then moved them onto the live reactor, where they hit everything a sim doesn't have: sensor noise, time delay, reagent drift, BLE dropouts. Closing that sim-to-real gap was most of the real work. We also had some fun hack-y moments: Frying our first ESP32 when we were switching pins while it was plugged in (oops!) Almost melting the table because we forgot that NaOH releases heat when it dissolves Hand-whittling a stick with a pocketknife to hold the RGB corrector Doordashing duct tape when we needed waterproof adhesive at midnight Tech Stack The Chemistry The reactor runs the Blue Bottle reaction: a dye that flips between blue (oxidized, driven by stirring in oxygen) and clear (reduced by the glucose in solution). It oscillates blue → clear → blue and stalls once the glucose or base substrate runs out; a system that has to be fed and kept in rhythm. That's why it stands in for bacteriophage dynamics: a phage population fighting bacteria rises and crashes in the same predator-prey rhythm, and dosing it is the same closed-loop control problem. Blue intensity stands in for population, so the controller that holds the color on target is the one that would hold a phage treatment on target. A phage ⇄ blue converter in the UI makes the mapping explicit, as on a log scale, clear reads as zero phages and the deepest blue as a saturated culture of ~10¹². Hardware The hardware consists of an ESP32 chip connected to two pumps and the mixer through MOSFETs to control voltage. The RGB sensor plugs into the chip as well. Commands are streamed over a two-way Bluetooth Low Energy connection with the client laptop. Control Model The core of the project is a chemistry-informed ML model we trained to control the reaction. The model has four layers: State estimator. We clean the raw 20hz sensor data time series into meaningful signals about the true color, current pump states, and phage of the reaction. Gray-box ODE. Starting from a system of differential equations that model our target reaction, we fine-tune constants and dynamics on noised simulated and real-world run data, producing an accurate prior through which we can predict the future state of the reaction. Model-predictive control. The model searches through the action space and predicts the future using the learned ODEs for each action. It chooses the action that brings us closest to desired state. Continual learning. Each reaction is different, so as the reaction runs, we continue tuning the gray-box ODE predictors. The model is safety-gated by a classic PID loop; it isn’t allowed to take actions too drastic to prevent catastrophic failures. 3D Print We modeled and, using the provided 3d printers, printed a mount that caps onto the beaker for fixing the pumps' tubes, as well as a clip which slots into the mount for attaching the RGB sensor, and also provides a screen behind the fluid to diminish glare. This took a lot of troubleshooting and failed attempts with the different printers, but with some perseverance, we printed two iterations of our design, which, on the second try, met the tolerances and needs of our setup. Ethical Considerations Biochemical safety is critical, and we only trust scientists to handle critical decisions. Our core product ideology at Phagentic is that we're a tool to achieve what was earlier impossible in sensitive bio-manufacturing reactions, not a tool to replace scientists. As a result: The controller uses data from previous runs and a naive PID solver to gate possible actions to those deemed "safe." As a result, the model will never be allowed to drastically swing the reaction and cause a potential leak. The "Ask Phage" copilot is structurally read-only; it's designed as an analysis tool, and the safety-gated controller model is the only way to autonomously control the reaction. We incorporated easy fail-safes to quickly stop the reaction, and a manual mode for the scientist to take over driving if needed.
## README (from the GitHub repository)
# PHAGENTIC
A closed-loop controller for the **Blue Bottle** oscillating reaction. PHAGENTIC
watches the live colour of the solution (blue ⇄ colorless), estimates the
oscillation state — amplitude, period, phase, stall risk — and drives the
stirrer and glucose pump to hold the rhythm.
The **brains run headless in Python.** A backend process owns the device link,
the oscillation analysis, the control loop, and a **pluggable ML model** that
drives the reaction. The web UI is a thin client: it renders the state the
backend streams and sends commands back, over a single WebSocket. This means the
control loop keeps running with no browser open, and you can drop in a real
model (sklearn / torch / an RL policy) without touching the rest of the system.
```
phagentic/
├── backend/ # headless control backend ← the live system
│ ├── app.py # entrypoint: wires everything, serves uvicorn (python -m backend.app)
│ ├── server.py # FastAPI: /ws (state out + commands in) + GET /config
│ ├── hardware/ # device.py (DeviceWorker loop), roles.py, calibration.py, _hublink.py
│ ├── analysis/ # detector.py (oscillation extrema), signal.py
│ ├── control/ # model.py (Model interface), pi_model.py, arbiter.py, registry.py
│ ├── state/ # store.py (shared snapshot), commands.py, events.py
│ ├── protocol/ # messages.py (WebSocket message vocabulary)
│ ├── tools/ws_probe.py # headless WebSocket probe (verify without a browser)
│ └── tests/ # unit tests (no hardware needed)
├── frontend/ # the web UI — a thin WebSocket client
│ ├── index.html # built console (build.js assembles it from src/)
│ ├── api.js # WebSocket bridge to the backend
│ ├── logic.js # UI component (rendering + command sending; no analysis)
│ ├── runtime.js # vendored React-based template renderer
│ └── src/ # shell.html + widgets/*.html (built by build.js)
├── hub/ # DEPRECATED. Its device layer (controller/transport/config) is
│ # reused by backend/; dashboard.py + main.py are legacy tools.
├── controller/ # ESP32 firmware (generic pin API; pins configured at runtime)
├── experiment.md # the Blue Bottle experiment
└── Makefile # make backend / ui / setup / test / upload
```
## Architecture
## WebSocket protocol (`ws://:8080/ws`)
Every frame is JSON `{"type": ...}`.
| Direction | Messages |
|---|---|
| server → client | `state` (full snapshot + `narr_new[]`), `config` (layout/roles/models), `ack`, `calibration` |
| client → server | `set_actuator {role,value}`, `pulse_actuator {role,ms}`, `set_mode {mode}`, `set_model {name}`, `set_model_params {params}`, `recalibrate`, `reload_config`, `reset_run`, `ping` |
`role` ∈ `stirrer` · `light` (PWM 0–255) and `glucose` · `naoh` (digital pumps).
Roles resolve to physical pins from `hub/config.json` by name match, so pins can
move in config without code changes.
### What the UI sees and controls
- **Sees:** live solution colour (RGB swatch + lux), normalized blue intensity,
oscillation waveform, amplitude, period/half-period, phase, cycle count, stall risk.
- **Controls:** Stirrer (PWM), Glucose pump (auto trigger + manual pulse, dose
ms), NaOH pump (manual pulse), Sensor light (brightness), manual/auto/ml mode,
model params, sensor recalibration, live config reload.
## Run it locally
```bash
make setup # one-time: venv + deps (hub/.venv) for backend + hub
make backend # headless backend on http://localhost:8080 (ws://localhost:8080/ws)
make ui # web UI on http://localhost:5173 (UI_PORT=8000 to override)
```
Then open **http://localhost:5173/**. The UI connects to the backend over the
WebSocket and shows the hardware status in the header (**`⬡ HARDWARE`** when the
ESP32 is connected, **`◌ NO DEVICE`** when the backend is up but the device
isn't, **`◌ OFFLINE`** when the backend is unreachable — it auto-reconnects).
> First load needs internet (the renderer pulls React from a CDN).
Hardware: power on the bioreactor (ESP32 flashed with `controller/`, advertising
as `Bioreactor`). The backend scans for it on start and re-pushes the pin map +
re-asserts outputs on every reconnect.
### Verify without a browser
```bash
make test # unit tests (detector, PI model, arbiter)
python -m backend.tools.ws_probe # observe the live state stream
python -m backend.tools.ws_probe --mode auto
python -m backend.tools.ws_probe --set stirrer 200
```
### URL params (frontend)
- `?backend=ws://host:8080/ws` (or `http://host:8080`) — point the UI at a
specific backend. Defaults to `ws://:8080/ws`.
- `?view=console` — skip the landing page and open the console directly.
## Legacy: the `hub/` dashboard
`hub/` is deprecated as a UI but its device layer (`controller.py`, `transport/`,
`config.py`) is the reused, single source of truth for the wire protocol — the
backend imports it directly. The old wired tools still run if you need them:
```bash
make dashboard # legacy Flask dashboard (hub/dashboard.py, :8080)
make run # legacy terminal RGB stream (hub/main.py)
```
## Configuration
`hub/config.json` is the single source of truth for wiring: MOSFETs
(`name`, `pin`, `mode` = `pwm`/`digital`) plus optional `sensor_light`. Edit it
and either restart the backend or send `reload_config` from the UI. Set
`BLE_DEVICE` to override the device name, `BIOREACTOR_CONFIG` to point at a
different config file, and `PORT` to change the backend port.
## Detected evidence (automated analysis)
Indexed codebase: 86 recognized source files, 871 KB.
- Anthropic (technology) — detected in the code
- FastAPI (technology) — detected in the code
- Flask (technology) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- C++ (language) — claimed on Devpost, not found in the code
- React (technology) — claimed on Devpost, not found in the code
- AI coding agent: Claude Code — evidence: config files committed to the repository; commit authorship or trailers
## Codebase structure (from repository index)
### Files (96 of 96)
```
.gitignore
architecture.html
backend/__init__.py
backend/analysis/__init__.py
backend/analysis/detector.py
backend/analysis/signal.py
backend/app.py
backend/control/__init__.py
backend/control/amplitude_controller.py
backend/control/arbiter.py
backend/control/goal_model.py
backend/control/heuristic_controller.py
backend/control/model.py
backend/control/mpc_controller.py
backend/control/pi_model.py
backend/control/registry.py
backend/estimator/__init__.py
backend/estimator/state_estimator.py
backend/GOAL_CONTROLLER.md
backend/hardware/__init__.py
backend/hardware/_hublink.py
backend/hardware/calibration.py
backend/hardware/device.py
backend/hardware/roles.py
backend/protocol/__init__.py
backend/protocol/messages.py
backend/requirements.txt
backend/server.py
backend/sim/__init__.py
backend/sim/bluebottle_ode.py
backend/sim/fit.py
backend/sim/fitted_params.json
backend/sim/rollout.py
backend/state/__init__.py
backend/state/commands.py
backend/state/events.py
backend/state/store.py
backend/tests/__init__.py
backend/tests/test_amplitude_controller.py
backend/tests/test_arbiter.py
backend/tests/test_bluebottle_ode.py
backend/tests/test_detector.py
backend/tests/test_goal_model.py
backend/tests/test_pi_model.py
backend/tests/test_state_estimator.py
backend/tools/__init__.py
backend/tools/log_run.py
backend/tools/replay_eval.py
backend/tools/ws_probe.py
claude.md
controller/controller.ino
experiment.md
frontend/api.js
frontend/ble.js
frontend/build.js
frontend/index.html
frontend/logic.js
frontend/runtime.js
frontend/src/shell.html
frontend/src/widgets/00-frame-top.html
frontend/src/widgets/01-watercolor-desk.html
frontend/src/widgets/02-console.html
frontend/src/widgets/03-actuator-status.html
frontend/src/widgets/04-colour-log.html
frontend/src/widgets/05-world-model.html
frontend/src/widgets/06-narration.html
frontend/src/widgets/07-manual-console.html
frontend/src/widgets/08-calculator.html
frontend/src/widgets/09-ask-phage.html
frontend/src/widgets/10-runs.html
frontend/src/widgets/10b-temp-notes.html
frontend/src/widgets/11-dock-rail-closed-tools.html
frontend/src/widgets/12-settings-bluetooth.html
frontend/src/widgets/13-landing.html
frontend/src/widgets/14-floating-phages-drift-in-l.html
hub/chat_server.py
hub/config.json
hub/config.py
hub/controller.py
hub/dashboard.py
hub/main.py
hub/requirements.txt
hub/transport/__init__.py
hub/transport/base.py
hub/transport/ble_transport.py
hub/transport/serial_transport.py
LICENSE
Makefile
pitch/ARCHITECTURE.md
pitch/BANNER_PROMPT.md
pitch/PHAGENTIC Architecture.html
pitch/PITCH.md
README.md
run.sh
runs/manual-heuristic-20260620-234407.jsonl
runs/manual-heuristic-20260621-012639.jsonl
```
### Dependencies
- backend/requirements.txt: fastapi@>=0.110, uvicorn[standard]@>=0.27
- hub/requirements.txt: anthropic@>=0.40, bleak@>=0.21, flask@>=3.0, pyserial@>=3.5
### Recent commits (newest first)
- fix graph
- Update README with new architecture image and WebSocket info
- Update README.md
- Banner
- add architecture diagram
- add architecture diagram
- model change
- architect html
- Add files via upload
- more training data
- training data
- merge origin/main: reconcile heuristic/mpc toggle with new amplitude_lock AUTO (3-way controller toggle)
- auto console: heuristic/mpc controller toggle, pitch
- backend
- ordering widgets
- widget ordering
- backend adding
- small frontend tweak
- frontend: phage⇄blue converter, target-phage in AUTO, console-view layout, centered tidy, fixed console size, top-right soft-close
- word and shine
## Key source files (fetched from GitHub, selected and truncated for size)
### experiment.md
```markdown
# Blue Bottle Oscillating Reaction — Experiment Description
## Chemistry
The **Blue Bottle** reaction is a classic oscillating redox demonstration.
**Reagents (initial charge to the vessel):**
| Reagent | Role |
|---|---|
| Distilled water | Solvent |
| Dextrose (D-glucose) | Reducing agent; fuel for oscillations |
| NaOH | Alkaline medium required for glucose to reduce methylene blue |
| Methylene Blue (trace) | Redox indicator; blue when oxidized, colorless (leuco form) when reduced |
**Mechanism:**
1. In alkaline solution, glucose slowly reduces methylene blue to its colorless leuco form.
2. Stirring introduces dissolved O₂, which re-oxidizes leuco-MB back to blue.
3. Net result: the solution cycles blue → colorless → blue as the competing redox reactions proceed.
4. Oscillations dampen as glucose is consumed; adding more glucose (via the Glucose Pump) restarts them.
**Reagents added by the model during the run:**
- **Glucose Pump** — pulses a concentrated dextrose solution to replenish reducing power when oscillations dampen.
- **Chloride Pump** — reserved; can deliver a NaOH top-up to maintain alkalinity if pH drift damps oscillations early.
---
## Hardware Used
| Device | Pin | Mode | Role |
|---|---|---|---|
| Glucose Pump | GPIO 19 | digital (on/off) | Injects dextrose bolus |
| Chloride Pump | GPIO 18 | digital (on/off) | NaOH top-up (if needed) |
| Stirrer | GPIO 23 | PWM 0–255 | Controls O₂ introduction rate |
| Sensor Light | GPIO 25 | PWM 0–255 | Illuminates solution for TCS34725 |
| TCS34725 | I²C (SDA 21, SCL 22) | sensor | Measures RGBC at 20 Hz |
---
## What the Model Controls
The model is a feedback controller that reads the RGB sensor and actuates the stirrer and glucose pump to sustain oscillations at a target period.
**Observation:** raw RGBC from TCS34725, IR-corrected and white-balance-normalized → scalar blue channel intensity B̂ ∈ [0, 1].
**Control outputs:**
1. **Stirrer PWM** (continuous, 0–255) — primary knob. High duty cycle introduces O₂ quickly (drives blue phase); low duty cycle lets glucose reduction dominate (drives colorless phase). The model modulates this to shape oscillation period and amplitude.
2. **Glucose Pump pulse** (binary, triggered event) — fires a fixed-duration on pulse (~0.5 s) when oscillation amplitude has decayed below a threshold for two consecutive cycles. Restores reducing power without flooding the vessel.
**Control objective:** maintain peak-to-trough blue amplitude > 0.4 (normalized) for the full run duration, with an oscillation half-period in the 15–45 s range.
**Model type:** rule-seeded PID on stirrer PWM, with an event trigger for the glucose pump. The PID setpoint tracks the midpoint of the last observed swing; the event trigger fires if peak amplitude < 0.4 over a 2-cycle window. This gives interpretable behavior for a live demonstration.
---
## Time Horizon
| Phase | Duration | Description |
|---|---|---|
| Setup & calibration | ~2 min | Push `configure`, warm up
[truncated — 1079 more characters]
```
### claude.md
```markdown
# Bioreactor
Closed-loop controller for the Blue Bottle oscillating reaction. The **brains run headless in Python** (`backend/`): a backend process owns the ESP32 link, the oscillation analysis, the control loop, and a pluggable ML model, and serves a WebSocket API. The web UI (`frontend/`) is a thin client — it renders streamed state and sends commands. ESP32 firmware is `controller/`. The `hub/` is deprecated as a UI but its device layer is reused by the backend.
## Structure
### `backend/` — the live headless system (run: `python -m backend.app`, port 8080)
- `backend/app.py` — entrypoint. Loads config, builds the pieces, starts the DeviceWorker thread, serves uvicorn. Registers a shutdown hook that drives actuators safe.
- `backend/server.py` — FastAPI app. `/ws` WebSocket (broadcasts full state ~15 Hz, accepts commands) + `GET /config` (hardware layout + roles + model list). Inbound handlers validate then enqueue/set-event; they never touch the device.
- `backend/hardware/device.py` — `DeviceWorker`: the **only** thread that touches the Controller. 20 Hz loop: handle reload/recalibrate events → drain commands → `get_rgb` → analyse → arbiter decides → apply (diffed) → publish to StateStore. Supervises BLE reconnect (re-pushes pin map + re-asserts outputs) and zeroes actuators on shutdown.
- `backend/hardware/roles.py` — `RoleMap`: resolves logical roles (`stirrer`/`glucose`/`naoh`/`light`) to pins by config name match (ported from the old browser logic).
- `backend/hardware/calibration.py` — `sample_wb` + `to_rgb8` colour math (moved from `hub/dashboard.py`).
- `backend/hardware/_hublink.py` — adds `hub/` to `sys.path` and re-exports `Controller`, `BLETransport`, `config` (zero edits to `hub/`).
- `backend/analysis/detector.py` — `Detector`: extrema detection on the blue signal → amplitude, half/period, cycle count, phase (ported from the browser).
- `backend/control/model.py` — the pluggable `Model` interface: `observe(ReactionState) -> Action`, plus `get_params`/`set_params`/`reset`.
- `backend/control/pi_model.py` — `PIModel` baseline (PI on stirrer + auto glucose pulse). `arbiter.py` — mode machine (manual/auto/ml) + pump-pulse timer + Action→actuator merge; contains model exceptions. `registry.py` — name→Model factory for `ml` mode.
- `backend/state/` — `store.py` (thread-safe shared snapshot), `commands.py` (queue), `events.py` (recalibrate/reload/shutdown event pairs).
- `backend/protocol/messages.py` — WebSocket message-type constants for both directions.
- `backend/tools/ws_probe.py` — headless WebSocket probe; `backend/tests/` — unit tests (no hardware).
### `frontend/` — thin WebSocket client
- `frontend/api.js` — `PhagenticBackend`: WebSocket bridge (`ws://host:8080/ws`). Routes `state`/`config`/`calibration` to callbacks; exposes `setActuator`/`pulseActuator`/`setMode`/`setModelParams`/`reloadConfig`/`resetRun`/`recalibrate`. Auto-reconnects.
- `frontend/logic.js` — UI component. `applyState(msg)` mirrors backend state into
[truncated — 4308 more characters]
```
### hub/requirements.txt
```
pyserial>=3.5
flask>=3.0
bleak>=0.21
anthropic>=0.40
```
### backend/requirements.txt
```
# Backend extends the hub device layer (controller/transport/config), so the
# hub's own deps are still required (bleak for BLE; pyserial pulled in by the
# transport package). Install both: pip install -r hub/requirements.txt -r backend/requirements.txt
fastapi>=0.110
uvicorn[standard]>=0.27
```
### backend/app.py
```python
#!/usr/bin/env python3
"""Headless bioreactor backend entrypoint.
Loads config, wires the pieces, starts the DeviceWorker thread (owns the BLE
link + control loop), and serves the WebSocket/REST API with uvicorn. Run from
the repo root: ``python -m backend.app`` (override port with PORT env var).
"""
import logging
import os
import threading
import uvicorn
from backend.control.arbiter import ControlArbiter
from backend.analysis.detector import Detector
from backend.hardware._hublink import config
from backend.hardware.device import DeviceWorker
from backend.hardware.roles import RoleMap
from backend.server import create_app
from backend.state.commands import CommandQueue
from backend.state.events import DeviceEvents
from backend.state.store import StateStore
LOOP_HZ = int(os.environ.get("LOOP_HZ", "20"))
HTTP_PORT = int(os.environ.get("PORT", "8080"))
def build():
cfg = config.load_config()
store = StateStore()
commands = CommandQueue()
events = DeviceEvents()
roles = RoleMap(cfg)
arbiter = ControlArbiter()
detector = Detector()
worker = DeviceWorker(cfg, store, commands, events, arbiter, detector, roles, loop_hz=LOOP_HZ)
thread = threading.Thread(target=worker.run, name="device", daemon=True)
thread.start()
app = create_app(worker, store, commands, events)
return app
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
app = build()
log = logging.getLogger("backend")
log.info("Bioreactor backend → http://localhost:%d (ws://localhost:%d/ws)", HTTP_PORT, HTTP_PORT)
# Use the standard asyncio loop, not uvloop. uvicorn[standard] defaults to
# uvloop; on macOS a uvloop loop on the main thread measurably slows bleak's
# CoreBluetooth BLE connect (~5x in testing) since its callbacks land on the
# transport's own loop thread. Our async workload is just a 15 Hz broadcast,
# so we don't need uvloop's throughput — asyncio keeps BLE snappy.
uvicorn.run(app, host="0.0.0.0", port=HTTP_PORT, log_level="warning", loop="asyncio")
if __name__ == "__main__":
main()
```
### hub/main.py
```python
#!/usr/bin/env python3
"""Streams RGB from a TCS34725 sensor and prints a live color swatch to the terminal."""
import sys
import time
import config
from controller import Controller
from transport.ble_transport import BLETransport
SAMPLE_RATE = 20 # Hz
def calibrate(ctrl: Controller, samples: int = 30) -> tuple[float, float, float, float]:
"""Point sensor at white; returns (sr, sg, sb, white_c) where white_c is the clear
channel value for white — used as the 100% brightness reference."""
print("Point sensor at a white surface, then press Enter to calibrate white balance...")
input()
rs, gs, bs, cs = [], [], [], []
for _ in range(samples):
d = ctrl.get_rgb()
if d:
rs.append(d["r"]); gs.append(d["g"]); bs.append(d["b"]); cs.append(d["c"])
time.sleep(1.0 / SAMPLE_RATE)
if not rs:
print("No sensor data — skipping calibration.")
return 1.0, 1.0, 1.0, 1.0
r_avg = sum(rs) / len(rs)
g_avg = sum(gs) / len(gs)
b_avg = sum(bs) / len(bs)
white_c = sum(cs) / len(cs)
peak = max(r_avg, g_avg, b_avg)
sr, sg, sb = peak / r_avg, peak / g_avg, peak / b_avg
print(f" R×{sr:.2f} G×{sg:.2f} B×{sb:.2f} white_c={white_c:.0f}\n")
return sr, sg, sb, white_c
def to_rgb8(
r: int, g: int, b: int, c: int,
wb: tuple[float, float, float], white_c: float,
) -> tuple[int, int, int]:
ir = max(0, (r + g + b - c) // 2)
rf = max(0.0, (r - ir) * wb[0])
gf = max(0.0, (g - ir) * wb[1])
bf = max(0.0, (b - ir) * wb[2])
peak = max(rf, gf, bf)
if peak == 0:
return 0, 0, 0
brightness = min(1.0, c / white_c) # 1.0 = white-level light, 0.0 = dark
s = 255.0 / peak * brightness
return int(rf * s), int(gf * s), int(bf * s)
def main() -> None:
cfg = config.load_config()
transport = BLETransport(config.device_name(cfg))
with Controller(transport) as ctrl:
print("Pinging...", end=" ", flush=True)
if not ctrl.ping():
print("no response. Is the firmware flashed?")
sys.exit(1)
print("OK")
pins = config.pin_map(cfg)
print(f"Configuring {len(pins)} pin(s)... ", end="", flush=True)
print(ctrl.configure(pins).get("status", "no response"), "\n")
light = cfg.get("sensor_light")
if light: # light on so the sensor sees lit conditions
ctrl.set_pwm(light["pin"], 255)
*wb, white_c = calibrate(ctrl)
print("Streaming RGB from TCS34725. Ctrl+C to stop.\n")
step = 1.0 / SAMPLE_RATE
try:
while True:
t0 = time.monotonic()
data = ctrl.get_rgb()
if data is None:
print("\rSensor error — check wiring or reflash. ", end="", flush=True)
else:
r, g, b = to_rgb8(data["r"], data["g"], data["b"], data["c"], tuple(wb), white_c)
swatch = f"\033[48;2;{r};{g};{b}m \033[0m"
print(f"\rR:{r:3d} G:{g:3d} B:{b:3d} lux:{data['c']:5d} {swatch}", end="", flush=True)
elapsed = time.monotonic() - t0
time.sleep(max(0.0, step - elapsed))
except KeyboardInterrupt:
print("\nStopped.")
if __name__ == "__main__":
main()
```
### backend/server.py
```python
"""FastAPI app: a state-broadcasting WebSocket + one REST route for the layout.
The asyncio side never blocks on the device: inbound frames are validated and
pushed onto the command queue / event flags, and a single broadcast task fans
the latest StateStore snapshot out to all connected clients. The few blocking
operations (recalibrate, reload) set an event and await its 'done' flag in a
thread-pool so the event loop stays responsive.
"""
import asyncio
import contextlib
import json
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from backend.protocol import messages as M
from backend.state.commands import (
CommandQueue, PULSE_ACTUATOR, RESET_RUN, SET_ACTUATOR, SET_MODE,
SET_MODEL, SET_MODEL_PARAMS,
)
from backend.state.events import DeviceEvents
from backend.state.store import StateStore
log = logging.getLogger("backend.server")
BROADCAST_HZ = 15
def create_app(worker, store: StateStore, commands: CommandQueue, events: DeviceEvents) -> FastAPI:
clients: set[WebSocket] = set()
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(_broadcast_loop(store, clients))
try:
yield
finally:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
# Drive actuators safe on the device thread, then let it exit.
events.shutdown_req.set()
await asyncio.get_event_loop().run_in_executor(None, events.shutdown_done.wait, 5.0)
app = FastAPI(lifespan=lifespan)
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)
@app.get("/config")
def config_route():
return JSONResponse(worker.layout())
@app.get("/")
def index():
return JSONResponse({"service": "bioreactor-backend", "ws": "/ws", "config": "/config"})
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await ws.accept()
clients.add(ws)
with contextlib.suppress(Exception):
await ws.send_text(json.dumps({"type": M.S_CONFIG, **worker.layout()}))
await ws.send_text(json.dumps({"type": M.S_STATE, **store.snapshot(), "narr_new": []}))
try:
while True:
raw = await ws.receive_text()
await _handle_inbound(ws, raw, worker, commands, events)
except WebSocketDisconnect:
pass
except Exception:
log.warning("ws error", exc_info=True)
finally:
clients.discard(ws)
return app
async def _broadcast_loop(store: StateStore, clients: set[WebSocket]) -> None:
interval = 1.0 / BROADCAST_HZ
while True:
await asyncio.sleep(interval)
if not clients:
store.drain_narr() # don't let narration pile up with nobody listening
continue
snap = store.snapshot()
narr = store.drain_narr()
data = json.dumps({"type": M.S_STATE, **snap, "narr_new": narr})
for ws in list(clients):
try:
await ws.send_text(data)
except Exception:
clients.discard(ws)
async def _ack(ws: WebSocket, ref, ok: bool, msg: str = "") -> None:
with contextlib.suppress(Exception):
await ws.send_text(json.dumps({"type": M.S_ACK, "ref": ref, "ok": ok, "msg": msg}))
async def _handle_inbound(ws, raw, worker, commands: CommandQueue, events: DeviceEvents) -> None:
try:
msg = json.loads(raw)
except json.JSONDecodeError:
return await _ack(ws, None, False, "bad json")
typ = msg.get("type")
ref = msg.get("ref")
if typ == M.C_SET_ACTUATOR:
if msg.get("role") not in M.VALID_ROLES:
return await _ack(ws, ref, False, "bad role")
commands.put(SET_ACTUATOR, {"role": msg["role"], "value": msg.get("value")})
return await _ack(ws, ref, True)
if typ == M.C_PULSE_ACTUATOR:
if msg.get("role") not in M.VALID_ROLES:
return await _ack(ws, ref, False, "bad role")
commands.put(PULSE_ACTUATOR, {"role": msg["role"], "ms": int(msg.get("ms", 500))})
return await _ack(ws, ref, True)
if typ == M.C_SET_MODE:
if msg.get("mode") not in M.VALID_MODES:
return await _ack(ws, ref, False, "bad mode")
commands.put(SET_MODE, {"mode": msg["mode"]})
return await _ack(ws, ref, True)
if typ == M.C_SET_MODEL:
commands.put(SET_MODEL, {"name": msg.get("name")})
return await _ack(ws, ref, True)
if typ == M.C_SET_MODEL_PARAMS:
commands.put(SET_MODEL_PARAMS, {"params": msg.get("params", {})})
return await _ack(ws, ref, True)
if typ == M.C_RESET_RUN:
commands.put(RESET_RUN, {})
return await _ack(ws, ref, True)
if typ == M.C_RECALIBRATE:
events.recalib_done.clear()
events.recalib_req.set()
ok = await asyncio.get_event_loop().run_in_executor(None, events.recalib_done.wait, 15.0)
with contextlib.suppress(Exception):
await ws.send_text(json.dumps({
"type": M.S_CALIBRATION,
"status": "ok" if ok else "timeout",
"wb": list(events.wb), "white_c": events.white_c,
}))
return await _ack(ws, ref, ok, "" if ok else "timeout")
if typ == M.C_RELOAD_CONFIG:
events.reload_done.clear()
events.reload_req.set()
ok = await asyncio.get_event_loop().run_in_executor(None, events.reload_done.wait, 5.0)
with contextlib.suppress(Exception):
await ws.send_text(json.dumps({"type": M.S_CONFIG, **worker.layout()}))
return await _ack(ws, ref, ok, "" if ok else "timeout")
if typ == M.C_PING:
return await _ack(ws, ref, True, "pong")
retu
[truncated — 55 more characters]
```
### run.sh
```shell
#!/usr/bin/env bash
# PHAGENTIC — run the web UI locally (with Web Bluetooth support).
#
# ./run.sh # serve on http://localhost:5173 and open the browser
# ./run.sh 8000 # use a different port
#
# Web Bluetooth needs a secure context, which "localhost" satisfies — so the
# ⌁ connect button works straight from here, no hub required. Use Chrome or Edge
# (Firefox/Safari don't support Web Bluetooth). First load needs internet (the
# renderer pulls React from a CDN).
set -euo pipefail
PORT="${1:-5173}"
ROOT="$(cd "$(dirname "$0")" && pwd)"
DIR="$ROOT/frontend"
URL="http://localhost:$PORT/"
if ! command -v python3 >/dev/null 2>&1; then
echo "python3 is required to serve the UI." >&2
exit 1
fi
if [ ! -f "$DIR/index.html" ]; then
echo "frontend/index.html not found at $DIR" >&2
exit 1
fi
echo "PHAGENTIC UI → $URL"
echo "Open in Chrome or Edge, then click ⌁ connect to pair the Bioreactor over Bluetooth."
echo "(No hardware? It runs a built-in simulation. Ctrl+C to stop.)"
# Assemble index.html from src/shell.html + src/widgets/*.html.
( cd "$DIR" && node build.js ) || echo "build.js skipped (node missing?) — serving existing index.html"
# Open the browser a moment after the server comes up (best-effort, non-fatal).
( sleep 1; (xdg-open "$URL" >/dev/null 2>&1 || open "$URL" >/dev/null 2>&1 || true) ) &
cd "$DIR"
exec python3 -m http.server "$PORT"
```
### architecture.html
```html
PHAGENTIC — System Architecture
Phagentic — System Architecture
closed-loop controller for the Blue Bottle oscillating reaction · hardware → backend → web console + agent