diff --git a/docs/facsmelody-re.md b/docs/facsmelody-re.md new file mode 100644 index 00000000000..447353a7323 --- /dev/null +++ b/docs/facsmelody-re.md @@ -0,0 +1,140 @@ +--- +orphan: true +--- + +# Reverse-engineering the BD FACSMelody + +## Why this is worth doing + +Sorting is the one step in plate-based single-cell sequencing that usually has to +happen off-automation. A liquid handler can build libraries end to end, but the +sort itself, one gated cell per well or a targeted, enriched population, is +typically a manual hand-off. Bring the sorter under PyLabRobot and that gap closes: +the deck stages the plate, calls `sort_to_plate`, and finishes the prep. Walkaway +library prep becomes possible. + +The same control also unlocks sorting for a target population. Enrich a cell type, +a marker-positive subset, or live singlets before the assay, and every downstream +measurement inherits that resolution. For cell-type-specific epigenomics this is +the whole point: you want to read regulatory state in the exact cell type a +variant acts in, not in a bulk average that blurs it away. + +The BD FACSMelody is a closed instrument driven by BD FACSChorus, so getting there +means reverse-engineering its control link. This document is the playbook. + +## The method (credit where it is due) + +This follows the methodology **Rick Wierenga** used to build PyLabRobot's device +backends: do not invent a protocol, work down to the OEM's own command layer, +sniff the OEM-to-device traffic, correlate each UI action to its bytes, decode the +framing, and replay over PyUSB or pyserial before wrapping it in a backend. The +same approach produced PLR's Hamilton STAR and Tecan EVO backends from captured +traffic and firmware command strings. + +The steps below are numbered so the path is reproducible and so a reviewer can see +exactly where the safety checks sit: the backend refuses to drive hardware until +every required command is decoded, and it refuses to actuate without a human in +the loop. + +The reverse-engineering toolkit that produces the deliverable lives in a separate +repository ([plr-clarity](https://github.com/di-omics/plr-clarity)) and is not +vendored into PyLabRobot. Only the consumer side ships here: the backend loads the +toolkit's output, a `ProtocolMap`, and turns it into `CellSorter` operations. + +## The deliverable: a ProtocolMap + +A `ProtocolMap` is a JSON file mapping each logical command (`start_sort`, +`clean`, and so on) to the exact bytes that drive it, plus how each frame is framed +and checksummed. The minimum set a sort-to-plate run needs is fixed: + +`connect`, `get_status`, `load_template`, `set_deposition`, `prime`, `start_sort`, +`wait_complete`, `abort`, `clean`. + +Combinatorial indexing tolerates tens to a hundred cells per well and does not need +index-sorting (recording which cell landed where). So the target is small: trigger +a saved gate template, configure count-controlled deposition, and poll status. We +mostly reverse-engineer how to trigger a saved template and move plates, not BD's +gating math. + +## Step 1: map the OEM stack and transport + +FACSChorus talks to the Melody over USB and/or an Ethernet cart link. Enumerate +everything, unplug and replug the instrument, and diff the enumeration to isolate +its link. Record the endpoint (for example `usb:0x1fbd:0x0002`, `COM4`, or +`10.0.0.5:9100`) and which transport it uses. + +The highest-leverage move is often not the wire but FACSChorus itself: its logs, +a local control daemon, and the experiment database. If Chorus logs the command +bytes or exposes a localhost service, you may not need to sniff USB at all. + +## Step 2: capture traffic while performing labeled UI actions + +Start your platform sniffer on the Melody's interface, then drive Chorus by hand. + +- Windows: Wireshark with USBPcap on the USB interface, exported to `.pcapng`. +- Linux: `usbmon` or Wireshark. +- Serial: a COM sniffer saved as ` ` lines. + +The core trick: perform one discrete Chorus action, mark the instant, and look +only at the bytes inside that window. Cover every required command (click "Start +Sort" and mark `start_sort`, click "Abort" and mark `abort`, and so on). + +## Step 3: correlate action to bytes, decode the framing + +For each labeled window, isolate the frames unique to that action by diffing across +windows to cancel the periodic keep-alive and status chatter. Then work out the +structure: + +- framing: fixed length, terminator bytes, or length-prefixed. +- opcode: the longest common prefix across a command's frames. +- parameters: the bytes that change when you vary one setting (cells per well, + well count). Find their encoders by changing one thing in Chorus and diffing. +- checksum: brute-force the common schemes (sum8, xor8, CRC16 variants) over + candidate byte ranges. + +Each hypothesis carries its evidence (the capture frames it came from) so a human +confirms it before anything is replayed. + +## Step 4: build the ProtocolMap with coverage tracking + +Fold the decoded commands into a `ProtocolMap` seeded with the required set, so the +result reports exactly which required commands are decoded and which are still +missing. Coverage is the gate: the backend will not open a live link until it is +complete. + +## Step 5: guarded replay + +The Melody is a laser plus pressurized fluidics, so replay is conservative by +default, with two independent safety switches, both off by default: + +- `armed` opens the link and permits transmission. Without it, everything is a + dry-run that logs the exact bytes and sends nothing. +- Commands that move fluid, open a nozzle, or fire a sort additionally require an + actuation opt-in (`allow_actuation`) and a human present. + +Confirm the read-only command first. Only once `get_status` round-trips cleanly do +you touch actuating commands, with BD service or your safety officer in the loop. +The backend refuses actuating commands without the opt-in and refuses any live run +against an incomplete map, so a half-mapped protocol can never drive the hardware. + +## Step 6: validate against the instrument + +Run a real sort-to-plate and confirm the deposition. This is the step that moves +the backend's label from "not yet hardware-validated" to "hardware-validated". The +backend reports which state it is in and does not claim a sort it has not run. + +## Physical bridge (still required) + +Software control is not full autonomy. You still need to move a sample tube onto +the SIP and the plate on and off the deposition stage (a bench cobot or a shared +plate hotel), and to expose the sorter's clog and error status as an interlock. A +clogged nozzle silently ruins a plate, so surface it. + +## Decision gate + +Reverse-engineering a closed sorter is real work. If the Melody is not a fixed +constraint, an API-controllable single-cell dispenser (cellenONE, WOLF G2, +Namocell Hana) does the same job, doublet exclusion plus count-controlled plate +deposition, with a documented interface and none of this RE. The orchestrator does +not change: the sorter becomes a different backend behind the same `sort_to_plate` +call. Decide the go or no-go after Step 1, once you know how open Chorus really is. diff --git a/docs/user_guide/capabilities/cell-sorting.md b/docs/user_guide/capabilities/cell-sorting.md new file mode 100644 index 00000000000..b4aecc80add --- /dev/null +++ b/docs/user_guide/capabilities/cell-sorting.md @@ -0,0 +1,61 @@ +# Cell sorting (FACS) + +`CellSorter` controls fluorescence-activated cell sorters that deposit gated +events into the wells of a plate. + +## Why this capability exists + +Sorting is the one step in plate-based single-cell sequencing that usually has to +happen off-automation. A liquid handler can build libraries end to end, but the +sort itself, one gated cell per well or a targeted, enriched population, is +typically a manual hand-off. Modeling the sorter as a capability closes that gap: a +deck stages the plate, calls `sort_to_plate`, and finishes the prep, so walkaway +library prep becomes possible. + +It also lets a workflow sort for a target population, enriching a cell type, a +marker-positive subset, or live singlets before the assay. That is what gives +cell-type resolution to downstream measurements such as cell-type-specific +epigenomics, where the goal is to read regulatory state in the exact cell type a +variant acts in rather than in a bulk average. + +## Interface + +The frontend is hardware-agnostic. The primitive backend operations are: + +- `get_status()` -- a coarse instrument state (`idle`, `running`, `error`). +- `load_template(name)` -- select a pre-built sort template (gate hierarchy). +- `set_deposition(cells_per_well, plate_format)` -- configure the deposition target. +- `prime()` -- prime fluidics and stabilize the stream. +- `start_sort(wells)` -- begin depositing into the staged plate. +- `wait_for_completion(poll_interval, timeout)` -- block until the sort finishes. +- `abort()` -- stop the current sort immediately. +- `clean()` -- run the clean or flush cycle between samples. + +The frontend adds validation and the `sort_to_plate` convenience method, which +sequences load, deposition, prime, sort, wait, and clean in one call. + +The interface is intentionally small and event-oriented so it can sit next to a +future acquisition/cytometry capability that reuses the same fluidics and gate +template concepts, letting a single instrument expose both without duplicating an +interface. + +## Device-free testing + +`CellSorterChatterboxBackend` implements the interface with logging and no I/O, so +you can build and test a workflow without an instrument: + +```python +from pylabrobot.capabilities.cell_sorting import CellSorter, CellSorterChatterboxBackend + +sorter = CellSorter(backend=CellSorterChatterboxBackend()) +await sorter._on_setup() +await sorter.sort_to_plate(cells_per_well=1, wells=96, template="singlet_deposit") +``` + +## Supported hardware + +- BD FACSMelody (`pylabrobot.becton_dickinson.facsmelody`). This backend replays a + decoded `ProtocolMap` and is not yet hardware-validated; it runs end to end dry + and refuses to open a live link until a complete map is supplied. See + [Reverse-engineering the BD FACSMelody](../../facsmelody-re.md) for how the map + is produced and the safety model that governs a live sort. diff --git a/docs/user_guide/capabilities/index.md b/docs/user_guide/capabilities/index.md index 2a091c83f64..6aaf7801e51 100644 --- a/docs/user_guide/capabilities/index.md +++ b/docs/user_guide/capabilities/index.md @@ -48,6 +48,7 @@ shaking fan-control humidity-control centrifuging +cell-sorting sealing peeling tilting diff --git a/pylabrobot/becton_dickinson/__init__.py b/pylabrobot/becton_dickinson/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/becton_dickinson/facsmelody/__init__.py b/pylabrobot/becton_dickinson/facsmelody/__init__.py new file mode 100644 index 00000000000..d4f443c8f7d --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/__init__.py @@ -0,0 +1,12 @@ +from .backend import FACSMelodyCellSorterBackend, FACSMelodyDriver +from .chatterbox import FACSMelodyChatterbox +from .constants import DEFAULT_SORT_TEMPLATE, REQUIRED_COMMANDS, Transport +from .errors import ( + FACSMelodyError, + ProtocolMapIncompleteError, + SortActuationError, + SortNotReadyError, + SortTimeoutError, +) +from .facsmelody import FACSMelody +from .protocol_map import Command, ProtocolMap, seed_required diff --git a/pylabrobot/becton_dickinson/facsmelody/backend.py b/pylabrobot/becton_dickinson/facsmelody/backend.py new file mode 100644 index 00000000000..791ac4aa9d0 --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/backend.py @@ -0,0 +1,350 @@ +"""BD FACSMelody driver and cell-sorting backend. + +The ``FACSMelodyDriver`` is the wire: it owns the control link, loads a decoded +``ProtocolMap``, and sends frames behind two independent safety switches. The +``FACSMelodyCellSorterBackend`` is the protocol: it turns ``CellSorter`` operations +into the frames named in the map. + +Safety +------ +The Melody is a laser plus pressurized fluidics, so the driver is conservative by +default and mirrors the reverse-engineering toolkit's replay guards: + +* Dry-run by default. ``send`` logs the exact bytes and transmits nothing unless + the driver was constructed with ``armed=True`` and the call passes ``live=True``. + Two switches, both off by default. +* Commands that move fluid, open a nozzle, or fire a sort additionally require + ``allow_actuation=True``; otherwise ``send`` raises ``SortActuationError``. +* A live run refuses to start unless every required command in the ProtocolMap is + decoded, so a half-mapped protocol cannot drive hardware. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Optional + +from pylabrobot.capabilities.capability import BackendParams +from pylabrobot.capabilities.cell_sorting.backend import CellSorterBackend +from pylabrobot.device import Driver + +from .constants import ACTUATING_COMMANDS, Transport +from .errors import ( + ProtocolMapIncompleteError, + SortActuationError, + SortNotReadyError, + SortTimeoutError, +) +from .protocol_map import ProtocolMap, seed_required + +logger = logging.getLogger(__name__) + + +class FACSMelodyDriver(Driver): + """Owns the FACSMelody control link and the guarded frame transport. + + Args: + protocol_path: Path to a decoded ProtocolMap JSON. Required for a live run; + omit it to run dry (a required-command map is seeded so orchestration works + end-to-end without hardware). + armed: Open the physical link and allow transmission. Off by default. + allow_actuation: Permit commands that physically actuate the sorter. Off by + default; a human should be present when this is on. + """ + + def __init__( + self, + protocol_path: Optional[str] = None, + *, + armed: bool = False, + allow_actuation: bool = False, + ): + super().__init__() + self.protocol_path = protocol_path + self.armed = armed + self.allow_actuation = allow_actuation + self.pm: Optional[ProtocolMap] = None + self._conn: Optional[_Connection] = None + + async def setup(self, backend_params: Optional[BackendParams] = None) -> None: + if self.protocol_path is not None: + self.pm = ProtocolMap.from_json(self.protocol_path) + else: + self.pm = seed_required() + + if not self.armed: + logger.warning("FACSMelody dry-run (armed=False): no link opened, nothing will transmit.") + return + + coverage = self.pm.coverage() + if coverage["missing"]: + raise ProtocolMapIncompleteError(coverage["missing"]) + if self.pm.endpoint is None: + raise SortNotReadyError("ProtocolMap has no endpoint; cannot open a live link.") + self._conn = _open_connection(self.pm.transport, self.pm.endpoint) + logger.info("FACSMelody link open via %s @ %s", self.pm.transport.value, self.pm.endpoint) + + async def stop(self) -> None: + if self._conn is not None: + self._conn.close() + self._conn = None + + async def send( + self, + name: str, + frame_hex: str, + *, + live: bool = False, + actuating: bool = False, + expect: Optional[bytes] = None, + ) -> Optional[bytes]: + """Send a decoded frame, subject to the safety guards. + + Args: + name: Logical command name (for logging and the actuation guard). + frame_hex: The command bytes as a hex string. + live: Whether this call intends to transmit (still requires ``armed``). + actuating: Force the actuation guard on regardless of ``name``. + expect: Optional bytes expected in the response; a mismatch is logged. + + Returns: + The response bytes, or ``None`` in dry-run. + """ + will_transmit = self.armed and live + if will_transmit and (actuating or name in ACTUATING_COMMANDS) and not self.allow_actuation: + raise SortActuationError( + f"'{name}' actuates the sorter; construct the driver with " + "allow_actuation=True and a human present to send it." + ) + if not will_transmit: + logger.warning("[dry-run] would send '%s': %s", name, frame_hex or "") + return None + if self._conn is None: + raise SortNotReadyError("link is not open; call setup() with armed=True first.") + if not frame_hex: + raise SortNotReadyError( + f"'{name}' has no decoded frame; refusing to transmit an empty frame on a live " + "run. For an emergency stop this fails loud rather than silently doing nothing." + ) + data = bytes.fromhex(frame_hex) + logger.info("SEND '%s': %s", name, frame_hex) + self._conn.write(data) + resp = self._conn.read() + logger.info("RECV: %s", resp.hex() if resp else "") + if expect is not None and resp is not None and expect not in resp: + logger.warning("response did not contain expected %s", expect.hex()) + return resp + + def serialize(self) -> dict: + return { + **super().serialize(), + "protocol_path": self.protocol_path, + "armed": self.armed, + "allow_actuation": self.allow_actuation, + } + + +class FACSMelodyCellSorterBackend(CellSorterBackend): + """Translates ``CellSorter`` operations into FACSMelody frames via the driver.""" + + def __init__(self, driver: FACSMelodyDriver): + self._driver = driver + + def _frame(self, command: str, **params: object) -> str: + """Build the hex frame for a command, substituting ``{param}`` tokens. + + Returns an empty string when the command is undecoded (dry-run), which the + driver logs harmlessly. Live runs always have decoded frames because setup + refuses an incomplete map. + """ + if self._driver.pm is None: + raise SortNotReadyError("driver is not set up; call setup() first.") + cmd = self._driver.pm.commands.get(command) + template = cmd.frame_template if cmd is not None else None + if template is None: + return "" + for key, value in params.items(): + template = template.replace("{" + key + "}", _encode_param(value)) + if "{" in template or "}" in template: + raise ValueError( + f"frame for '{command}' has unsubstituted tokens after filling {list(params)}: " + f"{template!r}. The ProtocolMap template and the backend parameters disagree." + ) + return template + + async def get_status(self) -> str: + resp = await self._driver.send("get_status", self._frame("get_status"), live=True) + # Response parsing is confirmed during hardware validation; until then a live + # link reports 'unknown' rather than a fabricated state. + return "idle" if resp is None else "unknown" + + async def load_template(self, name: str) -> None: + await self._driver.send("load_template", self._frame("load_template", name=name), live=True) + + async def set_deposition(self, cells_per_well: int, plate_format: str) -> None: + await self._driver.send( + "set_deposition", + self._frame("set_deposition", cells=cells_per_well, plate=plate_format), + live=True, + ) + + async def prime(self) -> None: + await self._driver.send("prime", self._frame("prime"), live=True) + + async def start_sort( + self, + wells: int, + backend_params: Optional[BackendParams] = None, + ) -> None: + await self._driver.send("start_sort", self._frame("start_sort", wells=wells), live=True) + + async def wait_for_completion(self, poll_interval: float, timeout: float) -> None: + if not self._driver.armed: + return # dry-run: nothing physical to wait for + waited = 0.0 + while waited < timeout: + if (await self.get_status()) in ("idle", "complete"): + return + await asyncio.sleep(poll_interval) + waited += poll_interval + raise SortTimeoutError(f"sort did not complete within {timeout}s") + + async def abort(self) -> None: + await self._driver.send("abort", self._frame("abort"), live=True) + + async def clean(self) -> None: + await self._driver.send("clean", self._frame("clean"), live=True) + + +def _encode_param(value: object) -> str: + """Default parameter encoder, resolved during hardware validation. + + The byte encoding for each parameter is determined during reverse engineering + (vary one setting, diff the frames) and set per parameter. Until then ints are + emitted as a single byte and everything else as UTF-8 hex, so dry-runs are + legible before the per-parameter encoders are confirmed on the instrument. + + An int that does not fit one byte (e.g. filling all 384 wells) raises rather than + silently wrapping: the provisional width is unresolved, so a value it cannot + represent must fail closed, not encode to a different number. The true width + (and whether such a parameter is multi-byte) is fixed during validation. + """ + if isinstance(value, int): + if not 0 <= value <= 0xFF: + raise ValueError( + f"cannot encode integer parameter {value} as a single byte (0-255); the " + "per-parameter byte width is resolved during hardware validation " + "(see docs/facsmelody-re.md)." + ) + return f"{value:02x}" + return str(value).encode().hex() + + +class _Connection: + """Minimal transport interface: write bytes, read a response, close.""" + + def write(self, data: bytes) -> None: + raise NotImplementedError + + def read(self, size: int = 512) -> bytes: + raise NotImplementedError + + def close(self) -> None: + raise NotImplementedError + + +def _open_connection(transport: Transport, endpoint: str, read_timeout: float = 1.0) -> _Connection: + if transport == Transport.TCP: + return _TcpConnection(endpoint, read_timeout) + if transport == Transport.SERIAL: + return _SerialConnection(endpoint, read_timeout) + if transport == Transport.USB: + return _UsbConnection(endpoint, read_timeout) + raise SortNotReadyError(f"unsupported transport {transport!r}") + + +class _TcpConnection(_Connection): + def __init__(self, endpoint: str, read_timeout: float): + import socket + + host, port = endpoint.rsplit(":", 1) + self._sock = socket.create_connection((host, int(port)), timeout=read_timeout) + + def write(self, data: bytes) -> None: + self._sock.sendall(data) + + def read(self, size: int = 512) -> bytes: + try: + return self._sock.recv(size) + except OSError: + return b"" + + def close(self) -> None: + self._sock.close() + + +class _SerialConnection(_Connection): + def __init__(self, endpoint: str, read_timeout: float): + import serial + + self._port = serial.Serial(endpoint, timeout=read_timeout) + + def write(self, data: bytes) -> None: + self._port.write(data) + + def read(self, size: int = 512) -> bytes: + return bytes(self._port.read(size)) + + def close(self) -> None: + self._port.close() + + +class _UsbConnection(_Connection): + """PyUSB bulk-endpoint wrapper. Endpoint format: ``usb:0xVID:0xPID``. + + Endpoints are auto-detected as the first bulk OUT/IN pair; override if the + Melody turns out to use interrupt endpoints. + """ + + def __init__(self, endpoint: str, read_timeout: float): + import usb.core + import usb.util + + self._timeout_ms = int(read_timeout * 1000) + _, vid, pid = endpoint.split(":") + self._dev = usb.core.find(idVendor=int(vid, 16), idProduct=int(pid, 16)) + if self._dev is None: + raise SortNotReadyError(f"USB device {endpoint} not found") + try: + self._dev.set_configuration() + except Exception: # noqa: BLE001 - device may already be configured + pass + cfg = self._dev.get_active_configuration() + intf = cfg[(0, 0)] + self._ep_out = usb.util.find_descriptor( + intf, + custom_match=lambda e: ( + usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT + ), + ) + self._ep_in = usb.util.find_descriptor( + intf, + custom_match=lambda e: ( + usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN + ), + ) + + def write(self, data: bytes) -> None: + self._ep_out.write(data, timeout=self._timeout_ms) + + def read(self, size: int = 512) -> bytes: + try: + return bytes(self._ep_in.read(size, timeout=self._timeout_ms)) + except Exception: # noqa: BLE001 - a read timeout is a normal empty response + return b"" + + def close(self) -> None: + import usb.util + + usb.util.dispose_resources(self._dev) diff --git a/pylabrobot/becton_dickinson/facsmelody/chatterbox.py b/pylabrobot/becton_dickinson/facsmelody/chatterbox.py new file mode 100644 index 00000000000..c526bb72b0d --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/chatterbox.py @@ -0,0 +1,17 @@ +from typing import Optional + +from .backend import FACSMelodyDriver + + +class FACSMelodyChatterbox(FACSMelodyDriver): + """Device-free FACSMelody driver. + + Pinned to ``armed=False`` so it can never open a link or transmit: it exercises + the real protocol translation (frame building) and logs what it would send, + which makes it useful for building and testing a full ``FACSMelody`` device with + no instrument present. For pure capability-level testing, the ``CellSorter`` + capability also ships ``CellSorterChatterboxBackend``. + """ + + def __init__(self, protocol_path: Optional[str] = None): + super().__init__(protocol_path=protocol_path, armed=False, allow_actuation=False) diff --git a/pylabrobot/becton_dickinson/facsmelody/constants.py b/pylabrobot/becton_dickinson/facsmelody/constants.py new file mode 100644 index 00000000000..2ba87842b18 --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/constants.py @@ -0,0 +1,49 @@ +"""Constants for the BD FACSMelody sorter backend. + +The command set and transport model here are the consumer-side contract: a +``ProtocolMap`` produced by an external reverse-engineering step (see +``docs/facsmelody-re.md``) is expected to decode exactly these commands. This +module holds no decoded bytes; it only names what a complete map must contain. +""" + +from enum import Enum +from typing import List, Tuple + +DEFAULT_SORT_TEMPLATE = "singlet_deposit" +SUPPORTED_PLATE_FORMATS = ("96", "384") + + +class Transport(str, Enum): + """How the host reaches the instrument's control link.""" + + USB = "usb" # PyUSB bulk/interrupt endpoints + SERIAL = "serial" # pyserial COM/tty + TCP = "tcp" # raw TCP socket (some BD carts expose an Ethernet link) + UNKNOWN = "unknown" + + +# The minimum command set a sort-to-plate run needs. A ProtocolMap must decode all +# of these before the backend will drive live hardware. Kept as (name, purpose) +# pairs so an incomplete map can report exactly what is missing. +# +# Note: `connect` and `wait_complete` are part of the required decode set so a +# validated map is provably complete, but the current backend does not yet emit +# them as discrete frames -- setup() opens the link directly and wait_for_completion +# polls get_status. Whether a real Melody needs an explicit connect handshake and a +# blocking wait_complete (vs. status polling) is resolved during hardware validation; +# until then they are decode-required but unsent. See docs/facsmelody-re.md. +REQUIRED_COMMANDS: List[Tuple[str, str]] = [ + ("connect", "open the control link / handshake with the cart"), + ("get_status", "poll instrument state (idle/running/clog/error)"), + ("load_template", "select a pre-built sort template (gate hierarchy) by name"), + ("set_deposition", "set plate format + target cells-per-well for sort-to-plate"), + ("prime", "prime fluidics / start stream, verify break-off stable"), + ("start_sort", "begin depositing into the staged plate"), + ("wait_complete", "block/poll until the plate is fully sorted"), + ("abort", "emergency stop the sort"), + ("clean", "run the clean/flush cycle between samples"), +] + +# Commands that physically actuate the instrument (laser + pressurized fluidics). +# Sending any of these requires an explicit actuation opt-in on the driver. +ACTUATING_COMMANDS = frozenset({"prime", "start_sort", "clean", "set_deposition"}) diff --git a/pylabrobot/becton_dickinson/facsmelody/errors.py b/pylabrobot/becton_dickinson/facsmelody/errors.py new file mode 100644 index 00000000000..0c09e9df045 --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/errors.py @@ -0,0 +1,36 @@ +from typing import List + +from pylabrobot.capabilities.cell_sorting.errors import ( + CellSorterError, + SortActuationError, + SortNotReadyError, + SortTimeoutError, +) + +__all__ = [ + "FACSMelodyError", + "ProtocolMapIncompleteError", + "SortActuationError", + "SortNotReadyError", + "SortTimeoutError", +] + + +class FACSMelodyError(CellSorterError): + """Base class for BD FACSMelody backend errors.""" + + +class ProtocolMapIncompleteError(FACSMelodyError): + """Raised when a live sort is requested but the ProtocolMap is not fully decoded. + + The backend refuses to drive the instrument until every required command has + been reverse-engineered, so a half-mapped protocol can never actuate hardware. + """ + + def __init__(self, missing: List[str]): + self.missing = missing + super().__init__( + "ProtocolMap is incomplete; the following required commands are not yet " + f"decoded: {missing}. Finish the RE decode (see docs/facsmelody-re.md) " + "before a live sort." + ) diff --git a/pylabrobot/becton_dickinson/facsmelody/facsmelody.py b/pylabrobot/becton_dickinson/facsmelody/facsmelody.py new file mode 100644 index 00000000000..bbea99e2e4c --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/facsmelody.py @@ -0,0 +1,72 @@ +"""BD FACSMelody cell sorter. + +Why this matters +---------------- +The FACSMelody is a benchtop sorter with a plate-deposition module. Sorting is the +one step in plate-based single-cell sequencing that usually has to happen +off-automation, so bringing the Melody under PyLabRobot lets a deck run the sort +as a normal device call: one gated cell per well for single-cell work, or a +targeted, enriched population before the assay. That population mode (a cell type, +a marker-positive subset, live singlets) is what gives cell-type resolution to +downstream measurements such as cell-type-specific epigenomics. + +Validation status +----------------- +This backend drives the instrument by replaying a decoded ``ProtocolMap`` produced +by an external reverse-engineering step (see ``docs/facsmelody-re.md``). As +shipped it is not yet hardware-validated: it runs end-to-end dry (building and +logging every frame) and refuses to open a live link until a complete map is +supplied and the instrument is armed. It does not report a sort it has not run. +""" + +from typing import Optional + +from pylabrobot.capabilities.cell_sorting import CellSorter +from pylabrobot.device import Device + +from .backend import FACSMelodyCellSorterBackend, FACSMelodyDriver + +__all__ = ["FACSMelody"] + + +class FACSMelody(Device): + """BD FACSMelody sorter exposing the :class:`CellSorter` capability. + + Args: + protocol_path: Path to a decoded ProtocolMap JSON. Required for a live run; + omit it to build the device in dry-run. + armed: Open the physical link and allow transmission. Off by default. + allow_actuation: Permit commands that physically actuate the sorter. Off by + default; a human should be present when this is on. + """ + + driver: FACSMelodyDriver + + def __init__( + self, + protocol_path: Optional[str] = None, + *, + armed: bool = False, + allow_actuation: bool = False, + ): + driver = FACSMelodyDriver( + protocol_path=protocol_path, + armed=armed, + allow_actuation=allow_actuation, + ) + super().__init__(driver=driver) + self.driver: FACSMelodyDriver = driver + self.sorter = CellSorter(backend=FACSMelodyCellSorterBackend(driver)) + self._capabilities = [self.sorter] + + # serialize is inherited from Device: it emits {"driver": driver.serialize()}, which + # already carries protocol_path/armed/allow_actuation. deserialize is overridden + # because __init__ constructs its own driver rather than taking a driver= argument. + @classmethod + def deserialize(cls, data: dict) -> "FACSMelody": + driver_data = data.get("driver") or data.get("backend") or {} + return cls( + protocol_path=driver_data.get("protocol_path"), + armed=driver_data.get("armed", False), + allow_actuation=driver_data.get("allow_actuation", False), + ) diff --git a/pylabrobot/becton_dickinson/facsmelody/facsmelody_tests.py b/pylabrobot/becton_dickinson/facsmelody/facsmelody_tests.py new file mode 100644 index 00000000000..6e785282a35 --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/facsmelody_tests.py @@ -0,0 +1,355 @@ +"""Tests for the BD FACSMelody backend. + +All tests are device-free. The armed-transmission tests use an in-memory fake +connection so the safety guards can be exercised without an instrument. +""" + +import unittest +from typing import List + +from pylabrobot.becton_dickinson.facsmelody import ( + FACSMelody, + ProtocolMapIncompleteError, + SortActuationError, + SortNotReadyError, + SortTimeoutError, +) +from pylabrobot.becton_dickinson.facsmelody.backend import ( + FACSMelodyCellSorterBackend, + FACSMelodyDriver, + _Connection, +) +from pylabrobot.becton_dickinson.facsmelody.constants import ( + ACTUATING_COMMANDS, + REQUIRED_COMMANDS, + Transport, +) +from pylabrobot.becton_dickinson.facsmelody.protocol_map import Command, ProtocolMap + + +def _complete_map() -> ProtocolMap: + """A fully decoded map: every required command has a frame template.""" + pm = ProtocolMap(transport=Transport.TCP, endpoint="10.0.0.5:9100") + for name, note in REQUIRED_COMMANDS: + template = { + "start_sort": "aa{wells}55", + "set_deposition": "bb{cells}{plate}", + "load_template": "cc{name}", + }.get(name, "00") + pm.commands[name] = Command( + name=name, + transport=Transport.TCP, + frame_template=template, + decoded=True, + notes=note, + ) + return pm + + +class _FakeConnection(_Connection): + """Records written frames; returns no response.""" + + def __init__(self) -> None: + self.written: List[bytes] = [] + + def write(self, data: bytes) -> None: + self.written.append(data) + + def read(self, size: int = 512) -> bytes: + return b"" + + def close(self) -> None: + pass + + +class TestFACSMelodyDryRun(unittest.IsolatedAsyncioTestCase): + async def test_dry_run_sort_runs_without_hardware(self): + dev = FACSMelody() # armed=False + await dev.setup() + await dev.sorter.sort_to_plate(cells_per_well=1, wells=96, template="singlet_deposit") + await dev.stop() + + async def test_armed_with_incomplete_map_refuses(self): + dev = FACSMelody(armed=True) # no protocol_path -> seeded, all undecoded + with self.assertRaises(ProtocolMapIncompleteError) as ctx: + await dev.setup() + self.assertEqual(set(ctx.exception.missing), {name for name, _ in REQUIRED_COMMANDS}) + + +class TestFACSMelodyActuationGuard(unittest.IsolatedAsyncioTestCase): + def _armed_backend(self, allow_actuation: bool): + driver = FACSMelodyDriver(armed=True, allow_actuation=allow_actuation) + driver.pm = _complete_map() + conn = _FakeConnection() + driver._conn = conn # bypass a real link + return FACSMelodyCellSorterBackend(driver), conn + + async def test_actuating_command_blocked_without_opt_in(self): + backend, conn = self._armed_backend(allow_actuation=False) + with self.assertRaises(SortActuationError): + await backend.start_sort(wells=96) + self.assertEqual(conn.written, []) + + async def test_actuating_command_transmits_when_allowed(self): + backend, conn = self._armed_backend(allow_actuation=True) + await backend.start_sort(wells=96) + # 0xaa + (96 & 0xff = 0x60) + 0x55 + self.assertEqual(conn.written, [bytes.fromhex("aa6055")]) + + async def test_readonly_command_transmits_without_actuation_opt_in(self): + backend, conn = self._armed_backend(allow_actuation=False) + await backend.get_status() # not an actuating command + self.assertEqual(conn.written, [bytes.fromhex("00")]) + + +class TestFACSMelodyFraming(unittest.IsolatedAsyncioTestCase): + async def test_frame_substitutes_parameters(self): + driver = FACSMelodyDriver() + driver.pm = _complete_map() + backend = FACSMelodyCellSorterBackend(driver) + # load_template template is "cc{name}"; name encodes as utf-8 hex of "gate" + self.assertEqual(backend._frame("load_template", name="gate"), "cc" + "gate".encode().hex()) + # set_deposition template "bb{cells}{plate}" with cells=50 (0x32), plate="96" + self.assertEqual( + backend._frame("set_deposition", cells=50, plate="96"), + "bb" + "32" + "96".encode().hex(), + ) + + async def test_undecoded_command_yields_empty_frame(self): + driver = FACSMelodyDriver() # seeded lazily on setup; set directly here + from pylabrobot.becton_dickinson.facsmelody.protocol_map import seed_required + + driver.pm = seed_required() + backend = FACSMelodyCellSorterBackend(driver) + self.assertEqual(backend._frame("start_sort", wells=1), "") + + +class TestFACSMelodySerialize(unittest.IsolatedAsyncioTestCase): + async def test_driver_serialize_roundtrips_config(self): + driver = FACSMelodyDriver(protocol_path="p.json", armed=True, allow_actuation=True) + data = driver.serialize() + self.assertEqual(data["protocol_path"], "p.json") + self.assertTrue(data["armed"]) + self.assertTrue(data["allow_actuation"]) + + async def test_device_serialize_roundtrips_config(self): + dev = FACSMelody(protocol_path="p.json", armed=True, allow_actuation=True) + restored = FACSMelody.deserialize(dev.serialize()) + self.assertEqual(restored.driver.protocol_path, "p.json") + self.assertTrue(restored.driver.armed) + self.assertTrue(restored.driver.allow_actuation) + + +class TestFACSMelodyCoverageGate(unittest.IsolatedAsyncioTestCase): + async def test_coverage_flags_required_commands_absent_from_map(self): + # A map that decodes only start_sort and OMITS every other required command must + # still report those as missing, so it cannot pass the live-run gate. + pm = ProtocolMap(transport=Transport.TCP, endpoint="10.0.0.5:9100") + pm.commands["start_sort"] = Command( + name="start_sort", transport=Transport.TCP, frame_template="aa{wells}55", decoded=True + ) + cov = pm.coverage() + required = {name for name, _ in REQUIRED_COMMANDS} + self.assertEqual(set(cov["missing"]), required - {"start_sort"}) + self.assertEqual(cov["total"], len(required)) + self.assertEqual(cov["decoded"], 1) + + async def test_armed_setup_refuses_map_missing_required_commands(self): + # Write a real map JSON that decodes only start_sort, then drive the actual + # setup() gate (via from_json) and confirm it refuses to open a live link. + import tempfile + + pm = ProtocolMap(transport=Transport.TCP, endpoint="10.0.0.5:9100") + pm.commands["start_sort"] = Command( + name="start_sort", transport=Transport.TCP, frame_template="aa{wells}55", decoded=True + ) + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + path = fh.name + pm.to_json(path) + + dev = FACSMelody(protocol_path=path, armed=True) + with self.assertRaises(ProtocolMapIncompleteError) as ctx: + await dev.setup() + required = {name for name, _ in REQUIRED_COMMANDS} + self.assertEqual(set(ctx.exception.missing), required - {"start_sort"}) + + +class TestFACSMelodyEncodeParam(unittest.IsolatedAsyncioTestCase): + async def test_encode_param_refuses_out_of_range_int(self): + from pylabrobot.becton_dickinson.facsmelody.backend import _encode_param + + self.assertEqual(_encode_param(96), "60") + self.assertEqual(_encode_param(255), "ff") + for overflow in (256, 384, 1000): + with self.assertRaises(ValueError): + _encode_param(overflow) + + +class TestFACSMelodyActuationGuardAllCommands(unittest.IsolatedAsyncioTestCase): + def _armed_backend(self, allow_actuation: bool): + driver = FACSMelodyDriver(armed=True, allow_actuation=allow_actuation) + driver.pm = _complete_map() + conn = _FakeConnection() + driver._conn = conn + return FACSMelodyCellSorterBackend(driver), conn + + async def _actuate(self, backend, name): + if name == "prime": + await backend.prime() + elif name == "clean": + await backend.clean() + elif name == "set_deposition": + await backend.set_deposition(cells_per_well=1, plate_format="96") + elif name == "start_sort": + await backend.start_sort(wells=4) + else: + raise AssertionError(f"unhandled actuating command {name}") + + async def test_every_actuating_command_blocked_without_opt_in(self): + backend, conn = self._armed_backend(allow_actuation=False) + # Exercise the exact set the backend declares actuating, so a newly added + # actuating command without a guard test makes this fail. + for name in sorted(ACTUATING_COMMANDS): + with self.subTest(command=name), self.assertRaises(SortActuationError): + await self._actuate(backend, name) + self.assertEqual(conn.written, []) + + async def test_every_actuating_command_transmits_with_opt_in(self): + backend, conn = self._armed_backend(allow_actuation=True) + for name in sorted(ACTUATING_COMMANDS): + await self._actuate(backend, name) + self.assertEqual(len(conn.written), len(ACTUATING_COMMANDS)) + + +class TestFACSMelodyAbort(unittest.IsolatedAsyncioTestCase): + async def test_abort_transmits_and_is_not_actuation_gated(self): + # abort is an emergency stop: it must transmit even with allow_actuation=False. + driver = FACSMelodyDriver(armed=True, allow_actuation=False) + driver.pm = _complete_map() # abort decoded as "00" + conn = _FakeConnection() + driver._conn = conn + backend = FACSMelodyCellSorterBackend(driver) + await backend.abort() + self.assertEqual(conn.written, [bytes.fromhex("00")]) + + async def test_abort_refuses_empty_frame_instead_of_silent_noop(self): + # A "decoded" abort with no frame template must fail loud on a live run, not + # silently transmit nothing (a no-op emergency stop is dangerous). + driver = FACSMelodyDriver(armed=True, allow_actuation=True) + pm = _complete_map() + pm.commands["abort"].frame_template = None + driver.pm = pm + driver._conn = _FakeConnection() + backend = FACSMelodyCellSorterBackend(driver) + with self.assertRaises(SortNotReadyError): + await backend.abort() + + +class TestFACSMelodyFrameGuards(unittest.IsolatedAsyncioTestCase): + async def test_send_refuses_empty_frame_on_live_run(self): + driver = FACSMelodyDriver(armed=True, allow_actuation=True) + driver.pm = _complete_map() + driver._conn = _FakeConnection() + with self.assertRaises(SortNotReadyError): + await driver.send("get_status", "", live=True) + + async def test_frame_raises_on_unsubstituted_token(self): + driver = FACSMelodyDriver() + pm = _complete_map() + pm.commands["start_sort"].frame_template = "aa{wells}{missing}55" + driver.pm = pm + backend = FACSMelodyCellSorterBackend(driver) + with self.assertRaises(ValueError): + backend._frame("start_sort", wells=1) + + +class TestFACSMelodyWaitTimeout(unittest.IsolatedAsyncioTestCase): + async def test_wait_for_completion_times_out_when_never_idle(self): + # Armed against an instrument that never reports idle: wait must raise, not hang. + driver = FACSMelodyDriver(armed=True, allow_actuation=False) + driver.pm = _complete_map() + driver._conn = _FakeConnection() # read() -> b"" -> get_status returns "unknown" + backend = FACSMelodyCellSorterBackend(driver) + with self.assertRaises(SortTimeoutError): + await backend.wait_for_completion(poll_interval=0.001, timeout=0.005) + + +class TestFACSMelodyFrameSequence(unittest.IsolatedAsyncioTestCase): + async def test_backend_primitives_write_expected_frames(self): + driver = FACSMelodyDriver(armed=True, allow_actuation=True) + driver.pm = _complete_map() + conn = _FakeConnection() + driver._conn = conn + backend = FACSMelodyCellSorterBackend(driver) + await backend.load_template(name="gate") + await backend.set_deposition(cells_per_well=50, plate_format="96") + await backend.prime() + await backend.start_sort(wells=4) + await backend.clean() + self.assertEqual( + conn.written, + [ + bytes.fromhex("cc" + "gate".encode().hex()), + bytes.fromhex("bb" + "32" + "96".encode().hex()), + bytes.fromhex("00"), + bytes.fromhex("aa" + "04" + "55"), + bytes.fromhex("00"), + ], + ) + + +class TestProtocolMapRoundTrip(unittest.IsolatedAsyncioTestCase): + async def test_to_json_from_json_roundtrip(self): + import os + import tempfile + + pm = _complete_map() + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + path = fh.name + try: + pm.to_json(path) + loaded = ProtocolMap.from_json(path) + self.assertEqual(loaded.transport, pm.transport) + self.assertEqual(loaded.endpoint, pm.endpoint) + self.assertEqual(set(loaded.commands), set(pm.commands)) + for name, cmd in pm.commands.items(): + self.assertEqual(loaded.commands[name].frame_template, cmd.frame_template) + self.assertEqual(loaded.commands[name].decoded, cmd.decoded) + self.assertEqual(loaded.commands[name].transport, cmd.transport) + self.assertEqual(loaded.coverage()["missing"], []) + finally: + os.unlink(path) + + async def test_from_json_tolerates_unknown_keys(self): + import json + import os + import tempfile + + payload = { + "device": "BD FACSMelody", + "transport": "tcp", + "endpoint": "x:1", + "commands": { + "start_sort": { + "name": "start_sort", + "transport": "tcp", + "frame_template": "aa", + "decoded": True, + "future_field": "ignored", + "another": 123, + }, + }, + } + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + json.dump(payload, fh) + path = fh.name + try: + loaded = ProtocolMap.from_json(path) # must not raise on unknown keys + self.assertEqual(loaded.commands["start_sort"].frame_template, "aa") + self.assertTrue(loaded.commands["start_sort"].decoded) + finally: + os.unlink(path) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/becton_dickinson/facsmelody/protocol_map.py b/pylabrobot/becton_dickinson/facsmelody/protocol_map.py new file mode 100644 index 00000000000..4f977fa0528 --- /dev/null +++ b/pylabrobot/becton_dickinson/facsmelody/protocol_map.py @@ -0,0 +1,120 @@ +"""Consumer-side loader for a decoded FACSMelody ``ProtocolMap``. + +A ``ProtocolMap`` is the deliverable of the reverse-engineering step described in +``docs/facsmelody-re.md``: a JSON file that maps each logical command +(``start_sort``, ``clean``, ...) to the exact bytes that drive it, plus how each +frame is framed and checksummed. That reverse-engineering toolkit is deliberately +kept out of this package; only the read side lives here, so the backend can load a +trusted map and refuse to run against an incomplete one. + +Nothing in this module imports a hardware library, so it is always importable. +""" + +from __future__ import annotations + +import json +import time +from dataclasses import asdict, dataclass, field, fields +from typing import Dict, List, Optional + +from .constants import REQUIRED_COMMANDS, Transport + + +@dataclass +class Command: + """A single, replayable command in the Melody protocol. + + ``frame_template`` is the bytes to send as a hex string, optionally containing + ``{param}`` tokens filled at send time. Everything stays optional until decoding + fills it in; an undecoded ``Command`` still documents intent, which keeps the + backend accurate about what it cannot yet do. + """ + + name: str + transport: Transport = Transport.UNKNOWN + frame_template: Optional[str] = None # hex string, may contain {param} tokens + response_regex: Optional[str] = None + params: Dict[str, str] = field(default_factory=dict) # param -> encoder spec + terminator: Optional[str] = None # hex of frame terminator, if any + checksum: Optional[str] = None # checksum scheme name, if any + evidence: List[str] = field(default_factory=list) # capture-frame refs + decoded: bool = False + notes: str = "" + + +@dataclass +class ProtocolMap: + """Everything needed to drive the Melody headlessly.""" + + device: str = "BD FACSMelody" + transport: Transport = Transport.UNKNOWN + endpoint: Optional[str] = None + commands: Dict[str, Command] = field(default_factory=dict) + created: float = field(default_factory=time.time) + notes: str = "" + + def to_json(self, path: str) -> None: + payload = { + "device": self.device, + "transport": self.transport.value, + "endpoint": self.endpoint, + "created": self.created, + "notes": self.notes, + "commands": { + name: {**asdict(c), "transport": c.transport.value} for name, c in self.commands.items() + }, + } + with open(path, "w") as fh: + json.dump(payload, fh, indent=2) + + @classmethod + def from_json(cls, path: str) -> "ProtocolMap": + with open(path) as fh: + d = json.load(fh) + pm = cls( + device=d.get("device", "BD FACSMelody"), + transport=Transport(d.get("transport", "unknown")), + endpoint=d.get("endpoint"), + created=d.get("created", time.time()), + notes=d.get("notes", ""), + ) + # Tolerate unknown keys: this is the consumer side of an external toolkit whose + # map schema may grow. Drop fields this version does not model rather than + # crashing the whole load on a single extra key. + known = {f.name for f in fields(Command)} + for name, c in d.get("commands", {}).items(): + c = {k: v for k, v in c.items() if k in known} + c["transport"] = Transport(c.get("transport", "unknown")) + pm.commands[name] = Command(**c) + return pm + + def coverage(self) -> dict: + """Report required-command coverage: how many are decoded and which are missing. + + Checked against ``REQUIRED_COMMANDS``, not merely the commands present in the + map, so a map that *omits* a required command entirely is still reported as + missing it. This is the gate a live run must clear (see ``FACSMelodyDriver.setup``), + so it fails closed: a command that is absent, or present but undecoded, counts as + missing. + """ + required = [name for name, _ in REQUIRED_COMMANDS] + missing = [ + name for name in required if name not in self.commands or not self.commands[name].decoded + ] + return { + "decoded": len(required) - len(missing), + "total": len(required), + "missing": missing, + } + + +def seed_required(device: str = "BD FACSMelody") -> ProtocolMap: + """Build a ProtocolMap seeded with the required command set, all undecoded. + + Used as the map for a dry-run instrument and as the explicit target list a real + reverse-engineering pass fills in. + """ + pm = ProtocolMap(device=device) + for name, note in REQUIRED_COMMANDS: + pm.commands[name] = Command(name=name, notes=note, decoded=False) + return pm diff --git a/pylabrobot/capabilities/cell_sorting/__init__.py b/pylabrobot/capabilities/cell_sorting/__init__.py new file mode 100644 index 00000000000..78847af4632 --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/__init__.py @@ -0,0 +1,9 @@ +from .backend import CellSorterBackend +from .cell_sorting import CellSorter +from .chatterbox import CellSorterChatterboxBackend +from .errors import ( + CellSorterError, + SortActuationError, + SortNotReadyError, + SortTimeoutError, +) diff --git a/pylabrobot/capabilities/cell_sorting/backend.py b/pylabrobot/capabilities/cell_sorting/backend.py new file mode 100644 index 00000000000..a971b65119e --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/backend.py @@ -0,0 +1,66 @@ +from abc import ABCMeta, abstractmethod +from typing import Optional + +from pylabrobot.capabilities.capability import BackendParams, CapabilityBackend + + +class CellSorterBackend(CapabilityBackend, metaclass=ABCMeta): + """Abstract backend for cell sorters (FACS) that deposit sorted events into wells. + + A cell sorter gates a stream of cells and dispenses a controlled number of them + into each well of a plate. This is the one step in plate-based single-cell + sequencing that usually has to happen off-automation; exposing it as a + capability lets a deck orchestrate sort-to-plate as a normal device call. + + The interface is intentionally small and event-oriented so it can be shared with + an acquisition/cytometry capability that reuses the same fluidics and gate + template concepts (see the module docstring in ``cell_sorting.py``). Only the + operations a sorter must support are declared here; convenience and validation + live on the ``CellSorter`` frontend. + """ + + @abstractmethod + async def get_status(self) -> str: + """Return a coarse instrument state, e.g. ``idle``, ``running``, ``error``.""" + + @abstractmethod + async def load_template(self, name: str) -> None: + """Select a pre-built sort template (gate hierarchy + sort logic) by name.""" + + @abstractmethod + async def set_deposition(self, cells_per_well: int, plate_format: str) -> None: + """Configure the deposition target: cells per well and plate format.""" + + @abstractmethod + async def prime(self) -> None: + """Prime fluidics and stabilize the stream before sorting.""" + + @abstractmethod + async def start_sort( + self, + wells: int, + backend_params: Optional[BackendParams] = None, + ) -> None: + """Begin depositing into the staged plate. + + Args: + wells: Number of wells to fill. + backend_params: Vendor-specific parameters. + """ + + @abstractmethod + async def wait_for_completion(self, poll_interval: float, timeout: float) -> None: + """Block until the sort completes, polling ``get_status``. + + Args: + poll_interval: Seconds between status polls. + timeout: Maximum seconds to wait before raising. + """ + + @abstractmethod + async def abort(self) -> None: + """Stop the current sort immediately.""" + + @abstractmethod + async def clean(self) -> None: + """Run the clean/flush cycle between samples.""" diff --git a/pylabrobot/capabilities/cell_sorting/cell_sorting.py b/pylabrobot/capabilities/cell_sorting/cell_sorting.py new file mode 100644 index 00000000000..8d8c3f16c1e --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/cell_sorting.py @@ -0,0 +1,121 @@ +"""Cell sorting (FACS) capability. + +Why this exists +--------------- +Sorting is the one step in plate-based single-cell sequencing that usually has to +happen off-automation. A sorter gates a stream of cells and deposits a controlled +number of them into each well; the rest of the library prep can run on a liquid +handler, but the sort itself is typically a manual hand-off. Modeling the sorter +as a capability closes that gap: a deck can call ``sort_to_plate`` the same way it +calls ``spin`` on a centrifuge, depositing one gated cell per well for single-cell +work or a targeted, enriched population, then letting the deck finish the prep. + +Sorting for a target population is the other half: enrich a specific cell type, a +marker-positive subset, or live singlets before the assay. That is what gives +cell-type resolution to downstream measurements, for example cell-type-specific +epigenomics, where you want to read regulatory state in the exact cell type a +variant acts in rather than in a bulk average. + +Scope +----- +This capability covers the sort/deposition side of a FACS instrument. Acquisition +and analysis (event rates, gate statistics, population readout) belong to a +sibling cytometry capability that would reuse the same fluidics and gate-template +primitives; the two are designed to sit next to each other so a single instrument +can expose both without duplicating an interface. +""" + +from typing import Optional + +from pylabrobot.capabilities.capability import BackendParams, Capability, need_capability_ready + +from .backend import CellSorterBackend + +_SUPPORTED_PLATE_FORMATS = ("96", "384") +_WELLS_PER_FORMAT = {"96": 96, "384": 384} + + +class CellSorter(Capability): + """Cell sorting capability: deposit gated events into the wells of a plate. + + The frontend owns validation, orchestration, and convenience methods. The + backend owns the wire protocol for one instrument. See the module docstring for + what this capability unlocks scientifically. + """ + + def __init__(self, backend: CellSorterBackend): + super().__init__(backend=backend) + self.backend: CellSorterBackend = backend + + @need_capability_ready + async def get_status(self) -> str: + """Return a coarse instrument state, e.g. ``idle``, ``running``, ``error``.""" + return await self.backend.get_status() + + @need_capability_ready + async def sort_to_plate( + self, + cells_per_well: int, + wells: int, + template: str, + plate_format: str = "96", + poll_interval: float = 5.0, + timeout: float = 3600.0, + backend_params: Optional[BackendParams] = None, + ) -> None: + """Run a full sort-to-plate cycle. + + Sequences the backend primitives: load the gate template, configure the + deposition target, prime the stream, start the sort, wait for completion, and + clean. Vendors that need a different order can override on the backend. + + Args: + cells_per_well: Target number of cells to deposit per well (must be > 0). + wells: Number of wells to fill (must be > 0). + template: Name of a pre-built sort template (gate hierarchy + sort logic). + plate_format: Plate format, one of ``96`` or ``384``. + poll_interval: Seconds between status polls while waiting. + timeout: Maximum seconds to wait for the sort to complete. + backend_params: Vendor-specific parameters passed to ``start_sort``. + """ + if cells_per_well <= 0: + raise ValueError(f"cells_per_well must be positive, got {cells_per_well}") + if wells <= 0: + raise ValueError(f"wells must be positive, got {wells}") + if plate_format not in _SUPPORTED_PLATE_FORMATS: + raise ValueError( + f"plate_format must be one of {_SUPPORTED_PLATE_FORMATS}, got {plate_format!r}" + ) + capacity = _WELLS_PER_FORMAT[plate_format] + if wells > capacity: + raise ValueError( + f"wells ({wells}) exceeds the {plate_format}-well plate capacity ({capacity})" + ) + + await self.backend.load_template(name=template) + await self.backend.set_deposition(cells_per_well=cells_per_well, plate_format=plate_format) + await self.backend.prime() + await self.backend.start_sort(wells=wells, backend_params=backend_params) + await self.backend.wait_for_completion(poll_interval=poll_interval, timeout=timeout) + await self.backend.clean() + + @need_capability_ready + async def abort(self) -> None: + """Stop the current sort immediately.""" + await self.backend.abort() + + @need_capability_ready + async def clean(self) -> None: + """Run the clean/flush cycle between samples.""" + await self.backend.clean() + + async def _on_stop(self) -> None: + """Leave the sorter safe when the parent Device stops: abort any run in flight. + + Mirrors Pump/Shaker/TemperatureController, which halt their actuation on stop. + ``abort`` is the non-actuating emergency stop, so this is safe in every config + (dry-run logs it, armed transmits it) and never trips the actuation guard. + """ + if self._setup_finished: + await self.backend.abort() + await super()._on_stop() diff --git a/pylabrobot/capabilities/cell_sorting/cell_sorting_tests.py b/pylabrobot/capabilities/cell_sorting/cell_sorting_tests.py new file mode 100644 index 00000000000..f01b5d12c41 --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/cell_sorting_tests.py @@ -0,0 +1,109 @@ +"""Tests for the CellSorter capability.""" + +import unittest +from typing import List, Optional, Tuple + +from pylabrobot.capabilities.capability import BackendParams +from pylabrobot.capabilities.cell_sorting.backend import CellSorterBackend +from pylabrobot.capabilities.cell_sorting.cell_sorting import CellSorter +from pylabrobot.capabilities.cell_sorting.chatterbox import CellSorterChatterboxBackend + + +class _RecordingBackend(CellSorterBackend): + """Records the sequence of primitive calls the frontend makes.""" + + def __init__(self) -> None: + self.calls: List[Tuple[str, tuple]] = [] + + async def get_status(self) -> str: + self.calls.append(("get_status", ())) + return "idle" + + async def load_template(self, name: str) -> None: + self.calls.append(("load_template", (name,))) + + async def set_deposition(self, cells_per_well: int, plate_format: str) -> None: + self.calls.append(("set_deposition", (cells_per_well, plate_format))) + + async def prime(self) -> None: + self.calls.append(("prime", ())) + + async def start_sort( + self, + wells: int, + backend_params: Optional[BackendParams] = None, + ) -> None: + self.calls.append(("start_sort", (wells,))) + + async def wait_for_completion(self, poll_interval: float, timeout: float) -> None: + self.calls.append(("wait_for_completion", ())) + + async def abort(self) -> None: + self.calls.append(("abort", ())) + + async def clean(self) -> None: + self.calls.append(("clean", ())) + + +class TestCellSorter(unittest.IsolatedAsyncioTestCase): + async def test_sort_to_plate_sequences_primitives_in_order(self): + backend = _RecordingBackend() + sorter = CellSorter(backend=backend) + await sorter._on_setup() + + await sorter.sort_to_plate(cells_per_well=25, wells=96, template="singlet_deposit") + + self.assertEqual( + [name for name, _ in backend.calls], + ["load_template", "set_deposition", "prime", "start_sort", "wait_for_completion", "clean"], + ) + self.assertEqual(backend.calls[0], ("load_template", ("singlet_deposit",))) + self.assertEqual(backend.calls[1], ("set_deposition", (25, "96"))) + self.assertEqual(backend.calls[3], ("start_sort", (96,))) + + async def test_requires_setup_before_use(self): + sorter = CellSorter(backend=_RecordingBackend()) + with self.assertRaises(RuntimeError): + await sorter.get_status() + + async def test_rejects_invalid_arguments(self): + sorter = CellSorter(backend=_RecordingBackend()) + await sorter._on_setup() + with self.assertRaises(ValueError): + await sorter.sort_to_plate(cells_per_well=0, wells=96, template="t") + with self.assertRaises(ValueError): + await sorter.sort_to_plate(cells_per_well=1, wells=0, template="t") + with self.assertRaises(ValueError): + await sorter.sort_to_plate(cells_per_well=1, wells=96, template="t", plate_format="6") + + async def test_chatterbox_backend_reports_idle(self): + sorter = CellSorter(backend=CellSorterChatterboxBackend()) + await sorter._on_setup() + self.assertEqual(await sorter.get_status(), "idle") + await sorter.sort_to_plate(cells_per_well=1, wells=4, template="t") + + async def test_rejects_wells_exceeding_plate_capacity(self): + sorter = CellSorter(backend=_RecordingBackend()) + await sorter._on_setup() + with self.assertRaises(ValueError): + await sorter.sort_to_plate(cells_per_well=1, wells=97, template="t", plate_format="96") + # Filling all wells of the matching format is allowed. + await sorter.sort_to_plate(cells_per_well=1, wells=384, template="t", plate_format="384") + + async def test_on_stop_aborts_when_set_up(self): + backend = _RecordingBackend() + sorter = CellSorter(backend=backend) + await sorter._on_setup() + await sorter._on_stop() + self.assertIn(("abort", ()), backend.calls) + self.assertFalse(sorter.setup_finished) + + async def test_on_stop_does_not_abort_when_never_set_up(self): + backend = _RecordingBackend() + sorter = CellSorter(backend=backend) + await sorter._on_stop() + self.assertNotIn(("abort", ()), backend.calls) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/capabilities/cell_sorting/chatterbox.py b/pylabrobot/capabilities/cell_sorting/chatterbox.py new file mode 100644 index 00000000000..9889b080fc6 --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/chatterbox.py @@ -0,0 +1,43 @@ +import logging +from typing import Optional + +from pylabrobot.capabilities.capability import BackendParams + +from .backend import CellSorterBackend + +logger = logging.getLogger(__name__) + + +class CellSorterChatterboxBackend(CellSorterBackend): + """Chatterbox backend for device-free testing. Logs every call, moves no fluid.""" + + async def get_status(self) -> str: + logger.info("Getting sorter status.") + return "idle" + + async def load_template(self, name: str) -> None: + logger.info("Loading sort template %s.", name) + + async def set_deposition(self, cells_per_well: int, plate_format: str) -> None: + logger.info( + "Setting deposition to %s cells/well on %s-well plate.", cells_per_well, plate_format + ) + + async def prime(self) -> None: + logger.info("Priming fluidics.") + + async def start_sort( + self, + wells: int, + backend_params: Optional[BackendParams] = None, + ) -> None: + logger.info("Starting sort into %s wells.", wells) + + async def wait_for_completion(self, poll_interval: float, timeout: float) -> None: + logger.info("Waiting for sort to complete (poll=%ss, timeout=%ss).", poll_interval, timeout) + + async def abort(self) -> None: + logger.info("Aborting sort.") + + async def clean(self) -> None: + logger.info("Running clean cycle.") diff --git a/pylabrobot/capabilities/cell_sorting/errors.py b/pylabrobot/capabilities/cell_sorting/errors.py new file mode 100644 index 00000000000..b5fd8a62d83 --- /dev/null +++ b/pylabrobot/capabilities/cell_sorting/errors.py @@ -0,0 +1,19 @@ +class CellSorterError(Exception): + """Base class for cell sorter errors.""" + + +class SortNotReadyError(CellSorterError): + """Raised when a sort is requested but the instrument is not ready to sort.""" + + +class SortActuationError(CellSorterError): + """Raised when an actuating command is issued without the required safety opt-in. + + Sorters combine a laser with pressurized fluidics, so backends should refuse to + physically actuate unless the caller has explicitly armed the instrument and + allowed actuation. + """ + + +class SortTimeoutError(CellSorterError): + """Raised when a sort does not complete within the allotted time."""