diff --git a/docs/_static/devices.json b/docs/_static/devices.json index 53b04127422..b1af611518f 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -2153,6 +2153,20 @@ "doc_slug": "li_cor/odyssey/hello-world", "notes": "HTTP scan control and TIFF retrieval; port pending hardware verification." }, + { + "id": "thermo-fisher-kingfisher-flex", + "vendor": "Thermo Fisher", + "name": "KingFisher Flex", + "kind": "magnetic particle processor", + "capabilities": ["magnetic bead processing"], + "status": "wip", + "needs_hardware_testing": true, + "api": "pylabrobot.thermo_fisher.kingfisher.flex.KingFisherFlex", + "api_version": "v1", + "code_slug": "thermo_fisher/kingfisher/flex", + "doc_slug": "thermo_fisher/kingfisher/flex/hello-world", + "notes": "96DW local BDZ execution tested on macOS/firmware 1.01.0, including 37 C heating, beads, drying and PLR confirmations. Partial tip pickup remains unresolved; broader formats/features and general protocol authoring remain unsupported." + }, { "id": "thermo-fisher-kingfisher-presto", "vendor": "Thermo Fisher", diff --git a/docs/api/pylabrobot.thermo_fisher.rst b/docs/api/pylabrobot.thermo_fisher.rst index ef6627708a3..00791335574 100644 --- a/docs/api/pylabrobot.thermo_fisher.rst +++ b/docs/api/pylabrobot.thermo_fisher.rst @@ -102,6 +102,31 @@ KingFisher Presto ProtocolStep ProtocolTimeLeft +KingFisher Flex +--------------- + +.. currentmodule:: pylabrobot.thermo_fisher.kingfisher.flex + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + KingFisherFlex + FlexError + FlexEvent + FlexRunResult + FlexTranscriptEntry + FlexProtocol + FlexProtocolError + FlexProtocolIssue + FlexCommand + FlexPlate + FlexDispense + FlexTip + FlexPrompt + read_flex_bdz + BindIt protocol files --------------------- diff --git a/docs/user_guide/thermo_fisher/kingfisher/flex/hello-world.ipynb b/docs/user_guide/thermo_fisher/kingfisher/flex/hello-world.ipynb new file mode 100644 index 00000000000..c6dcf43abfd --- /dev/null +++ b/docs/user_guide/thermo_fisher/kingfisher/flex/hello-world.ipynb @@ -0,0 +1,250 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "76d09de2", + "metadata": {}, + "source": [ + "# KingFisher Flex\n", + "\n", + "```{device-card} thermo-fisher-kingfisher-flex\n", + "```\n", + "\n", + "Run a local BindIt BDZ directly through PLR with explicit plate and protocol-pause confirmations.\n", + "The tested setup is a matching **96DW head, tip comb and plates**, firmware 1.01.0 on macOS.\n", + "Hardware checks include mixing, bead collection/release, both drying positions, 37 C heating,\n", + "and a reagent-addition pause. Other documented temperature setpoints are accepted but not\n", + "individually hardware-tested. Linux and other head/plate formats remain unvalidated.\n", + "\n", + "**Experimental driver:** partial tip-comb attachment may occur without a firmware error.\n", + "Command completion does not verify attachment. Inspect pickup and\n", + "physical clearances; there is no automatic retry or recovery. See the\n", + "[support and operation guide](validation.md).\n", + "\n", + "The first version executes supported exports; general protocol authoring is deferred.\n", + "Onboard virtual Start and protocol upload remain unsupported.\n" + ] + }, + { + "cell_type": "markdown", + "id": "00b27c3c", + "metadata": {}, + "source": [ + "## How it communicates\n", + "\n", + "The Flex uses length-prefixed 64-byte HID reports carrying ASCII command lines. Command acknowledgment and execution completion are distinct. The driver continually reads responses, including temperature reports during motion and operator waits. It does not use the Presto XML protocol or its acknowledgment commands.\n", + "\n", + "## Physical setup\n", + "\n", + "Prepare the installed 96DW head, matching tips and supported plates according to the instrument instructions. Close BindIt before connecting PLR. On Linux install PLR with its HID extra, the system HIDAPI library, and device permissions as described in the validation guide. Each hardware cell below is a separate operation to review; do not run the entire notebook unattended." + ] + }, + { + "cell_type": "markdown", + "id": "45ae786f", + "metadata": {}, + "source": [ + "## Inspect a protocol offline\n", + "\n", + "Set the path to your original BindIt export. Reading and preflight do not open USB. Unknown script forms remain inspectable but cannot execute. Plate labels are not printed tray numbers." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4c82c377", + "metadata": {}, + "outputs": [], + "source": [ + "from pathlib import Path\n", + "from pylabrobot.thermo_fisher.kingfisher.flex import KingFisherFlex, read_flex_bdz\n", + "\n", + "protocol = read_flex_bdz(Path(\"protocol.bdz\"))\n", + "commands = protocol.preflight()\n", + "(protocol.name, protocol.sha256, len(commands), protocol.prompts())" + ] + }, + { + "cell_type": "markdown", + "id": "51bb761a", + "metadata": {}, + "source": [ + "## Connect and identify\n", + "\n", + "This cell accesses real hardware. `setup()` sends only VER; it does not run initialization or acknowledge faults. Specify `serial_number` if more than one instrument is connected." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c03729f7", + "metadata": {}, + "outputs": [], + "source": [ + "flex = KingFisherFlex()\n", + "await flex.setup()\n", + "(flex.version, flex.serial)" + ] + }, + { + "cell_type": "markdown", + "id": "b053d5bd", + "metadata": {}, + "source": [ + "## Initialize explicitly\n", + "\n", + "**This moves the instrument.** Check the instrument and clearances first. Initialization\n", + "has been tested on firmware 1.01.0; the observed final orientation is printed tray 1\n", + "at processing and tray 4 at loading. Initialization is separate from opening USB." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2ab44d80", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.initialize()\n", + "flex.processing_slot" + ] + }, + { + "cell_type": "markdown", + "id": "42e2d8b8", + "metadata": {}, + "source": [ + "## Observe events\n", + "\n", + "Start one consumer before running. Host load/unload/pause events announce an action;\n", + "`host_prompt_completed` records its confirmation. Temperature events preserve heater and\n", + "ambient readings. They do not certify well-liquid temperature or target attainment." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8af32021", + "metadata": {}, + "outputs": [], + "source": [ + "import asyncio\n", + "import logging\n", + "\n", + "logging.basicConfig(level=logging.INFO)\n", + "\n", + "\n", + "async def observe_events():\n", + " async for event in flex.events():\n", + " logging.info(\"%s: %s; prompt=%s\", event.kind, event.text, event.prompt)\n", + "\n", + "\n", + "observer = asyncio.create_task(observe_events())" + ] + }, + { + "cell_type": "markdown", + "id": "bc0f00d7", + "metadata": {}, + "source": [ + "## Define operator confirmations\n", + "\n", + "This interactive example uses a separate input thread so instrument telemetry continues\n", + "while you respond. Returning exactly `True` confirms the action. Any other response stops\n", + "streaming and leaves the session uncertain; it does not automatically abort physical motion.\n", + "The same handler is passed separately for plate actions and protocol pauses." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "123ebf83", + "metadata": {}, + "outputs": [], + "source": [ + "async def confirm_action(prompt):\n", + " print(f\"{prompt.action.upper()}: {prompt.label}\")\n", + " print(f\"Loading tray {prompt.loading_slot}; processing tray {prompt.processing_slot}\")\n", + " answer = await asyncio.to_thread(\n", + " input, \"Complete the action, clear your hands, then type CONFIRM: \"\n", + " )\n", + " return answer.strip() == \"CONFIRM\"" + ] + }, + { + "cell_type": "markdown", + "id": "3488faef", + "metadata": {}, + "source": [ + "## Execute the existing protocol\n", + "\n", + "This cell starts motion and may start heating before loading. Complete each requested\n", + "action before confirming in Python; do not press physical Start in this host-confirmed mode.\n", + "For reagent additions, the prompt includes the plate, reagent and volume per well.\n", + "Text mentioning seconds in a pause label does not set a timer.\n", + "\n", + "The await returns only after all script commands and finalization complete. The driver\n", + "serializes commands; validate your method's physical behavior and outcome for its intended use.\n", + "To use physical Start instead, omit both confirmation callbacks from `run_protocol`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5f73c396", + "metadata": {}, + "outputs": [], + "source": [ + "result = await flex.run_protocol(\n", + " protocol,\n", + " confirm_prompt=confirm_action,\n", + " confirm_pause=confirm_action,\n", + " operator_timeout=None,\n", + ")\n", + "result" + ] + }, + { + "cell_type": "markdown", + "id": "8ff0e980", + "metadata": {}, + "source": [ + "## Abort an active run when needed\n", + "\n", + "Use `await flex.abort()` from a concurrently running task to request abort while a command is pending. The abort path sends the Feature/ESC abort sequence and awaits ABO END. In a script, keep the run as an `asyncio.Task` if another task must abort it.\n", + "\n", + "Canceling the Python task or interrupting a notebook cell **does not automatically stop the instrument**. Use explicit abort if the connection remains available, or the instrument controls if communication is lost. After any interruption inspect physical positions, call `stop()`, then establish a fresh session. Do not replay the unfinished script automatically." + ] + }, + { + "cell_type": "markdown", + "id": "59f31c80", + "metadata": {}, + "source": [ + "## Disconnect\n", + "\n", + "`stop()` closes USB and wakes the event observer. It is not a motion-stop operation; use it here after the run has completed or after explicitly handling interruption." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1e56dfff", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.stop()\n", + "await observer" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/thermo_fisher/kingfisher/flex/validation.md b/docs/user_guide/thermo_fisher/kingfisher/flex/validation.md new file mode 100644 index 00000000000..15d4ff5af76 --- /dev/null +++ b/docs/user_guide/thermo_fisher/kingfisher/flex/validation.md @@ -0,0 +1,128 @@ +# KingFisher Flex support and operation + +The driver executes local BindIt BDZ exports for the **96DW head, tip comb and +plates**. It preserves exported motion coordinates, command parameters and order. +It does not calculate positions from liquid volumes or generate protocols. + +Hardware validation covers firmware 1.01.0 on macOS: identification, initialization, +loading/unloading through Python callbacks, mixing, bead collection/release, +precollection, postmix, drying outside and above wells, automatic timed pauses, +expanded mixing loops, manual pauses with reagent additions, and 37 °C heating +with and without preheat. Other firmware versions and operating systems require +validation. Standard KF96, PCR and 24-head configurations are not supported. + +The driver remains experimental. Partial tip-comb attachment can occur without a +firmware error; command completion does not verify attachment. Inspect pickup and +clearance. The driver does not retry motion or recover an interrupted method. + +## Installation + +Install PyLabRobot with the HID extra and a system HIDAPI library: + +```console +pip install 'pylabrobot[hid]' +``` + +On macOS, install HIDAPI with `brew install hidapi`. If Python cannot find +`libhidapi.dylib`, set `DYLD_LIBRARY_PATH` to the Homebrew library directory when +launching Python (typically `/opt/homebrew/lib` on Apple Silicon). +On Linux, install the distribution's HIDAPI library and configure USB permissions +for vendor `0x0AB6`, product `0x02C7`. Close BindIt before opening the device in PLR. +All communication uses `pylabrobot.io.HID`. + +## Inspect before executing + +`read_flex_bdz(path)` accepts a path or bytes. It validates the binary container, +compressed XML, Flex identity and references, retaining the original bytes and +script. `compatibility_issues()` reports unsupported features with source locations; +`preflight()` raises if any blocker remains. Inspection does not open USB. + +```python +from pylabrobot.thermo_fisher.kingfisher.flex import read_flex_bdz + +protocol = read_flex_bdz("protocol.bdz") +issues = protocol.compatibility_issues() +commands = protocol.preflight() +prompts = protocol.prompts() +``` + +An accepted export is supported by the parser; this does not establish validation +of that particular method. Unsupported commands are never silently dropped. +Streaming excludes only `DTL` and `TIM` metadata. BindIt expands loops into commands; +the driver sends that sequence once. + +The CLI provides offline checks and an explicit identification-only probe: + +```console +plr dev kingfisher-flex inspect protocol.bdz +plr dev kingfisher-flex audit /path/to/protocols +plr dev kingfisher-flex probe +``` + +`audit` scans directories nonrecursively, reports each invalid or unsupported file, +and exits with status 1 if any input fails. `probe` opens real hardware, sends `VER` +and disconnects. It does not initialize or start a method. + +## Execution and confirmations + +Follow the [hello-world notebook](hello-world.ipynb) for connection and execution. +`setup()` identifies the device. `initialize()` explicitly moves the axes, disables +heating and presents tray 4 at loading, with tray 1 at processing. + +`run_protocol(path_or_protocol)` checks the entire export before transmitting its +execution preamble. It serializes commands, waiting for acknowledgment and terminal +completion separately. Timed mixing and timer waits receive duration-aware deadlines. +Temperature telemetry continues during motion and operator waits. + +By default, `PAU` commands wait for the physical Start button. Pass an asynchronous +`confirm_prompt` callback to handle loading/unloading in Python, and a separate +`confirm_pause` callback for protocol pauses. Each callback must return exactly +`True` after its action is complete. The corresponding `PAU` is then omitted from +the transmitted sequence. This does not remotely press Start on an existing pause. + +`FlexPrompt` contains the action, label, processing tray and loading tray. +`additions` describes any required reagent additions with plate, reagent and +per-well volume in microliters. The label also includes these instructions. +Confirmation acknowledges completion; the driver does not dispense liquid. +Pause text mentioning a duration is descriptive: only exported timer commands +create automatic timed waits. Each addition pause presents one plate. + +`operator_timeout=None` allows an indefinite operator wait. A rejected confirmation, +callback error, timeout, cancellation, instrument error or disconnect prevents +subsequent commands and invalidates the session. No failed motion is resent. + +## Heating and events + +`TEM target interval` uses integer degrees Celsius and a reporting interval in +centiseconds. Supported targets are `0` (RT/off) or `10..96`; the interval is `0` +(disabled) or at least `100`. The instrument can reject a target based on calibration. +Hardware validation covers 37 °C. `MOI 310` raises the heater block and `MOI 0` lowers +it; the driver preserves those commands independently of the temperature setting. + +`TEM END 0` acknowledges setting completion, not target attainment. The driver adds +no temperature wait beyond the exported sequence. Temperature reports describe the +heater and ambient sensors, not independently measured liquid temperature. + +`events()` supplies temperature, step, prompt, run-completion and failure events. +A single consumer should read the event stream. `FlexRunResult` reports the source +BDZ digest, completed script-command count and host-confirmation count after +finalization. `transcript` exposes timestamped command/response entries for diagnostics. + +## Interruption and disconnect + +`abort()` sends the instrument abort sequence and waits for `ABO END`. It can run +from a supervising task while execution or a confirmation is pending. Abort can +move axes; completion does not establish a safe physical position. + +`stop()` only disconnects. Canceling a Python task does not stop the instrument. +After interruption, inspect physical state, close the connection and establish a +fresh session before deliberately restarting. There is no automatic resume. + +## Onboard protocols + +`list_protocols()` reads onboard names, including maintenance routines. +`start_stored_protocol(name)` requires an exact unique inventory match and starts +that sequence using `BGN`. Hardware validation covers launch to the first loading +prompt only. Internal loading pauses are not individually exposed to PLR callbacks; +physical Start remains necessary. `resume()` raises `NotImplementedError`. +Protocol upload and virtual advancement of onboard protocols are unsupported. diff --git a/docs/user_guide/thermo_fisher/kingfisher/index.md b/docs/user_guide/thermo_fisher/kingfisher/index.md index d0fcf3ec489..fd5af6296bd 100644 --- a/docs/user_guide/thermo_fisher/kingfisher/index.md +++ b/docs/user_guide/thermo_fisher/kingfisher/index.md @@ -4,4 +4,6 @@ :maxdepth: 1 presto/hello-world +flex/hello-world +flex/validation ``` diff --git a/pylabrobot/cli_registry.py b/pylabrobot/cli_registry.py index 2218a5d0f99..23ecf47a171 100644 --- a/pylabrobot/cli_registry.py +++ b/pylabrobot/cli_registry.py @@ -27,4 +27,7 @@ def load(self) -> DeviceCLIHandler: DEVICE_CLIS: Mapping[str, DeviceCLI] = { "celigo": DeviceCLI("pylabrobot.revvity.celigo.cli", "Celigo gRPC service and inspection tools"), + "kingfisher-flex": DeviceCLI( + "pylabrobot.thermo_fisher.kingfisher.flex.cli", "Flex protocol inspection and identification" + ), } diff --git a/pylabrobot/thermo_fisher/__init__.py b/pylabrobot/thermo_fisher/__init__.py index 125b55ed5d9..5c464760f3b 100644 --- a/pylabrobot/thermo_fisher/__init__.py +++ b/pylabrobot/thermo_fisher/__init__.py @@ -1,3 +1,3 @@ """Thermo Fisher instruments.""" -from .kingfisher import KingFisherPresto +from .kingfisher import KingFisherFlex, KingFisherPresto diff --git a/pylabrobot/thermo_fisher/kingfisher/__init__.py b/pylabrobot/thermo_fisher/kingfisher/__init__.py index fed0656ee4f..57bed5c184b 100644 --- a/pylabrobot/thermo_fisher/kingfisher/__init__.py +++ b/pylabrobot/thermo_fisher/kingfisher/__init__.py @@ -1,3 +1,4 @@ """Thermo Fisher KingFisher instruments and BindIt protocol files.""" +from .flex import KingFisherFlex from .presto import KingFisherPresto diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/__init__.py b/pylabrobot/thermo_fisher/kingfisher/flex/__init__.py new file mode 100644 index 00000000000..de41067f7f6 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/__init__.py @@ -0,0 +1,15 @@ +"""Experimental Flex driver and lossless BindIt protocol inspection.""" + +from .connection import FlexError, FlexTranscriptEntry +from .flex import FlexEvent, FlexRunResult, KingFisherFlex +from .protocol import ( + FlexCommand, + FlexDispense, + FlexPlate, + FlexPrompt, + FlexProtocol, + FlexProtocolError, + FlexProtocolIssue, + FlexTip, + read_flex_bdz, +) diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/cli.py b/pylabrobot/thermo_fisher/kingfisher/flex/cli.py new file mode 100644 index 00000000000..e84709ce301 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/cli.py @@ -0,0 +1,130 @@ +"""Offline Flex export inspection and an explicit VER-only hardware probe.""" + +import argparse +import asyncio +import json +from collections.abc import Sequence +from dataclasses import asdict +from pathlib import Path +from typing import Optional + +from pylabrobot.thermo_fisher.kingfisher.flex.flex import KingFisherFlex +from pylabrobot.thermo_fisher.kingfisher.flex.protocol import FlexProtocolError, read_flex_bdz + + +def inspect_protocol(path: Path) -> dict: + """Return inspection and execution-coverage information without constructing HID.""" + protocol = read_flex_bdz(path) + result = { + "name": protocol.name, + "sha256": protocol.sha256, + "bindit_version": protocol.bindit_version, + "script_lines": len(protocol.script.splitlines()), + "plates": [asdict(plate) for plate in protocol.plates], + "tips": [asdict(tip) for tip in protocol.tips], + } + issues = protocol.compatibility_issues() + result["execution_supported"] = not issues + result["issues"] = [asdict(issue) for issue in issues] + result["hardware_validation"] = "Not determined by offline inspection" + if not issues: + commands = protocol.preflight() + result.update( + { + "stream_preflight": "supported", + "stream_commands": len(commands), + "prompts": [asdict(prompt) for prompt in protocol.prompts()], + } + ) + else: + result.update( + { + "stream_preflight": "unsupported", + "reason": "; ".join(f"{issue.location}: {issue.message}" for issue in issues), + } + ) + return result + + +def audit_protocols(paths: Sequence[Path]) -> dict: + """Inspect files or directories of BDZs, retaining failures instead of stopping early. + + Directories are scanned nonrecursively. Explicit files need not have a particular + extension. An empty/unreadable directory or invalid export counts as a failure. + No source is rewritten and no HID connection is constructed. + """ + results = [] + for path in paths: + try: + files = ( + sorted(p for p in path.iterdir() if p.is_file() and p.suffix.lower() == ".bdz") + if path.is_dir() + else [path] + ) + if not files: + raise ValueError("Directory contains no BDZ files") + except (OSError, ValueError) as error: + results.append({"path": str(path), "status": "invalid", "reason": str(error)}) + continue + for file in files: + try: + inspected = inspect_protocol(file) + except (OSError, FlexProtocolError) as error: + results.append({"path": str(file), "status": "invalid", "reason": str(error)}) + else: + results.append( + { + "path": str(file), + "status": "supported" if inspected["execution_supported"] else "unsupported", + **inspected, + } + ) + return { + "hardware_accessed": False, + "total": len(results), + "supported": sum(r["status"] == "supported" for r in results), + "unsupported": sum(r["status"] == "unsupported" for r in results), + "invalid": sum(r["status"] == "invalid" for r in results), + "protocols": results, + } + + +async def _probe(serial_number: Optional[str], transcript_path: Optional[Path]) -> dict: + """Open the selected Flex, send VER only, and close it even on failure.""" + instrument = KingFisherFlex(serial_number=serial_number) + try: + await instrument.setup() + return {"model": "KingFisherFlex", "firmware": instrument.version, "serial": instrument.serial} + finally: + await instrument.stop() + if transcript_path is not None: + transcript_path.write_text( + json.dumps([asdict(entry) for entry in instrument.transcript], indent=2) + "\n" + ) + + +def main(argv: Optional[Sequence[str]] = None) -> None: + """Dispatch offline commands or an explicitly selected identification-only probe.""" + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="operation", required=True) + inspect = commands.add_parser("inspect", help="Inspect a Flex .bdz without USB access") + inspect.add_argument("path", type=Path) + audit = commands.add_parser("audit", help="Check all BDZ files in folders, without USB access") + audit.add_argument("paths", type=Path, nargs="+", help="Files or nonrecursive directories") + probe = commands.add_parser("probe", help="Connect to real hardware and send VER only") + probe.add_argument("--serial-number") + probe.add_argument("--transcript", type=Path) + args = parser.parse_args(argv) + if args.operation == "inspect": + result = inspect_protocol(args.path) + elif args.operation == "audit": + result = audit_protocols(args.paths) + else: + result = asyncio.run(_probe(args.serial_number, args.transcript)) + print(json.dumps(result, indent=2)) + if args.operation == "audit" and (result["unsupported"] or result["invalid"]): + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/cli_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/cli_tests.py new file mode 100644 index 00000000000..9b66bc5050c --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/cli_tests.py @@ -0,0 +1,66 @@ +"""Offline CLI behavior using temporary synthetic input files.""" + +import contextlib +import io +import json +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from .cli import _probe, main +from .test_helpers import bdz_bytes + + +class FlexCLITests(unittest.TestCase): + """Inspection and auditing never create a hardware connection.""" + + def test_inspect_and_audit_report_failures_without_usb(self) -> None: + """Directory audits retain invalid inputs and use a failing exit status.""" + with ( + TemporaryDirectory() as tmp, + patch( + "pylabrobot.thermo_fisher.kingfisher.flex.cli.KingFisherFlex", + side_effect=AssertionError("offline command opened hardware"), + ), + ): + folder = Path(tmp) + path = folder / "input.bdz" + path.write_bytes(bdz_bytes()) + out = io.StringIO() + with contextlib.redirect_stdout(out): + main(["inspect", str(path)]) + self.assertTrue(json.loads(out.getvalue())["execution_supported"]) + (folder / "unsupported.BDZ").write_bytes(bdz_bytes("ABC 1\n")) + (folder / "invalid.bdz").write_bytes(b"invalid") + out = io.StringIO() + with contextlib.redirect_stdout(out), self.assertRaises(SystemExit) as error: + main(["audit", str(folder)]) + self.assertEqual(error.exception.code, 1) + result = json.loads(out.getvalue()) + self.assertEqual((result["supported"], result["unsupported"], result["invalid"]), (1, 1, 1)) + self.assertFalse(result["hardware_accessed"]) + + def test_help_requires_no_hardware(self) -> None: + """Help exits without importing or constructing a live transport.""" + with contextlib.redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as error: + main(["--help"]) + self.assertEqual(error.exception.code, 0) + + +class FlexProbeTests(unittest.IsolatedAsyncioTestCase): + """Probe lifecycle with a simulated transport.""" + + async def test_probe_identifies_and_closes(self) -> None: + """The explicit probe sends only VER and closes its transport.""" + from .flex import KingFisherFlex + from .flex_tests import ScriptedHID + + flex = KingFisherFlex() + transport = ScriptedHID() + flex.io = transport + with patch("pylabrobot.thermo_fisher.kingfisher.flex.cli.KingFisherFlex", return_value=flex): + result = await _probe(None, None) + self.assertEqual(transport.commands, ["VER"]) + self.assertEqual(transport.closed, 1) + self.assertEqual(result["firmware"], "1.01.0") diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/connection.py b/pylabrobot/thermo_fisher/kingfisher/flex/connection.py new file mode 100644 index 00000000000..8742ad91167 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/connection.py @@ -0,0 +1,320 @@ +"""Flex ASCII framing and serialized command exchange over PyLabRobot HID.""" + +import asyncio +import logging +import math +import re +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Literal, Optional + +from pylabrobot.io.hid import HID + +logger = logging.getLogger(__name__) + + +class FlexError(Exception): + """A Flex failure preserving the command and raw instrument error code.""" + + def __init__( + self, message: str, command: Optional[str] = None, code: Optional[int] = None + ) -> None: + """Store failure details without borrowing another instrument's error table.""" + super().__init__(message) + self.command = command + self.code = code + + +def validate_timeout(value: Optional[float], name: str, allow_none: bool = False) -> None: + """Reject invalid deadlines before any command can be transmitted.""" + if value is None: + if allow_none: + return + raise ValueError(f"{name} is required") + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{name} must be positive and finite") + + +def reports(payload: bytes) -> tuple[bytes, ...]: + """Encode payload bytes into 64-byte reports, excluding the HID report ID.""" + if not payload: + raise ValueError("Cannot encode an empty payload") + return tuple( + bytes([len(payload[offset : offset + 63])]) + payload[offset : offset + 63].ljust(63, b"\x00") + for offset in range(0, len(payload), 63) + ) + + +class FlexLines: + """Reassemble CRLF lines, ignoring all bytes beyond a report's length prefix.""" + + def __init__(self) -> None: + """Start with an empty receive buffer.""" + self._buffer = bytearray() + + def feed(self, report: bytes) -> tuple[str, ...]: + """Return complete ASCII lines from a bounded HID report.""" + if not report or len(report) > 64 or report[0] > 63 or report[0] > len(report) - 1: + raise FlexError("Invalid Flex HID report length") + self._buffer.extend(report[1 : 1 + report[0]]) + lines = [] + while b"\r\n" in self._buffer: + end = self._buffer.index(b"\r\n") + raw = bytes(self._buffer[:end]) + del self._buffer[: end + 2] + if not raw or len(raw) > 4096 or any(byte < 32 or byte > 126 for byte in raw): + raise FlexError("Invalid ASCII response line") + lines.append(raw.decode("ascii")) + if len(self._buffer) > 4096: + raise FlexError("Unterminated Flex response exceeds 4096 bytes") + return tuple(lines) + + +@dataclass(frozen=True) +class FlexMessage: + """An acknowledgment, data report, completion, or unsolicited termination.""" + + kind: Literal["ack", "report", "completion", "interrupted"] + opcode: str + text: str + code: Optional[int] = None + + +def parse_message(line: str) -> FlexMessage: + """Parse response structure without inferring undocumented numeric meanings.""" + if line == "QIT": + return FlexMessage("interrupted", "QIT", line) + match = re.fullmatch(r"([A-Z]{3}) END (-?[0-9]+)", line) + if match: + return FlexMessage("completion", match[1], line, int(match[2])) + if re.fullmatch(r"[A-Z]{3}", line): + return FlexMessage("ack", line, line) + if re.fullmatch(r"[A-Z]{3} [ -~]+", line): + return FlexMessage("report", line[:3], line) + raise FlexError(f"Malformed Flex response: {line!r}") + + +@dataclass(frozen=True) +class FlexTranscriptEntry: + """Monotonic session transcript; feature report payloads are hexadecimal.""" + + timestamp: float + direction: Literal["out", "in", "feature"] + text: str + + +@dataclass +class _Pending: + """One command's independent acknowledgment and terminal state.""" + + opcode: str + ack: asyncio.Event = field(default_factory=asyncio.Event) + done: asyncio.Event = field(default_factory=asyncio.Event) + data: list[str] = field(default_factory=list) + + +class FlexConnection: + """Keep one command in flight while continuously consuming instrument reports. + + Timeouts/cancellation invalidate the session because replies have no unique + transaction IDs. The reader stays available for explicit abort; no motion is + resent, and a fresh setup requires stop first. + """ + + def __init__( + self, + io: HID, + on_message: Callable[[FlexMessage], None], + on_failure: Callable[[FlexError], None], + timeout: float, + ) -> None: + """Configure transport and observers without connecting to hardware.""" + validate_timeout(timeout, "timeout") + self.io = io + self.timeout = timeout + self._on_message = on_message + self._on_failure = on_failure + self._reader: Optional[asyncio.Task[None]] = None + self._pending: Optional[_Pending] = None + self._failure: Optional[FlexError] = None + self._commands = asyncio.Lock() + self._writes = asyncio.Lock() + self._abort_lock = asyncio.Lock() + self._abort_done: Optional[asyncio.Future[int]] = None + self.transcript: list[FlexTranscriptEntry] = [] + + def _record(self, direction: Literal["out", "in", "feature"], text: str) -> None: + """Keep exact lines with monotonic timestamps and log diagnostic traffic.""" + self.transcript.append(FlexTranscriptEntry(time.monotonic(), direction, text)) + logger.debug("Flex %s: %s", direction, text) + + def check_open(self) -> None: + """Require an open reader and an unambiguous session.""" + if self._failure is not None: + raise self._failure + if self._reader is None or self._reader.done(): + raise FlexError("Call setup() before using the Flex") + + def _fail(self, error: FlexError, *, notify: bool = True) -> None: + """Invalidate pending work and notify observers once, without auto-recovery.""" + if self._failure is None: + self._failure = error + if notify: + logger.error("Flex session invalidated: %s", error) + self._on_failure(error) + if self._pending is not None: + self._pending.ack.set() + self._pending.done.set() + + def invalidate(self, error: FlexError) -> None: + """Invalidate interrupted host-side work without stopping physical execution.""" + self._fail(error) + + def _route(self, message: FlexMessage) -> None: + """Route exact opcode matches; unsolicited temperature data never finishes a command.""" + self._on_message(message) + if message.opcode == "ABO" and message.kind == "completion": + if self._abort_done is not None and not self._abort_done.done(): + self._abort_done.set_result(message.code if message.code is not None else -1) + self._fail(FlexError("Instrument aborted; reconnect and inspect physical state", "ABO")) + return + if message.kind == "interrupted": + self._fail(FlexError("Instrument reported QIT; run outcome is uncertain", "QIT")) + return + if self._failure is not None: + return + pending = self._pending + if message.opcode == "TEM" and message.kind == "report": + if pending is not None and pending.opcode == "TEM": + pending.data.append(message.text) + return + if pending is None or message.opcode != pending.opcode: + self._fail(FlexError(f"Unexpected response: {message.text}", message.opcode, message.code)) + return + if message.kind == "ack": + if pending.ack.is_set(): + self._fail(FlexError("Duplicate command acknowledgment", pending.opcode)) + pending.ack.set() + elif message.kind == "completion": + if message.code != 0: + self._fail(FlexError(f"Instrument returned {message.text}", pending.opcode, message.code)) + elif not pending.ack.is_set() or pending.done.is_set(): + self._fail(FlexError("Completion without a unique acknowledgment", pending.opcode)) + pending.done.set() + else: + pending.data.append(message.text) + + async def _read_loop(self) -> None: + """Read with short deadlines so the shared HID worker remains available to abort.""" + lines = FlexLines() + try: + while True: + report = await self.io.read(size=64, timeout=100) + if not report: + await asyncio.sleep(0) + continue + for line in lines.feed(report): + self._record("in", line) + self._route(parse_message(line)) + except asyncio.CancelledError: + raise + except Exception as error: + self._fail(FlexError(f"Flex reader failed: {error}")) + if self._abort_done is not None and not self._abort_done.done(): + self._abort_done.cancel() + + async def setup(self) -> None: + """Open a fresh reader; repeated healthy setup performs no I/O.""" + if self._reader is not None: + self.check_open() + return + await self.io.setup() + self._failure = None + self._pending = None + self.transcript = [] + self._reader = asyncio.create_task(self._read_loop()) + + async def _write(self, payload: bytes, check_session: bool = True) -> None: + """Write all fragments atomically with respect to other commands and abort.""" + async with self._writes: + for report in reports(payload): + if check_session: + self.check_open() + written = await self.io.write(report, report_id=b"\x00") + if written != 65: + raise FlexError(f"Short HID write: {written} of 65 bytes") + + async def command(self, line: str, completion_timeout: Optional[float]) -> tuple[str, ...]: + """Wait for acceptance and then execution; only operator waits may be unbounded.""" + validate_timeout(completion_timeout, "completion_timeout", allow_none=True) + if re.fullmatch(r"[A-Z]{3}(?: [ -~]+)?", line) is None: + raise ValueError("Expected exactly one printable ASCII command") + async with self._commands: + self.check_open() + pending = _Pending(line[:3]) + self._pending = pending + try: + self._record("out", line) + await asyncio.wait_for(self._write(line.encode("ascii") + b"\r\n"), self.timeout) + await asyncio.wait_for(pending.ack.wait(), self.timeout) + self.check_open() + await asyncio.wait_for(pending.done.wait(), completion_timeout) + self.check_open() + return tuple(pending.data) + except BaseException as error: + self._fail(FlexError(f"Command interrupted or failed: {line}: {error}", pending.opcode)) + raise + finally: + self._pending = None + + async def abort(self) -> None: + """Send the observed Feature 01 00 then ESC-only sequence and wait for ABO END. + + Abort bypasses the command lock, including an indefinite PAU wait. Terminal + acknowledgment is not a guarantee of a safe final physical position. + """ + async with self._abort_lock: + if self._reader is None or self._reader.done(): + raise FlexError("Abort requires a live connection") + if self._abort_done is not None: + raise FlexError("Abort already attempted; reconnect before another attempt") + terminal = asyncio.get_running_loop().create_future() + self._abort_done = terminal + self._fail(FlexError("Abort requested; pending operations are uncertain", "ABO")) + try: + async with self._writes: + self._record("feature", "0100") + await asyncio.wait_for( + self.io.send_feature_report(b"\x01\x00", report_id=b"\x00"), self.timeout + ) + self._record("out", "") + await asyncio.wait_for(self._write(b"\x1b", check_session=False), self.timeout) + code = await asyncio.wait_for(asyncio.shield(terminal), self.timeout) + if code != 0: + raise FlexError(f"Abort failed with code {code}", "ABO", code) + finally: + if not terminal.done(): + terminal.cancel() + + async def stop(self) -> None: + """Close transport and wake waiters; this does not stop physical execution.""" + self._fail( + FlexError("Flex connection closed; unfinished operations are uncertain"), + notify=self._pending is not None, + ) + if self._abort_done is not None and not self._abort_done.done(): + self._abort_done.cancel() + reader, self._reader = self._reader, None + try: + if reader is not None: + reader.cancel() + try: + await reader + except asyncio.CancelledError: + pass + finally: + await self.io.stop() + async with self._commands: + pass + self._abort_done = None diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/connection_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/connection_tests.py new file mode 100644 index 00000000000..0c828f989ff --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/connection_tests.py @@ -0,0 +1,37 @@ +"""HID framing and response parsing without hardware access.""" + +import unittest + +from .connection import FlexError, FlexLines, parse_message, reports + + +class FlexFramingTests(unittest.TestCase): + """Validate fragmentation, stale padding and response structure.""" + + def test_arbitrary_fragmentation_coalescing_and_stale_padding(self) -> None: + """A line can span any report boundary; padding can contain unrelated old text.""" + payload = b"VER\r\nVER KingFisherFlex 1.01.0 fixture\r\nVER END 0\r\nTEM 24 22\r\n" + expected = payload.decode().splitlines() + for chunk_size in range(1, 64): + parser = FlexLines() + actual: list[str] = [] + for offset in range(0, len(payload), chunk_size): + chunk = payload[offset : offset + chunk_size] + stale = b"PAU END 0\r\n" * 6 + report = bytes([len(chunk)]) + (chunk + stale)[:63] + actual.extend(parser.feed(report)) + self.assertEqual(actual, expected) + for invalid in (b"", b"\x40" + b"x" * 63, b"\x02x", b"\x01" + b"x" * 64): + with self.assertRaises(FlexError): + FlexLines().feed(invalid) + self.assertEqual(reports(b"\x1b")[0][:3], b"\x01\x1b\x00") + + def test_response_types_and_invalid_lines(self) -> None: + """Keep ACK, telemetry, interruption and terminal error codes distinct.""" + self.assertEqual(parse_message("MOH").kind, "ack") + self.assertEqual(parse_message("TEM 37 22").kind, "report") + self.assertEqual(parse_message("QIT").kind, "interrupted") + self.assertEqual(parse_message("MOM END 13").code, 13) + for line in ("", "MO", "bad"): + with self.assertRaises(FlexError): + parse_message(line) diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/flex.py b/pylabrobot/thermo_fisher/kingfisher/flex/flex.py new file mode 100644 index 00000000000..df57da27d9b --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/flex.py @@ -0,0 +1,485 @@ +"""KingFisher Flex control using local BindIt command scripts.""" + +import asyncio +import logging +import re +from collections.abc import AsyncIterator, Awaitable, Callable +from dataclasses import dataclass +from typing import Literal, Optional, Union + +from pylabrobot.io.hid import HID +from pylabrobot.thermo_fisher.kingfisher.flex.connection import ( + FlexConnection, + FlexError, + FlexMessage, + FlexTranscriptEntry, + validate_timeout, +) +from pylabrobot.thermo_fisher.kingfisher.flex.protocol import ( + FlexCommand, + FlexPrompt, + FlexProtocol, + ProtocolSource, + _command_prompt, + read_flex_bdz, +) + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class FlexEvent: + """An execution event distinguishing instrument waits from host confirmations.""" + + kind: Literal[ + "temperature", + "step", + "load", + "unload", + "pause", + "prompt_completed", + "host_load", + "host_unload", + "host_pause", + "host_prompt_completed", + "run_started", + "run_completed", + "failure", + ] + text: str + prompt: Optional[FlexPrompt] = None + + +@dataclass(frozen=True) +class FlexRunResult: + """Completion counts and the SHA-256 digest of the source BDZ.""" + + protocol_name: str + source_sha256: str + completed_commands: int + host_confirmations: int = 0 + source_kind: Literal["bdz"] = "bdz" + + +class KingFisherFlex: + """Run supported 96DW BindIt exports over USB HID. + + Hardware validation covers firmware 1.01.0 on macOS, including mixing, bead + collection/release, drying, 37 C heating, loops and host confirmations. + Firmware completion does not verify correct tip-comb attachment. + ``setup`` identifies the device; ``initialize`` moves its axes. + Commands execute serially. ``stop`` disconnects; ``abort`` requests a motion stop. + Protocol upload and remote advancement of onboard pauses are unsupported. + + Args: + serial_number: Select one instrument when multiple Flex devices are attached. + timeout: HID write/acknowledgment and abort deadline, in seconds. + operation_timeout: Base motion deadline and margin for timed operations, in seconds. + """ + + _events_closed: bool + _events: asyncio.Queue[Optional[FlexEvent]] + _processing_slot: Optional[int] + _current_command: Optional[FlexCommand] + _current_prompt: Optional[FlexPrompt] + _run_prompts: dict[int, FlexPrompt] + state: Literal["disconnected", "idle", "initializing", "ready", "running", "waiting", "uncertain"] + + def _publish(self, event: FlexEvent) -> None: + """Publish workflow events to one consumer and the module logger.""" + if not self._events_closed: + self._events.put_nowait(event) + logger.info("Flex %s: %s", event.kind, event.text) + + def _close_events(self) -> None: + """Wake an event consumer when the session ends.""" + if not self._events_closed: + self._events_closed = True + self._events.put_nowait(None) + + def _on_failure(self, error: FlexError) -> None: + """Forget inferred geometry and initialization on every uncertain outcome.""" + self._initialized = False + self._processing_slot = None + self.state = "uncertain" + self._session_failed.set() + self._publish(FlexEvent("failure", str(error))) + self._close_events() + + def _prompt_for_command(self, command: FlexCommand, orientation: int) -> FlexPrompt: + """Use validated source metadata while checking the completed physical orientation.""" + prompt = self._run_prompts.get(command.line_number) + if prompt is None: + return _command_prompt(command, orientation) + if prompt.processing_slot != orientation: + raise FlexError("Completed orientation differs from the validated prompt") + return prompt + + def _on_message(self, message: FlexMessage) -> None: + """Publish telemetry and prompt/step events only after command acknowledgment.""" + if self._events_closed: + return + if message.opcode == "TEM" and message.kind == "report": + self._publish(FlexEvent("temperature", message.text)) + command = self._current_command + if message.kind != "ack" or command is None or message.opcode != command.opcode: + return + if command.opcode == "PAU" and self._processing_slot is not None: + prompt = self._prompt_for_command(command, self._processing_slot) + self._current_prompt = prompt + self.state = "waiting" + self._publish(FlexEvent(prompt.action, prompt.label, prompt)) + elif command.opcode == "TXT": + self._publish(FlexEvent("step", command.text.split('"')[1])) + + def __init__( + self, + serial_number: Optional[str] = None, + timeout: float = 30.0, + operation_timeout: float = 120.0, + ) -> None: + """Configure a Flex without enumerating or opening USB devices.""" + validate_timeout(timeout, "timeout") + validate_timeout(operation_timeout, "operation_timeout") + self.io = HID("KingFisher Flex", vid=0x0AB6, pid=0x02C7, serial_number=serial_number) + self.timeout = timeout + self.operation_timeout = operation_timeout + self._connection = FlexConnection(self.io, self._on_message, self._on_failure, timeout) + self._lifecycle = asyncio.Lock() + self._operations = asyncio.Lock() + self._events = asyncio.Queue() + self._session_failed = asyncio.Event() + self._events_closed = False + self._connected = False + self._initialized = False + self._processing_slot = None + self._current_command = None + self._current_prompt = None + self._run_prompts = {} + self.state = "disconnected" + self.version: Optional[str] = None + self.serial: Optional[str] = None + + @property + def transcript(self) -> tuple[FlexTranscriptEntry, ...]: + """Return the session transcript, including messages after interruption.""" + return tuple(self._connection.transcript) + + @property + def processing_slot(self) -> Optional[int]: + """Return the last completed MOR target, or None when physical state is unknown.""" + return self._processing_slot + + async def events(self) -> AsyncIterator[FlexEvent]: + """Consume session events from one task until disconnect or failure. + + Run this iterator alongside ``run_protocol``. A successful run emits + ``run_completed``; the session remains open for telemetry until ``stop``. + """ + queue = self._events + while True: + event = await queue.get() + if event is None: + queue.put_nowait(None) + return + yield event + + async def setup(self) -> None: + """Identify the instrument without EAK, fault acknowledgment or motion commands.""" + async with self._lifecycle: + if self._connected: + self._connection.check_open() + return + logger.warning( + "KingFisher Flex is experimental. Hardware checks cover firmware 1.01.0 on macOS " + "with 96DW mixing/bead methods, 37 C heating and host confirmations. " + "Partial tip pickup remains unresolved. " + "See the Flex validation guide for tested scope and remaining limitations." + ) + self._events = asyncio.Queue() + self._session_failed = asyncio.Event() + self._events_closed = False + self._connection.io = self.io + try: + await self._connection.setup() + response = await self._connection.command("VER", self.timeout) + identity = [line.split() for line in response if line.startswith("VER KingFisherFlex ")] + if len(identity) != 1 or len(identity[0]) != 4: + raise FlexError(f"Expected one KingFisherFlex identification, received {response!r}") + self.version, self.serial = identity[0][2:] + self._connected = True + self.state = "idle" + logger.info("Connected to Flex firmware %s, serial %s", self.version, self.serial) + except BaseException: + await self._connection.stop() + self.state = "disconnected" + self.version = self.serial = None + raise + + def _check_operation(self) -> None: + """Reject concurrent operations rather than queuing an unintended second run.""" + self._connection.check_open() + if self._operations.locked(): + raise FlexError("Another Flex operation is active") + + async def list_protocols(self) -> tuple[str, ...]: + """Read onboard names with REP 4 without initialization or motion. + + Includes maintenance and factory entries. Numeric inventory metadata is + retained in ``transcript`` without interpreting its undocumented fields. + Unexpected formats fail explicitly instead of returning an incomplete list. + """ + self._check_operation() + async with self._operations: + response = await self._connection.command("REP 4", self.timeout) + names = [] + for line in response: + match = re.fullmatch(r'REP 4 [0-9]+ [0-9]+ "((?:[^"\r\n]|"")+)" [0-9]+\s*', line) + if match: + names.append(match[1].replace('""', '"')) + elif re.fullmatch(r"REP 4(?: [0-9]+){5}\s*", line) is None: + raise FlexError(f"Unrecognized protocol inventory response: {line!r}", "REP") + return tuple(names) + + async def initialize(self) -> None: + """Initialize axes and disable heating; this can move axes and present tray 4 for loading. + + Call only after checking the instrument and consumables. Initialization + acknowledges faults, lowers the heater and resets head/magnet positions. + Repeated calls in the same initialized session do not repeat motion. + """ + self._check_operation() + if self._initialized: + return + async with self._operations: + self.state = "initializing" + try: + for command in ("EAK", "TEM 0 300", "MOI 0", "MOS 0", "MOM 0", "MOH 0 100", "MOR 1"): + await self._connection.command(command, self.operation_timeout) + self._processing_slot = 1 + self._initialized = True + self.state = "ready" + except BaseException: + self._initialized = False + self._processing_slot = None + self.state = "uncertain" + raise + + async def _confirm_host_prompt( + self, + command: FlexCommand, + confirm: Callable[[FlexPrompt], Awaitable[bool]], + timeout: Optional[float], + ) -> None: + """Wait for positive host confirmation while monitoring session failure.""" + self._connection.check_open() + if self._processing_slot is None: + raise FlexError("Host confirmation requires a completed carousel orientation") + prompt = self._prompt_for_command(command, self._processing_slot) + self._current_prompt = prompt + self.state = "waiting" + self._publish( + FlexEvent( + "host_pause" + if prompt.action == "pause" + else "host_load" + if prompt.action == "load" + else "host_unload", + prompt.label, + prompt, + ) + ) + + async def confirmation() -> bool: + """Call the operator handler inside a cancellable task.""" + return await confirm(prompt) + + confirmed = asyncio.create_task(confirmation()) + failed = asyncio.create_task(self._session_failed.wait()) + try: + done, _ = await asyncio.wait( + (confirmed, failed), timeout=timeout, return_when=asyncio.FIRST_COMPLETED + ) + self._connection.check_open() + if self._session_failed.is_set(): + raise FlexError("Session ended while awaiting host confirmation") + if confirmed not in done: + raise asyncio.TimeoutError("Host confirmation timed out") + if await confirmed is not True: + raise FlexError("Host did not confirm the requested action") + self._publish(FlexEvent("host_prompt_completed", prompt.label, prompt)) + self._current_prompt = None + self.state = "running" + finally: + for task in (confirmed, failed): + if not task.done(): + task.cancel() + await asyncio.gather(confirmed, failed, return_exceptions=True) + + async def run_protocol( + self, + protocol: Union[FlexProtocol, ProtocolSource], + *, + operator_timeout: Optional[float] = None, + confirm_prompt: Optional[Callable[[FlexPrompt], Awaitable[bool]]] = None, + confirm_pause: Optional[Callable[[FlexPrompt], Awaitable[bool]]] = None, + ) -> FlexRunResult: + """Stream a supported BindIt export and await terminal completion. + + Preflight finishes before any I/O. Initialization must have been requested + separately. By default PAU waits for physical Start. Supplying ``confirm_prompt`` + opts into experimental host confirmation: PAU 1/2 are NOT sent, and the async + callback must return exactly True after each load/unload before streaming continues. + Standalone PAU 0 requires a separate ``confirm_pause`` callback to replace its + instrument wait. Otherwise PAU 0 is sent even when plate confirmations use PLR. + Its message is descriptive text, never a timer. Host pause handling emits + host_pause followed by host_prompt_completed. + This does not release an instrument PAU already in progress. Host waits emit + host_load/host_unload and host_prompt_completed, never a fictitious PAU completion. + Dispensing pauses include required per-well additions in both the prompt label + and ``prompt.additions``. Confirmation means those additions are complete. + Heating commands set a target without waiting for it to be reached; temperature + reports are telemetry, not command completion. Documented setpoints are supported; + hardware validation covers 37 C. + ``operator_timeout=None`` waits indefinitely while the + reader continues to detect disconnects and errors. Cancellation does not + abort hardware: it invalidates the session and requires explicit recovery. + """ + validate_timeout(operator_timeout, "operator_timeout", allow_none=True) + if confirm_prompt is not None and not callable(confirm_prompt): + raise TypeError("confirm_prompt must be an async callable") + if confirm_pause is not None and not callable(confirm_pause): + raise TypeError("confirm_pause must be an async callable") + parsed = protocol if isinstance(protocol, FlexProtocol) else read_flex_bdz(protocol) + commands = parsed.preflight() + prompts = {p.line_number: p for p in parsed.prompts()} + self._check_operation() + if not self._initialized: + raise FlexError("Call initialize() explicitly before running a protocol") + async with self._operations: + self._run_prompts = prompts + self.state = "running" + self._publish(FlexEvent("run_started", parsed.name)) + timer_duration = 0.0 + completed = 0 + host_confirmations = 0 + try: + # BindIt repeats this preamble at run start, separately from connection startup. + for line in ("EAK", "MOI 0", "MOS 0", "MOM 0", "MOH 0 100", "TEM 0 300"): + await self._connection.command(line, self.operation_timeout) + for command in commands: + self._current_command = command + if command.opcode == "PAU": + handler = confirm_pause if command.text.startswith("PAU 0 ") else confirm_prompt + if handler is not None: + await self._confirm_host_prompt(command, handler, operator_timeout) + host_confirmations += 1 + continue + fields = command.text.split() + deadline: Optional[float] = self.operation_timeout + if command.opcode == "PAU": + deadline = operator_timeout + elif command.opcode == "MOH" and len(fields) == 6: + deadline = int(fields[4]) / 100 + self.operation_timeout + elif command.opcode == "WAI": + if len(fields) == 3: + timer_duration = int(fields[2]) / 100 + else: + deadline = timer_duration + self.operation_timeout + if command.opcode == "MOR": + self._processing_slot = None + await self._connection.command(command.text, deadline) + completed += 1 + if command.opcode == "MOR": + self._processing_slot = int(fields[1]) + if command.opcode == "PAU": + self._publish(FlexEvent("prompt_completed", command.text, self._current_prompt)) + self._current_prompt = None + self.state = "running" + self._current_command = None + for line in ('TXT "END_OF_RUN"', "TEM 0 300", "TEM 0 300"): + await self._connection.command(line, self.operation_timeout) + self._connection.check_open() + self.state = "ready" + self._publish(FlexEvent("run_completed", parsed.name)) + return FlexRunResult(parsed.name, parsed.sha256, completed, host_confirmations) + except BaseException as error: + self._connection.invalidate(FlexError(f"Protocol interrupted or failed: {error}")) + self._initialized = False + self._processing_slot = None + self.state = "uncertain" + raise + finally: + self._current_command = None + self._current_prompt = None + self._run_prompts = {} + + async def resume(self) -> None: + """Refuse to invent a software equivalent of the physical Start button.""" + raise NotImplementedError("Flex remote Start is unverified; press physical Start at the prompt") + + async def start_stored_protocol( + self, name: str, *, completion_timeout: Optional[float] = None + ) -> None: + """Start an onboard sequence using BGN and wait for its terminal response. + + Experimental, based on the Flex Service Manual v1.1, section 7.1.11.2; + validated through the first loading prompt only. Motion may begin immediately. The exact name + must occur once in a fresh onboard inventory and is sent without truncation. + Name length acceptance is determined by the instrument firmware. + + Internal commands and loading pauses are not reported individually by BGN. + Host prompt callbacks therefore cannot advance them. Acknowledgment alone + is not completion. Timeout or cancellation invalidates the session but does + not stop motion; explicitly call ``abort`` from a supervising task. + + Args: + name: Exact stored name (nonblank printable ASCII). + completion_timeout: Seconds to wait for BGN END, or None for operator waits. + """ + if not isinstance(name, str) or re.fullmatch(r"[ -~]+", name) is None or not name.strip(): + raise ValueError("Stored protocol name must be nonblank printable ASCII") + validate_timeout(completion_timeout, "completion_timeout", allow_none=True) + names = await self.list_protocols() + if names.count(name) != 1: + raise FlexError("Stored protocol name must occur exactly once in the onboard inventory") + self._check_operation() + async with self._operations: + self._initialized = False + self._processing_slot = None + self.state = "running" + logger.warning("Experimental onboard launch: %s; motion may begin immediately", name) + self._publish(FlexEvent("run_started", name)) + try: + quoted_name = name.replace('"', '""') + await self._connection.command(f'BGN "{quoted_name}"', completion_timeout) + self.state = "idle" + self._publish(FlexEvent("run_completed", name)) + except BaseException: + self.state = "uncertain" + raise + + async def abort(self) -> None: + """Request PC abort without waiting for the active command or operator pause.""" + await self._connection.abort() + + async def stop(self) -> None: + """Disconnect and invalidate position; never claim that disconnect stops motion.""" + async with self._lifecycle: + if not self._connected and self.state == "disconnected": + self._close_events() + return + self._session_failed.set() + try: + await self._connection.stop() + async with self._operations: + pass + finally: + self._connected = False + self._initialized = False + self._processing_slot = None + self._current_command = None + self._current_prompt = None + self.version = self.serial = None + self.state = "disconnected" + self._close_events() diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/flex_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/flex_tests.py new file mode 100644 index 00000000000..3f1d90bdb2e --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/flex_tests.py @@ -0,0 +1,365 @@ +"""Driver lifecycle and fault handling against controlled HID responses.""" + +import asyncio +import unittest +from dataclasses import replace +from typing import Optional + +from pylabrobot.io.hid import HID +from pylabrobot.thermo_fisher import KingFisherFlex +from pylabrobot.thermo_fisher.kingfisher.flex import FlexError, FlexProtocolError, read_flex_bdz +from pylabrobot.thermo_fisher.kingfisher.flex.connection import FlexLines, reports + +from .test_helpers import bdz_bytes + + +class ScriptedHID(HID): + """Provide deterministic reports without importing a hardware connection.""" + + def __init__(self) -> None: + """Initialize counters and configurable fault/delay injection.""" + super().__init__("Flex fixture", 0x0AB6, 0x02C7) + self.incoming: asyncio.Queue[bytes | Exception] = asyncio.Queue() + self.commands: list[str] = [] + self.features: list[bytes] = [] + self.writes: list[bytes] = [] + self.opened = 0 + self.closed = 0 + self.lines = FlexLines() + self.responses: dict[str, Optional[str]] = {} + self.short_write = False + self.auto_abort = True + self.sent: asyncio.Queue[str] = asyncio.Queue() + + def emit(self, text: str) -> None: + """Deliver CRLF lines in real 64-byte report framing.""" + for report in reports(text.encode("ascii")): + self.incoming.put_nowait(report) + + async def setup(self) -> None: + """Open only the fixture session.""" + self.opened += 1 + self.incoming = asyncio.Queue() + self.lines = FlexLines() + + async def stop(self) -> None: + """Record transport closure without sending a hardware stop command.""" + self.closed += 1 + + async def read(self, size: int, timeout: int) -> bytes: + """Wait for data or an injected disconnection.""" + result = await self.incoming.get() + if isinstance(result, Exception): + raise result + return result + + async def write(self, data: bytes, report_id: bytes = b"\x00") -> int: + """Capture report IDs and respond to fully reassembled commands.""" + self.writes.append(report_id + data) + if self.short_write: + return 1 + if data[1 : 1 + data[0]] == b"\x1b": + if self.auto_abort: + self.emit("ABO END 0\r\n") + return 65 + for line in self.lines.feed(data): + self.commands.append(line) + self.sent.put_nowait(line) + response: Optional[str] + if line in self.responses: + response = self.responses[line] + elif line == "VER": + response = "VER\r\nVER KingFisherFlex 1.01.0 fixture\r\nVER END 0\r\n" + else: + response = f"{line[:3]}\r\n{line[:3]} END 0\r\n" + if response is not None: + self.emit(response) + return 65 + + async def send_feature_report(self, data: bytes, report_id: bytes = b"\x00") -> int: + """Record the HID API report-ID prefix separately from on-wire payload.""" + self.features.append(report_id + data) + return len(data) + len(report_id) + + async def wait_sent(self, expected: str) -> None: + """Wait for a particular transmitted line without sleep-based synchronization.""" + while await asyncio.wait_for(self.sent.get(), 1) != expected: + pass + + +class FlexTests(unittest.IsolatedAsyncioTestCase): + """Exercise public APIs with no USB access.""" + + async def asyncSetUp(self) -> None: + """Prepare a Flex and immutable protocol fixture without opening either.""" + self.io = ScriptedHID() + self.flex = KingFisherFlex(timeout=0.2, operation_timeout=0.2) + self.flex.io = self.io + self.protocol = read_flex_bdz(bdz_bytes()) + + async def asyncTearDown(self) -> None: + """Always drain connection tasks.""" + await self.flex.stop() + + async def ready(self) -> None: + """Explicitly initialize the fixture for motion tests.""" + await self.flex.setup() + await self.flex.initialize() + + async def test_identification_is_idempotent_and_motion_free(self) -> None: + """Setup sends VER only, and initialization must be requested separately.""" + await self.flex.setup() + await self.flex.setup() + self.assertEqual(self.io.commands, ["VER"]) + self.assertEqual(self.io.opened, 1) + self.assertEqual(self.flex.serial, "fixture") + self.assertIsNone(self.flex.processing_slot) + with self.assertRaisesRegex(FlexError, "initialize"): + await self.flex.run_protocol(self.protocol) + self.assertEqual(self.io.commands, ["VER"]) + + async def test_wrong_device_closes_connection(self) -> None: + """A matching VID/PID alone cannot establish instrument identity.""" + self.io.responses["VER"] = "VER\r\nVER KingFisherPresto 1.01.0 fixture\r\nVER END 0\r\n" + with self.assertRaises(FlexError): + await self.flex.setup() + self.assertEqual(self.io.closed, 1) + self.assertEqual(self.flex.state, "disconnected") + + async def test_complete_protocol_preserves_command_order(self) -> None: + """Preserve script commands between initialization and successful finalization.""" + await self.ready() + await self.flex.initialize() + before = len(self.io.commands) + result = await self.flex.run_protocol(self.protocol) + self.assertEqual(result.completed_commands, 3) + self.assertEqual(result.source_sha256, self.protocol.sha256) + self.assertEqual( + self.io.commands[before:], + [ + "EAK", + "MOI 0", + "MOS 0", + "MOM 0", + "MOH 0 100", + "TEM 0 300", + "MOR 6", + 'PAU 1 "plate"', + 'PAU 2 "plate"', + 'TXT "END_OF_RUN"', + "TEM 0 300", + "TEM 0 300", + ], + ) + self.assertEqual(self.flex.state, "ready") + await self.flex.stop() + events = [e async for e in self.flex.events()] + self.assertEqual(sum(e.kind == "prompt_completed" for e in events), 2) + self.assertEqual(sum(e.kind == "run_completed" for e in events), 1) + self.assertFalse(any(e.kind == "failure" for e in events)) + + async def test_entire_script_preflight_precedes_transmission(self) -> None: + """An unsupported final line prevents even the execution preamble.""" + await self.ready() + before = list(self.io.commands) + with self.assertRaises(FlexProtocolError): + await self.flex.run_protocol(replace(self.protocol, script=self.protocol.script + "REC 0\n")) + self.assertEqual(self.io.commands, before) + + async def test_pause_waits_for_instrument_completion(self) -> None: + """Temperature reports and a software resume call cannot complete PAU.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\nTEM 24 22\r\n" + protocol = replace(self.protocol, script='MOR 7\nPAU 1 "plate"\nTXT "done"\n') + task = asyncio.create_task(self.flex.run_protocol(protocol)) + await self.io.wait_sent('PAU 1 "plate"') + async for event in self.flex.events(): + if event.kind == "load": + self.assertIsNotNone(event.prompt) + if event.prompt is not None: + self.assertEqual(event.prompt.loading_slot, 2) + break + self.assertFalse(task.done()) + self.assertEqual(self.flex.state, "waiting") + with self.assertRaises(NotImplementedError): + await self.flex.resume() + self.assertNotIn('TXT "done"', self.io.commands) + self.io.emit("PAU END 0\r\n") + await task + self.assertIn('TXT "done"', self.io.commands) + + async def test_long_operation_uses_duration_not_ack_timeout(self) -> None: + """A timed mix survives past both the base motion and acceptance deadlines.""" + self.flex = KingFisherFlex(timeout=0.01, operation_timeout=0.01) + self.flex.io = self.io + await self.ready() + self.io.responses["MOH 547 669 0 60000 0"] = "MOH\r\nTEM 24 22\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script="MOH 547 669 0 60000 0\n")) + ) + await self.io.wait_sent("MOH 547 669 0 60000 0") + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(asyncio.shield(task), 0.04) + self.io.emit("MOH END 0\r\n") + await task + + async def test_abort_interrupts_loading_or_motion_without_replay(self) -> None: + """Abort bypasses the blocked command lock and preserves interrupted state.""" + for command in ('PAU 1 "plate"', "MOH 650 80"): + with self.subTest(command=command): + await self.ready() + self.io.responses[command] = command[:3] + "\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script=f"MOR 1\n{command}\n")) + ) + await self.io.wait_sent(command) + await self.flex.abort() + with self.assertRaises(FlexError): + await task + self.assertEqual(self.io.features[-1], b"\x00\x01\x00") + self.assertEqual(self.io.writes[-1], b"\x00\x01\x1b" + b"\x00" * 62) + self.assertIsNone(self.flex.processing_slot) + before = len(self.io.commands) + with self.assertRaises(FlexError): + await self.flex.run_protocol(self.protocol) + self.assertEqual(len(self.io.commands), before) + await self.flex.stop() + self.io.responses.clear() + + async def test_cancellation_requires_fresh_session(self) -> None: + """Late completion cannot satisfy a later command with the same opcode.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script='MOR 1\nPAU 1 "plate"\n')) + ) + await self.io.wait_sent('PAU 1 "plate"') + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.io.emit("PAU END 0\r\n") + with self.assertRaises(FlexError): + await self.flex.setup() + self.assertEqual(self.io.features, []) + await self.flex.stop() + await self.flex.setup() + self.assertEqual(self.flex.state, "idle") + self.assertEqual(self.io.commands[-1], "VER") + + async def test_timeout_does_not_resend(self) -> None: + """A missing acknowledgment invalidates the session without retry.""" + self.io.responses["VER"] = None + with self.assertRaises(asyncio.TimeoutError): + await self.flex.setup() + self.assertEqual(self.io.commands, ["VER"]) + + async def test_instrument_error_preserves_unknown_code(self) -> None: + """Nonzero END prevents subsequent commands and remains distinguishable.""" + await self.ready() + self.io.responses["MOR 7"] = "MOR\r\nMOR END 9876\r\n" + with self.assertRaises(FlexError) as raised: + await self.flex.run_protocol(replace(self.protocol, script='MOR 7\nPAU 1 "plate"\n')) + self.assertEqual(raised.exception.code, 9876) + self.assertNotIn('PAU 1 "plate"', self.io.commands) + + async def test_disconnect_wakes_operator_wait_and_event_consumer(self) -> None: + """Indefinite human waits still terminate on transport failure.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script='MOR 1\nPAU 1 "plate"\n')) + ) + await self.io.wait_sent('PAU 1 "plate"') + self.io.incoming.put_nowait(OSError("unplugged")) + with self.assertRaisesRegex(FlexError, "unplugged"): + await asyncio.wait_for(task, 1) + self.assertIn("failure", [e.kind async for e in self.flex.events()]) + + async def test_stop_during_pause_does_not_claim_abort(self) -> None: + """Connection teardown completes even while a run owns the operation lock.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script='MOR 1\nPAU 1 "plate"\n')) + ) + await self.io.wait_sent('PAU 1 "plate"') + await asyncio.wait_for(self.flex.stop(), 1) + with self.assertRaises(FlexError): + await task + self.assertEqual(self.flex.state, "disconnected") + self.assertFalse(self.io.features) + + async def test_unsolicited_quit_prevents_success(self) -> None: + """Unsolicited QIT cannot be interpreted as successful run completion.""" + await self.ready() + self.io.responses["MOR 7"] = "MOR\r\nQIT\r\n" + with self.assertRaises(FlexError): + await self.flex.run_protocol(replace(self.protocol, script="MOR 7\n")) + self.assertEqual(self.flex.state, "uncertain") + + async def test_short_write_and_wrong_completion_fail(self) -> None: + """A partial transfer or mismatched opcode never counts as success.""" + self.io.short_write = True + with self.assertRaises(FlexError): + await self.flex.setup() + self.io.short_write = False + self.io.responses["VER"] = "VER\r\nMOR END 0\r\n" + with self.assertRaises(FlexError): + await self.flex.setup() + + async def test_unknown_remote_commands_send_nothing(self) -> None: + """Unavailable remote resume fails before any hardware access.""" + with self.assertRaises(NotImplementedError): + await self.flex.resume() + self.assertEqual(self.io.opened, 0) + self.assertEqual(self.io.writes, []) + + async def test_concurrent_run_is_rejected_without_queueing(self) -> None: + """A second run cannot start later merely because the first was waiting.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\n" + task = asyncio.create_task( + self.flex.run_protocol(replace(self.protocol, script='MOR 1\nPAU 1 "plate"\n')) + ) + await self.io.wait_sent('PAU 1 "plate"') + before = list(self.io.commands) + with self.assertRaisesRegex(FlexError, "active"): + await self.flex.run_protocol(self.protocol) + with self.assertRaisesRegex(FlexError, "active"): + await self.flex.initialize() + self.assertEqual(self.io.commands, before) + self.io.emit("PAU END 0\r\n") + await task + + async def test_abort_requires_terminal_response(self) -> None: + """Sending abort bytes is not enough to claim the instrument acknowledged abort.""" + await self.ready() + self.io.auto_abort = False + task = asyncio.create_task(self.flex.abort()) + with self.assertRaises(asyncio.TimeoutError): + await task + self.assertEqual(self.flex.state, "uncertain") + self.assertIn("", [entry.text for entry in self.flex.transcript]) + + async def test_finite_operator_deadline_never_advances(self) -> None: + """A timed-out human wait invalidates the run rather than skipping its prompt.""" + await self.ready() + self.io.responses['PAU 1 "plate"'] = "PAU\r\n" + with self.assertRaises(asyncio.TimeoutError): + await self.flex.run_protocol( + replace(self.protocol, script='MOR 1\nPAU 1 "plate"\nTXT "next"\n'), + operator_timeout=0.01, + ) + self.assertNotIn('TXT "next"', self.io.commands) + self.assertFalse(self.io.features) + + async def test_duplicate_ack_invalidates_session(self) -> None: + """Ambiguous same-opcode replies cannot be accepted as a new transaction.""" + await self.ready() + self.io.responses["MOR 7"] = "MOR\r\nMOR\r\nMOR END 0\r\n" + with self.assertRaisesRegex(FlexError, "Duplicate"): + await self.flex.run_protocol(replace(self.protocol, script="MOR 7\n")) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/host_confirmation_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/host_confirmation_tests.py new file mode 100644 index 00000000000..c1d02015543 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/host_confirmation_tests.py @@ -0,0 +1,180 @@ +"""Host-managed loading waits with simulated HID only; no instrument access.""" + +import asyncio +import unittest +from dataclasses import replace + +from .connection import FlexError +from .flex import KingFisherFlex +from .flex_tests import ScriptedHID +from .protocol import FlexPrompt, read_flex_bdz +from .test_helpers import bdz_bytes + + +class HostConfirmationTests(unittest.IsolatedAsyncioTestCase): + """Verify confirmations gate every subsequent command and survive no session failure.""" + + async def asyncSetUp(self) -> None: + """Initialize an entirely simulated instrument and a two-prompt script.""" + self.io = ScriptedHID() + self.flex = KingFisherFlex(timeout=0.2, operation_timeout=0.2) + self.flex.io = self.io + self.original = read_flex_bdz(bdz_bytes()) + self.protocol = replace( + self.original, script='MOR 7\nPAU 1 "tips"\nMOR 6\nPAU 1 "sample"\nTXT "next"\n' + ) + self.entered: asyncio.Queue[FlexPrompt] = asyncio.Queue() + self.release = asyncio.Event() + self.cancelled = asyncio.Event() + await self.flex.setup() + await self.flex.initialize() + + async def asyncTearDown(self) -> None: + """Close simulation and wake pending operations.""" + await self.flex.stop() + + async def confirm(self, prompt: FlexPrompt) -> bool: + """Wait for this prompt's explicit confirmation and track cancellation.""" + self.entered.put_nowait(prompt) + try: + await self.release.wait() + self.release.clear() + return True + finally: + self.cancelled.set() + + async def test_full_original_preserves_all_non_pause_commands(self) -> None: + """Host mode changes only PAU execution and reports its confirmations separately.""" + seen = [] + + async def confirm(prompt: FlexPrompt) -> bool: + """Record a simulated operator confirmation.""" + seen.append(prompt) + return True + + before = len(self.io.commands) + result = await self.flex.run_protocol(self.original, confirm_prompt=confirm) + commands = [c.text for c in self.original.preflight() if c.opcode != "PAU"] + expected = ["EAK", "MOI 0", "MOS 0", "MOM 0", "MOH 0 100", "TEM 0 300"] + expected += commands + ['TXT "END_OF_RUN"', "TEM 0 300", "TEM 0 300"] + self.assertEqual(self.io.commands[before:], expected) + self.assertEqual(tuple(seen), self.original.prompts()) + self.assertEqual(result.completed_commands, len(commands)) + self.assertEqual(result.host_confirmations, len(seen)) + await self.flex.stop() + events = [e async for e in self.flex.events()] + self.assertEqual(sum(e.kind == "host_prompt_completed" for e in events), len(seen)) + self.assertFalse(any(e.kind in ("load", "unload", "prompt_completed") for e in events)) + self.assertFalse(any(e.text == "PAU END 0" for e in self.flex.transcript)) + + async def test_each_confirmation_gates_motion_and_reader_stays_live(self) -> None: + """A confirmation releases only its own prompt; telemetry never releases either.""" + task = asyncio.create_task(self.flex.run_protocol(self.protocol, confirm_prompt=self.confirm)) + try: + first = await asyncio.wait_for(self.entered.get(), 1) + self.assertEqual((first.processing_slot, first.loading_slot), (7, 2)) + self.assertEqual(self.flex.state, "waiting") + self.assertNotIn("MOR 6", self.io.commands) + self.io.emit("TEM 24 23\r\n") + async for event in self.flex.events(): + if event.kind == "temperature": + break + self.assertFalse(task.done()) + with self.assertRaisesRegex(FlexError, "active"): + await self.flex.initialize() + self.release.set() + second = await asyncio.wait_for(self.entered.get(), 1) + self.assertEqual((second.processing_slot, second.loading_slot), (6, 1)) + self.assertNotIn('TXT "next"', self.io.commands) + self.assertFalse(any(c.startswith("PAU") for c in self.io.commands)) + self.release.set() + result = await asyncio.wait_for(task, 1) + self.assertEqual(result.host_confirmations, 2) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_rejected_confirmation_invalidates_without_motion(self) -> None: + """A negative answer prevents every subsequent command and requires reconnect.""" + + async def reject(prompt: FlexPrompt) -> bool: + """Reject the requested plate action.""" + return False + + with self.assertRaisesRegex(FlexError, "did not confirm"): + await self.flex.run_protocol(self.protocol, confirm_prompt=reject) + self.assertEqual(self.io.commands[-1], "MOR 7") + self.assertEqual(self.flex.state, "uncertain") + with self.assertRaises(FlexError): + await self.flex.initialize() + self.assertFalse(self.io.features) + + async def test_callback_error_invalidates_without_motion(self) -> None: + """Input/robot failures cannot be treated as a completed plate action.""" + + async def fail(prompt: FlexPrompt) -> bool: + """Simulate an operator interface failure.""" + raise EOFError("operator input closed") + + with self.assertRaises(EOFError): + await self.flex.run_protocol(self.protocol, confirm_prompt=fail) + self.assertEqual(self.io.commands[-1], "MOR 7") + self.assertEqual(self.flex.state, "uncertain") + + async def test_timeout_cancels_callback_without_advancing(self) -> None: + """The timeout aborts the software run rather than confirming a plate.""" + with self.assertRaises(asyncio.TimeoutError): + await self.flex.run_protocol( + self.protocol, confirm_prompt=self.confirm, operator_timeout=0.01 + ) + self.assertTrue(self.cancelled.is_set()) + self.assertEqual(self.io.commands[-1], "MOR 7") + self.assertFalse(self.io.features) + + async def test_session_failure_wakes_host_wait_and_rejects_late_confirmation(self) -> None: + """Abort, stop, disconnect, and cancellation release waits without replay.""" + for failure in ("abort", "stop", "disconnect", "cancel"): + with self.subTest(failure=failure): + await self.flex.stop() + await self.flex.setup() + await self.flex.initialize() + self.release.clear() + self.cancelled.clear() + before = len(self.io.commands) + task = asyncio.create_task( + self.flex.run_protocol(self.protocol, confirm_prompt=self.confirm) + ) + await asyncio.wait_for(self.entered.get(), 1) + if failure == "abort": + await self.flex.abort() + elif failure == "stop": + await asyncio.wait_for(self.flex.stop(), 1) + elif failure == "disconnect": + self.io.incoming.put_nowait(OSError("unplugged")) + else: + task.cancel() + with self.assertRaises((FlexError, asyncio.CancelledError)): + await asyncio.wait_for(task, 1) + self.release.set() + self.assertTrue(self.cancelled.is_set()) + self.assertNotIn("MOR 6", self.io.commands[before:]) + self.assertFalse(any(c.startswith("PAU") for c in self.io.commands[before:])) + + async def test_confirmation_only_after_rotation_completion(self) -> None: + """An acknowledged but unfinished rotation must not expose a load confirmation.""" + self.io.responses["MOR 7"] = "MOR\r\n" + task = asyncio.create_task(self.flex.run_protocol(self.protocol, confirm_prompt=self.confirm)) + try: + await self.io.wait_sent("MOR 7") + self.assertTrue(self.entered.empty()) + self.io.emit("MOR END 0\r\n") + await asyncio.wait_for(self.entered.get(), 1) + self.assertEqual(self.flex.processing_slot, 7) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/pause_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/pause_tests.py new file mode 100644 index 00000000000..75b14c231c1 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/pause_tests.py @@ -0,0 +1,143 @@ +"""Operator pauses, reagent additions and timed waits with simulated HID.""" + +import asyncio +import unittest +from dataclasses import replace + +from .connection import FlexError +from .flex import KingFisherFlex +from .flex_tests import ScriptedHID +from .protocol import FlexPrompt, FlexProtocolError, read_flex_bdz +from .test_helpers import bdz_bytes + + +class PauseTests(unittest.IsolatedAsyncioTestCase): + """Keep manual confirmations distinct from firmware timers.""" + + async def asyncSetUp(self) -> None: + """Prepare one standalone addition pause and a simulated session.""" + self.io = ScriptedHID() + self.flex = KingFisherFlex(timeout=0.2, operation_timeout=0.2) + self.flex.io = self.io + self.protocol = read_flex_bdz( + bdz_bytes( + 'MOR 6\nPAU 1 "plate"\nTXT 11 "addition"\nMOR 6\nPAU 0 "add liquid"\n' + 'TXT "addition"\nMOR 1\nMOR 6\nPAU 2 "plate"\n', + steps='add liquid' + '' + '' + "", + ) + ) + await self.flex.setup() + await self.flex.initialize() + + async def asyncTearDown(self) -> None: + """Close the simulated reader.""" + await self.flex.stop() + + async def confirm_plate(self, prompt: FlexPrompt) -> bool: + """Accept only loading/unloading actions.""" + self.assertIn(prompt.action, ("load", "unload")) + return True + + async def test_plate_handler_does_not_release_manual_pause(self) -> None: + """PAU 0 needs its own handler or an instrument completion.""" + self.io.responses['PAU 0 "add liquid"'] = "PAU\r\nTEM 37 22\r\n" + task = asyncio.create_task( + self.flex.run_protocol(self.protocol, confirm_prompt=self.confirm_plate) + ) + try: + await self.io.wait_sent('PAU 0 "add liquid"') + self.assertFalse(task.done()) + self.assertNotIn('TXT "addition"', self.io.commands) + self.io.emit("PAU END 0\r\n") + result = await asyncio.wait_for(task, 1) + self.assertEqual(result.host_confirmations, 2) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_addition_metadata_and_host_gate(self) -> None: + """Expose reagent details and hold all following motion until confirmed.""" + entered: asyncio.Queue[FlexPrompt] = asyncio.Queue() + release = asyncio.Event() + + async def confirm(prompt: FlexPrompt) -> bool: + """Hold a single addition confirmation.""" + entered.put_nowait(prompt) + await release.wait() + return True + + task = asyncio.create_task( + self.flex.run_protocol( + self.protocol, + confirm_prompt=self.confirm_plate, + confirm_pause=confirm, + ) + ) + try: + prompt = await asyncio.wait_for(entered.get(), 1) + self.assertEqual(prompt.loading_slot, 1) + self.assertEqual(prompt.additions[0].volume, 50) + self.assertEqual(prompt.additions[0].reagent_name, "water") + self.assertIn("50 uL", prompt.label) + self.assertNotIn('TXT "addition"', self.io.commands) + self.assertFalse(any(c.startswith("PAU") for c in self.io.commands)) + release.set() + result = await asyncio.wait_for(task, 1) + self.assertEqual(result.host_confirmations, 3) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + def test_inconsistent_dispense_metadata_rejected(self) -> None: + """Reject missing reagents and incorrect plate presentation before transmission.""" + for xml in ( + self.protocol.exported_xml.replace(b'volume="50"', b'volume="nan"'), + self.protocol.exported_xml.replace(b'', b""), + self.protocol.exported_xml.replace(b'wellGroup="Plate"', b'wellGroup="missing"'), + ): + with self.subTest(xml=xml), self.assertRaises(FlexProtocolError): + replace(self.protocol, exported_xml=xml).preflight() + with self.assertRaises(FlexProtocolError): + replace( + self.protocol, + script=self.protocol.script.replace('TXT 11 "addition"\nMOR 6', 'TXT 11 "addition"\nMOR 7'), + ).preflight() + + async def test_two_timed_cycles_wait_for_each_completion(self) -> None: + """Already-expanded loops run once; every WAI completion gates subsequent commands.""" + cycle = "MOH 630 200 0 3000 0\nWAI 1 3000\nWAI 1\n" + protocol = read_flex_bdz(bdz_bytes(cycle * 2 + 'TXT "finished"\n')) + self.io.responses["WAI 1"] = "WAI\r\n" + task = asyncio.create_task(self.flex.run_protocol(protocol)) + try: + for count in (1, 2): + await self.io.wait_sent("WAI 1") + self.assertEqual(self.io.commands.count("MOH 630 200 0 3000 0"), count) + self.assertNotIn('TXT "finished"', self.io.commands) + self.io.emit("WAI END 0\r\n") + result = await asyncio.wait_for(task, 1) + self.assertEqual(result.completed_commands, 7) + self.assertEqual(result.host_confirmations, 0) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_heater_failure_blocks_following_motion(self) -> None: + """A heater error retains its raw code and prevents the rest of the method.""" + protocol = read_flex_bdz(bdz_bytes("TEM 37 100\nMOI 310\nMOH 630 200 0 3000 0\n")) + self.io.responses["MOI 310"] = "MOI\r\nMOI END 9\r\n" + with self.assertRaises(FlexError) as error: + await self.flex.run_protocol(protocol) + self.assertEqual(error.exception.code, 9) + self.assertNotIn("MOH 630 200 0 3000 0", self.io.commands) + self.assertEqual(self.flex.state, "uncertain") + + async def test_temperature_telemetry_does_not_wait_for_target(self) -> None: + """TEM END permits the next command even if telemetry is below the target.""" + protocol = read_flex_bdz(bdz_bytes("TEM 37 100\nMOI 310\n")) + self.io.responses["TEM 37 100"] = "TEM\r\nTEM 29 22\r\nTEM END 0\r\n" + await self.flex.run_protocol(protocol) + self.assertIn("MOI 310", self.io.commands) diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/protocol.py b/pylabrobot/thermo_fisher/kingfisher/flex/protocol.py new file mode 100644 index 00000000000..db0bbb57d7d --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/protocol.py @@ -0,0 +1,640 @@ +"""Lossless Flex BindIt inspection and script compatibility validation.""" + +import hashlib +import math +import re +import struct +import xml.etree.ElementTree as ET +import zlib +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Literal, Optional, Union + +FLEX_TYPE_ID = "844c2ce4-e8cf-4c19-b8e6-c43037658b8c" +MAX_XML_SIZE = 16 * 1024 * 1024 +ProtocolSource = Union[str, Path, bytes] + + +class FlexProtocolError(ValueError): + """Invalid container or unsupported script behavior.""" + + +@dataclass(frozen=True) +class FlexProtocolIssue: + """An execution blocker found offline, with a stable category and source location. + + Absence of issues means the driver accepts the export, not that this particular + protocol has passed a hardware test. Line numbers refer to the original script, + including metadata lines, rather than the filtered streaming command sequence. + """ + + code: str + location: str + message: str + line_number: Optional[int] = None + + +@dataclass(frozen=True) +class FlexCommand: + """An original script line and its one-based source line number.""" + + line_number: int + text: str + + @property + def opcode(self) -> str: + """Return the command's wire opcode.""" + return self.text.split(" ", 1)[0] + + +@dataclass(frozen=True) +class FlexPlate: + """A layout plate; its name is not a physical carousel slot.""" + + id: str + name: str + type_id: str + + +@dataclass(frozen=True) +class FlexTip: + """Tip configuration with all ordered pickup/return plate references.""" + + name: str + type_id: str + plate_ids: tuple[str, ...] + + +@dataclass(frozen=True) +class FlexDispense: + """Required reagent addition to every well of a plate; volume is per well in uL.""" + + plate_id: str + plate_name: str + reagent_name: str + volume: float + + +@dataclass(frozen=True) +class FlexPrompt: + """A plate request or protocol pause, with orientation from preceding MOR. + + For a pause, ``loading_slot`` is the accessible tray, not an unload instruction. + """ + + line_number: int + action: Literal["load", "unload", "pause"] + label: str + processing_slot: int + loading_slot: int + additions: tuple[FlexDispense, ...] = () + + +def loading_slot(processing_slot: int) -> int: + """Convert a completed MOR target to the printed tray at the loading station.""" + if not 1 <= processing_slot <= 8: + raise ValueError("Flex processing slot must be in 1..8") + return (processing_slot - 1 + 3) % 8 + 1 + + +def _xml(payload: bytes, root_name: str) -> ET.Element: + """Parse bounded XML without declarations that can expand external data.""" + # BindIt exports are UTF-8; reject alternate encodings rather than bypass this check. + try: + text = payload.decode("utf-8-sig") + if " tuple[str, int]: + """Read a length-prefixed UTF-8 container string with bounds checking.""" + if offset + 2 > len(data): + raise FlexProtocolError("Truncated BindIt string length") + size = struct.unpack_from(" len(data): + raise FlexProtocolError("Truncated BindIt string") + try: + return data[offset + 2 : end].decode("utf-8"), end + except UnicodeError as error: + raise FlexProtocolError("Invalid BindIt string") from error + + +def _gzip(data: bytes, offset: int, size: int) -> tuple[bytes, int]: + """Validate a complete size-bounded gzip member, including CRC and EOF.""" + end = offset + size + if size <= 0 or end > len(data): + raise FlexProtocolError("Invalid compressed stream size") + decoder = zlib.decompressobj(31) + try: + payload = decoder.decompress(data[offset:end], MAX_XML_SIZE + 1) + except zlib.error as error: + raise FlexProtocolError("Corrupt BindIt gzip member") from error + if len(payload) > MAX_XML_SIZE: + raise FlexProtocolError("BindIt XML exceeds the inspection size limit") + if not decoder.eof or decoder.unused_data or decoder.unconsumed_tail: + raise FlexProtocolError("BindIt gzip member does not match its declared size") + return payload, end + + +def _parts(data: bytes) -> tuple[str, str, bytes, bytes]: + """Parse variable-length header fields and validate both members and trailer.""" + if len(data) < 16 or data[:12] != bytes.fromhex("b6751cf201000a0001000000"): + raise FlexProtocolError("Unsupported BindIt header") + strings_size = struct.unpack_from(" len(data): + raise FlexProtocolError("Truncated BindIt header") + if data[offset : offset + 6] != bytes.fromhex("0b0001000000"): + raise FlexProtocolError("Unsupported BindIt object header") + content_size = struct.unpack_from(" len(data): + raise FlexProtocolError("Invalid second stream marker") + size = struct.unpack_from(" bool: + """Accept documented integer Celsius setpoints, including zero for RT/off mode.""" + return re.fullmatch(r"(?:0|[1-8][0-9]|9[0-6])", value) is not None + + +def _inspect_script(script: str) -> tuple[tuple[FlexCommand, ...], tuple[FlexProtocolIssue, ...]]: + """Collect syntax blockers and valid lines without authorizing partial execution.""" + issues = [] + if any((ord(char) < 32 and char not in "\r\n") or ord(char) > 126 for char in script): + issues.append( + FlexProtocolIssue( + "script_encoding", "CommandScript", "CommandScript contains non-ASCII or control characters" + ) + ) + commands = [] + for number, line in enumerate(script.splitlines(), 1): + location = f"CommandScript/line[{number}]" + if not line: + issues.append( + FlexProtocolIssue("empty_line", location, f"Empty script line {number}", number) + ) + continue + if any(ord(char) < 32 or ord(char) > 126 for char in line): + issues.append( + FlexProtocolIssue( + "script_encoding", + location, + f"Non-ASCII or control character on script line {number}", + number, + ) + ) + continue + opcode = line.split(" ", 1)[0] + form = _FORMS.get(opcode) + if form is None or re.fullmatch(form, line) is None: + issues.append( + FlexProtocolIssue( + "unsupported_command", location, f"Unsupported script line {number}: {line!r}", number + ) + ) + continue + if any( + len(value) > 11 or not -(2**31) <= int(value) < 2**31 + for value in re.findall(r"-?[0-9]+", line.split('"', 1)[0]) + ): + issues.append( + FlexProtocolIssue( + "integer_range", + location, + f"Integer exceeds the supported representation on line {number}", + number, + ) + ) + continue + if opcode == "TEM": + _, target, interval = line.split() + if not _temperature_supported(target) or (int(interval) != 0 and int(interval) < 100): + issues.append( + FlexProtocolIssue( + "temperature_range", + location, + "TEM requires an integer target of 0 or 10..96 C and a reporting interval " + "of 0 (disabled) or at least 100 centiseconds", + number, + ) + ) + continue + commands.append(FlexCommand(number, line)) + if not script.splitlines(): + issues.append(FlexProtocolIssue("empty_script", "CommandScript", "Empty CommandScript")) + return tuple(commands), tuple(issues) + + +def _preflight(script: str) -> tuple[FlexCommand, ...]: + """Reject the entire script if any line is unsupported; never stream a valid subset.""" + commands, issues = _inspect_script(script) + if issues: + raise FlexProtocolError("; ".join(issue.message for issue in issues)) + return commands + + +def _command_prompt(command: FlexCommand, orientation: int) -> FlexPrompt: + """Describe one validated PAU without conflating pauses and plate actions.""" + actions: dict[str, Literal["load", "unload", "pause"]] = { + "0": "pause", + "1": "load", + "2": "unload", + } + return FlexPrompt( + command.line_number, + actions[command.text.split()[1]], + command.text.split('"')[1], + orientation, + loading_slot(orientation), + ) + + +def _script_prompts(commands: tuple[FlexCommand, ...]) -> tuple[FlexPrompt, ...]: + """Describe plate requests and protocol pauses using preceding orientations.""" + prompts = [] + orientation = 0 + for command in commands: + if command.opcode == "MOR": + orientation = int(command.text.split()[1]) + elif command.opcode == "PAU": + prompts.append(_command_prompt(command, orientation)) + return tuple(prompts) + + +def _dispensing_prompts( + root: ET.Element, commands: tuple[FlexCommand, ...], plates: tuple[FlexPlate, ...] +) -> dict[int, tuple[FlexDispense, ...]]: + """Map standalone whole-plate additions to unambiguous exported PAU commands. + + Reject unmatched metadata, partial well groups and ambiguous plate/step names. + The observed TXT 11/MOR/PAU/TXT/MOR sequence presents one plate for addition. + """ + result: dict[int, tuple[FlexDispense, ...]] = {} + plate_map = {p.id: p for p in plates} + prompts = _script_prompts(commands) + markers = {c.line_number for c in commands if c.text.startswith("TXT 11 ")} + used_markers: set[int] = set() + for pause in root.findall("./Protocol/Run/Protocol/.//Steps/Pause"): + dispense = pause.find("Dispense") + if dispense is None or dispense.get("enabled") != "true": + continue + name = pause.get("name", "") + refs = pause.findall("Plates/Plate") + if len(refs) != 1 or refs[0].get("wellGroup") != "Plate": + raise FlexProtocolError("Dispense requires one whole-plate target") + plate = plate_map.get(refs[0].get("id", "")) + if plate is None or sum(p.name == plate.name for p in plates) != 1: + raise FlexProtocolError("Dispense target must identify a unique layout plate") + layout = root.findall("./Protocol/Run/PlateLayout/Plates/Plate") + groups = [ + g + for p in layout + if p.get("id") == plate.id + for g in p.findall("Wells/Group") + if g.get("name") == "Plate" + ] + if ( + len(groups) != 1 + or len(groups[0]) != 1 + or groups[0][0].tag != "Region" + or any( + groups[0][0].get(k) != v + for k, v in {"x": "0", "y": "0", "columns": "12", "rows": "8"}.items() + ) + ): + raise FlexProtocolError("Dispense requires the complete 96-well Plate group") + additions = [] + for reagent in dispense.findall("Reagents/Reagent"): + try: + volume = float(reagent.get("volume", "")) + except ValueError as error: + raise FlexProtocolError("Dispense volume must be a positive finite number") from error + reagent_name = reagent.get("name", "").strip() + if not reagent_name or not math.isfinite(volume) or volume <= 0: + raise FlexProtocolError("Dispense requires a reagent name and positive finite volume") + additions.append(FlexDispense(plate.id, plate.name, reagent_name, volume)) + if not additions: + raise FlexProtocolError("Dispense requires at least one reagent") + matches = [i for i, c in enumerate(commands) if c.text == f'TXT 11 "{name}"'] + if len(matches) != 1: + raise FlexProtocolError("Dispense requires one matching TXT 11 step marker") + i = matches[0] + block = commands[i : i + 5] + if ( + len(block) != 5 + or block[1].opcode != "MOR" + or block[2].text != f'PAU 0 "{pause.findtext("Message", "")}"' + or block[3].text != f'TXT "{name}"' + or block[4].opcode != "MOR" + ): + raise FlexProtocolError("Dispense requires the supported plate-presentation pause sequence") + loads = [ + p + for p in prompts + if p.action == "load" and p.label == plate.name and p.line_number < block[0].line_number + ] + if ( + len(loads) != 1 + or loads[0].loading_slot != loading_slot(int(block[1].text.split()[1])) + or int(block[4].text.split()[1]) != loads[0].loading_slot + ): + raise FlexProtocolError("Dispense orientation does not match the target plate's loaded tray") + if block[0].line_number in used_markers: + raise FlexProtocolError("Ambiguous duplicate dispensing step") + used_markers.add(block[0].line_number) + result[block[2].line_number] = tuple(additions) + if used_markers != markers: + raise FlexProtocolError("TXT 11 requires matching standalone Dispense metadata") + return result + + +@dataclass(frozen=True) +class FlexProtocol: + """An immutable BindIt export, inspected without reconstructing its XML. + + Reading supports inspection of unknown script features. Call ``preflight`` + to check executable coverage. Preserved XML includes reagents, empty shake + lists, disabled settings and any structures not interpreted by this reader. + """ + + name: str + bindit_version: str + original_bytes: bytes + properties_xml: bytes + exported_xml: bytes + script: str + plates: tuple[FlexPlate, ...] + tips: tuple[FlexTip, ...] + + @property + def sha256(self) -> str: + """Return the digest of the immutable original export.""" + return hashlib.sha256(self.original_bytes).hexdigest() + + def compatibility_issues(self) -> tuple[FlexProtocolIssue, ...]: + """Report all detected execution blockers without device access or script changes. + + This uses the same checks as ``preflight``. Unsupported features remain blocked + even if their generated command syntax is otherwise recognized. Malformed BDZ + containers are rejected earlier by ``read_flex_bdz``. No geometry calculation, + volume-profile lookup or hardware validation is performed here. + """ + issues: list[FlexProtocolIssue] = [] + try: + root = _xml(self.exported_xml, "ExportedData") + except FlexProtocolError as error: + issues.append(FlexProtocolIssue("invalid_xml", "ExportedData", str(error))) + else: + protocols = root.findall("./Protocol/Run/Protocol") + if len(protocols) != 1: + issues.append( + FlexProtocolIssue("not_executable", "Protocol", "Expected one executable protocol") + ) + else: + protocol = protocols[0] + if protocol.get("IsExecutable") != "true": + issues.append( + FlexProtocolIssue("not_executable", "Protocol", "Expected one executable protocol") + ) + if protocol.get("enabled", "true") != "true": + issues.append(FlexProtocolIssue("disabled_protocol", "Protocol", "Protocol is disabled")) + for index, step in enumerate(protocol.findall(".//Steps/*"), 1): + location = f"Steps/{step.tag}[{index}] {step.get('name', '')!r}" + if step.tag not in ("Tip", "Mix", "Dry", "ReleaseBeads", "CollectBeads", "Pause"): + issues.append( + FlexProtocolIssue("unsupported_step", location, f"Unsupported step type: {step.tag}") + ) + if step.tag != "Pause" and any( + pause.get("enabled") == "true" for pause in step.findall("Pause") + ): + issues.append( + FlexProtocolIssue( + "nested_pause", + location + "/Pause", + "Enabled nested Pause metadata is unsupported", + ) + ) + parents = {child: parent for parent in protocol.iter() for child in parent} + for feature in ("Heating", "PostTemperature", "Precollect", "Postmix", "Dispense"): + for index, element in enumerate(protocol.iter(feature), 1): + if element.get("enabled") == "true": + parent = parents[element] + # Mix precollection/postmix are sequences of existing supported motions + # and timers. Preserve the exported commands; do not regenerate them. + if feature in ("Precollect", "Postmix") and parent.tag == "Mix": + continue + if ( + feature == "Heating" + and parent.tag == "Mix" + and _temperature_supported(element.findtext("Temperature", "")) + and element.findtext("Preheat") in ("true", "false") + and f'TXT 7 "{parent.get("name", "")}"' in self.script.splitlines() + and any( + line.startswith(f"TEM {element.findtext('Temperature')} ") + for line in self.script.splitlines() + ) + and "MOI 310" in self.script.splitlines() + ): + continue + if feature == "Dispense" and parent.tag == "Pause": + continue + location = f"{parent.tag} {parent.get('name', '')!r}/{feature}[{index}]" + issues.append( + FlexProtocolIssue( + "unsupported_feature", + location, + f"Enabled {feature} metadata is unsupported", + ) + ) + if not self.plates: + issues.append(FlexProtocolIssue("missing_plates", "PlateLayout", "No layout plates")) + for plate in self.plates: + if plate.type_id != "8b7d7c98-275f-4285-8129-5f8ed46fb01e": + issues.append( + FlexProtocolIssue( + "unsupported_plate", + f"PlateLayout/{plate.id}", + f"Execution supports only 96 deep-well plates; " + f"plate {plate.name!r} uses {plate.type_id}", + ) + ) + if not self.tips: + issues.append(FlexProtocolIssue("missing_tips", "Steps/Tip", "No tip configuration")) + for tip in self.tips: + if tip.type_id != "2ab24e9a-b88e-4d3b-8fc3-babbc5a7c742": + issues.append( + FlexProtocolIssue( + "unsupported_tip", + f"Tip {tip.name!r}", + f"Execution supports only the 96DW tip configuration; " + f"tip {tip.name!r} uses {tip.type_id}", + ) + ) + commands, script_issues = _inspect_script(self.script) + issues.extend(script_issues) + orientation = None + timer_set = False + for command in commands: + location = f"CommandScript/line[{command.line_number}]" + if command.opcode == "MOR": + orientation = int(command.text.split()[1]) + if command.opcode == "PAU" and orientation is None: + issues.append( + FlexProtocolIssue( + "prompt_without_orientation", + location, + "PAU requires a preceding MOR orientation", + command.line_number, + ) + ) + if command.opcode == "WAI": + if len(command.text.split()) == 3: + timer_set = True + elif not timer_set: + issues.append( + FlexProtocolIssue( + "wait_without_timer", + location, + "WAI wait requires a preceding timer-setting command", + command.line_number, + ) + ) + if not issues: + try: + _dispensing_prompts(root, commands, self.plates) + except FlexProtocolError as error: + issues.append(FlexProtocolIssue("unsupported_dispense", "Dispense", str(error))) + return tuple(issues) + + def preflight(self, mode: Literal["stream", "record"] = "stream") -> tuple[FlexCommand, ...]: + """Require a compatible whole protocol before selecting stream or record commands. + + Recording retains DTL/TIM; streaming omits only those recognized metadata forms. + All compatibility blockers are included in the exception. Accepted syntax does + not establish physical validation, and no supported subset is returned on failure. + """ + if mode not in ("stream", "record"): + raise ValueError("mode must be 'stream' or 'record'") + issues = self.compatibility_issues() + if issues: + raise FlexProtocolError("; ".join(f"{issue.location}: {issue.message}" for issue in issues)) + commands = _preflight(self.script) + return tuple(c for c in commands if mode == "record" or c.opcode not in ("DTL", "TIM")) + + def prompts(self) -> tuple[FlexPrompt, ...]: + """Describe validated plate requests and protocol pauses in script order.""" + commands = self.preflight() + additions = _dispensing_prompts(_xml(self.exported_xml, "ExportedData"), commands, self.plates) + prompts = [] + for prompt in _script_prompts(commands): + required = additions.get(prompt.line_number, ()) + if required: + instructions = "; ".join( + f"Add {a.volume:g} uL of {a.reagent_name} per well to {a.plate_name}" for a in required + ) + prompt = replace(prompt, label=f"{prompt.label} — {instructions}", additions=required) + prompts.append(prompt) + return tuple(prompts) + + +def read_flex_bdz(source: ProtocolSource) -> FlexProtocol: + """Read a path or bytes, validating Flex identity without rewriting the export.""" + data = source if isinstance(source, bytes) else Path(source).read_bytes() + if len(data) > 2 * MAX_XML_SIZE: + raise FlexProtocolError("BindIt file exceeds the inspection size limit") + version, trailer_name, props, exported = _parts(data) + properties = _xml(props, "Properties") + objects = properties.findall("ExportedObject") + parameters = properties.find("InstrumentParameters") + if len(objects) != 1 or objects[0].findtext("InstrumentTypeId") != FLEX_TYPE_ID: + raise FlexProtocolError("Expected exactly one Flex exported object") + if parameters is None or parameters.get("type") != "711": + raise FlexProtocolError("Expected Flex InstrumentParameters type 711") + root = _xml(exported, "ExportedData") + protocols = root.findall("./Protocol/Run/Protocol") + if len(protocols) != 1: + raise FlexProtocolError("Expected exactly one protocol") + name = objects[0].get("name", "") + if not name or name != trailer_name or name != protocols[0].get("name"): + raise FlexProtocolError("Protocol names disagree across container sections") + scripts = root.findall("CommandScript") + if len(scripts) != 1 or not scripts[0].text: + if protocols[0].get("IsExecutable") == "false": + plate_ids = {p.get("id") for p in root.findall("./Protocol/Run/PlateLayout/Plates/Plate")} + unresolved = [ + step.get("name", step.tag) + for step in protocols[0].findall(".//Steps/*") + if any(p.get("id") not in plate_ids for p in step.findall("Plates/Plate")) + ] + detail = f" Unassigned or unknown plate in: {', '.join(unresolved)}." if unresolved else "" + raise FlexProtocolError( + "BindIt marked the export non-executable and supplied no usable CommandScript." + + detail + + " Correct the protocol in BindIt and re-export it." + ) + raise FlexProtocolError("Expected one nonempty generated CommandScript") + plates = tuple( + FlexPlate(p.get("id", ""), p.get("name", ""), p.get("plateTypeID", "")) + for p in root.findall("./Protocol/Run/PlateLayout/Plates/Plate") + ) + if any(not p.id or not p.type_id for p in plates) or len({p.id for p in plates}) != len(plates): + raise FlexProtocolError("Invalid or duplicate plate identities") + tips = tuple( + FlexTip( + tip.get("name", ""), + tip.get("id", ""), + tuple(p.get("id", "") for p in tip.findall("Plates/Plate")), + ) + for tip in protocols[0].findall("Steps/Tip") + ) + plate_ids = {p.id for p in plates} + for reference in protocols[0].findall(".//Plates/Plate"): + if reference.get("id") not in plate_ids: + raise FlexProtocolError("Protocol references a plate absent from its layout") + return FlexProtocol(name, version, data, props, exported, scripts[0].text, plates, tips) diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/protocol_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/protocol_tests.py new file mode 100644 index 00000000000..b20ebf062f8 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/protocol_tests.py @@ -0,0 +1,132 @@ +"""BDZ parsing, compatibility checks and malformed-input rejection.""" + +import hashlib +import unittest +from dataclasses import replace +from pathlib import Path +from tempfile import TemporaryDirectory + +from .protocol import FlexProtocolError, _preflight, loading_slot, read_flex_bdz +from .test_helpers import bdz_bytes + + +class FlexProtocolTests(unittest.TestCase): + """Exercise the container reader with synthetic, non-laboratory test data.""" + + def setUp(self) -> None: + """Build an in-memory container without files or USB.""" + self.raw = bdz_bytes() + self.protocol = read_flex_bdz(self.raw) + + def test_sections_and_path_input_are_preserved(self) -> None: + """Retain original bytes, metadata and source line numbers.""" + p = self.protocol + self.assertEqual(p.name, "unit test") + self.assertEqual(p.original_bytes, self.raw) + self.assertEqual(p.sha256, hashlib.sha256(self.raw).hexdigest()) + self.assertEqual(len(p.preflight("record")), 5) + self.assertEqual([c.line_number for c in p.preflight()], [3, 4, 5]) + self.assertEqual(p.plates[0].id, "plate") + self.assertEqual(p.tips[0].plate_ids, ("plate",)) + with TemporaryDirectory() as tmp: + path = Path(tmp) / "input.bdz" + path.write_bytes(self.raw) + self.assertEqual(read_flex_bdz(path), p) + for version in ("4.0.0.45", "4.1", "4.123.456.789"): + self.assertEqual(read_flex_bdz(bdz_bytes(version=version)).bindit_version, version) + + def test_corruption_truncation_and_trailing_data(self) -> None: + """Reject malformed lengths, magic, gzip content and trailers.""" + for offset in (0, 12, 16, 48, 56, 80, len(self.raw) - 1): + raw = bytearray(self.raw) + raw[offset] ^= 0xFF + with self.subTest(offset=offset), self.assertRaises(FlexProtocolError): + read_flex_bdz(bytes(raw)) + for length in (0, 15, 40, 59, len(self.raw) - 1): + with self.subTest(length=length), self.assertRaises(FlexProtocolError): + read_flex_bdz(self.raw[:length]) + with self.assertRaises(FlexProtocolError): + read_flex_bdz(self.raw + b"garbage") + + def test_unknown_commands_and_missing_context_reject_entire_script(self) -> None: + """Never silently filter unsupported commands or invent timer/orientation state.""" + for script in ( + 'REC 1 "x" 1', + "ABC 1", + "MOR 9", + 'PAU 3 "x"', + "MOH 1 2 1 3 0", + "WAI 2 100", + 'PAU 1 "plate"', + "WAI 1", + "MOR 1\vMOR 2", + "", + ): + with self.subTest(script=script), self.assertRaises(FlexProtocolError): + replace(self.protocol, script=script).preflight() + with self.assertRaises(FlexProtocolError): + _preflight("MOH 1 2 0 " + "9" * 5000 + " 0") + + def test_prompt_orientation_and_labels(self) -> None: + """Labels are not tray indices; MOR determines loading orientation.""" + p = replace(self.protocol, script='MOR 7\nPAU 1 "3"\nMOR 6\nPAU 2 "1"\n') + self.assertEqual([(q.label, q.loading_slot) for q in p.prompts()], [("3", 2), ("1", 1)]) + self.assertEqual([loading_slot(i) for i in range(1, 9)], [4, 5, 6, 7, 8, 1, 2, 3]) + + def test_temperature_range_and_interval(self) -> None: + """Validate documented setpoints independently of a particular method.""" + for target in (0, *range(10, 97)): + for interval in (0, 100, 300): + self.assertEqual(len(_preflight(f"TEM {target} {interval}")), 1) + for line in ("TEM 9 300", "TEM 97 300", "TEM 37 99", "TEM -1 300", "MOI 311"): + with self.subTest(line=line), self.assertRaises(FlexProtocolError): + _preflight(line) + + def test_features_require_supported_metadata_and_commands(self) -> None: + """Heating requires its target and block movement; unknown features stay blocked.""" + steps = ( + '37' + 'true' + '' + ) + p = read_flex_bdz(bdz_bytes('TXT 7 "mix"\nTEM 37 300\nMOI 310\n', steps=steps)) + self.assertEqual(p.compatibility_issues(), ()) + self.assertTrue(replace(p, script='TXT 7 "mix"\nTEM 37 300\n').compatibility_issues()) + for step in ( + '', + '', + '', + ): + with self.subTest(step=step), self.assertRaises(FlexProtocolError): + read_flex_bdz(bdz_bytes(steps=step)).preflight() + + def test_multiple_blockers_reported_without_mutation(self) -> None: + """Collect independent labware, syntax and timer failures before execution.""" + p = replace( + self.protocol, + plates=(replace(self.protocol.plates[0], type_id="other"),), + tips=(replace(self.protocol.tips[0], type_id="other"),), + script='PAU 1 "plate"\nWAI 1\nABC 1\n', + ) + self.assertEqual( + {i.code for i in p.compatibility_issues()}, + { + "unsupported_plate", + "unsupported_tip", + "unsupported_command", + "prompt_without_orientation", + "wait_without_timer", + }, + ) + with self.assertRaises(FlexProtocolError): + p.preflight() + + def test_xml_declarations_and_invalid_identity_rejected(self) -> None: + """Preflight rejects entity declarations and non-executable XML.""" + for xml in ( + b"", + b"", + self.protocol.exported_xml.replace(b'IsExecutable="true"', b'IsExecutable="false"'), + ): + with self.subTest(xml=xml[:40]), self.assertRaises(FlexProtocolError): + replace(self.protocol, exported_xml=xml).preflight() diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/stored_tests.py b/pylabrobot/thermo_fisher/kingfisher/flex/stored_tests.py new file mode 100644 index 00000000000..0605c36725f --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/stored_tests.py @@ -0,0 +1,105 @@ +"""Onboard inventory parsing and documented BGN lifecycle, without hardware.""" + +import asyncio +import unittest + +from .connection import FlexError +from .flex import KingFisherFlex +from .flex_tests import ScriptedHID + + +class FlexStoredTests(unittest.IsolatedAsyncioTestCase): + """Exercise inventory and launch with simulated replies.""" + + async def asyncSetUp(self) -> None: + """Construct a driver with an in-memory transport only.""" + self.io = ScriptedHID() + self.flex = KingFisherFlex(timeout=0.1) + self.flex.io = self.io + self.io.responses["REP 4"] = 'REP\r\nREP 4 1 1 "KF_Start_Test" 200\r\nREP END 0\r\n' + + async def asyncTearDown(self) -> None: + """Close the reader even after intentionally invalidating it.""" + await self.flex.stop() + + async def test_inventory_parses_names_and_summary(self) -> None: + """Return entry names without mistaking numeric summary fields for entries.""" + self.io.responses["REP 4"] = ( + 'REP\r\nREP 4 1 1 "one" 100\r\nREP 4 2 1 "two" 200\r\nREP 4 1 2 3 4 5\r\nREP END 0\r\n' + ) + await self.flex.setup() + self.assertEqual(await self.flex.list_protocols(), ("one", "two")) + self.assertEqual(self.io.commands, ["VER", "REP 4"]) + + async def test_invalid_names_and_deadlines_send_nothing(self) -> None: + """Empty names, control characters and invalid deadlines never reach USB.""" + for name in ("", "a\nBGN b", "é", " "): + with self.subTest(name=name), self.assertRaises(ValueError): + await self.flex.start_stored_protocol(name) + with self.assertRaises(ValueError): + await self.flex.start_stored_protocol("KF_Start_Test", completion_timeout=0) + self.assertEqual(self.io.opened, 0) + self.assertEqual(self.io.commands, []) + + async def test_long_inventory_names_are_transmitted_exactly(self) -> None: + """Stored names survive quoting and HID fragmentation without truncation.""" + await self.flex.setup() + for name in ("Long stored protocol name", 'Long "quoted" protocol ' + "x" * 70): + with self.subTest(name=name): + quoted = name.replace('"', '""') + self.io.responses["REP 4"] = f'REP\r\nREP 4 1 1 "{quoted}" 200\r\nREP END 0\r\n' + await self.flex.start_stored_protocol(name) + self.assertEqual(self.io.commands[-1], f'BGN "{quoted}"') + + async def test_missing_duplicate_and_malformed_inventory_cannot_launch(self) -> None: + """Reject ambiguous or unknown inventory instead of guessing a stored entry.""" + await self.flex.setup() + for data in ( + "", + 'REP 4 1 1 "KF_Start_Test" 1\r\n' * 2, + "REP 4 unknown\r\n", + ): + self.io.responses["REP 4"] = f"REP\r\n{data}REP END 0\r\n" + with self.assertRaises(FlexError): + await self.flex.start_stored_protocol("KF_Start_Test") + self.assertEqual(self.io.commands, ["VER", "REP 4", "REP 4", "REP 4"]) + + async def test_ack_does_not_finish_launch_and_abort_bypasses_wait(self) -> None: + """BGN hides internal PAU; abort must work without a BGN END response.""" + await self.flex.setup() + command = 'BGN "KF_Start_Test"' + self.io.responses[command] = "BGN\r\n" + task = asyncio.create_task(self.flex.start_stored_protocol("KF_Start_Test")) + await self.io.wait_sent(command) + self.assertFalse(task.done()) + with self.assertRaisesRegex(FlexError, "Another Flex operation"): + await self.flex.list_protocols() + await self.flex.abort() + with self.assertRaises(FlexError): + await task + self.assertEqual(self.io.commands, ["VER", "REP 4", command]) + self.assertEqual(self.io.features, [b"\x00\x01\x00"]) + self.assertEqual(self.flex.state, "uncertain") + + async def test_terminal_success_and_quoted_name(self) -> None: + """Only BGN END 0 finishes, with position and initialization still unknown.""" + self.io.responses["REP 4"] = 'REP\r\nREP 4 1 1 "Test ""A""" 200\r\nREP END 0\r\n' + await self.flex.setup() + await self.flex.start_stored_protocol('Test "A"') + self.assertEqual(self.io.commands[-1], 'BGN "Test ""A"""') + self.assertEqual(self.flex.state, "idle") + self.assertIsNone(self.flex.processing_slot) + + async def test_errors_timeout_and_qit_never_retry(self) -> None: + """A failed stored launch stays uncertain and is never resent.""" + for reply in ("BGN\r\nBGN END 8\r\n", "BGN\r\nQIT\r\n", "BGN\r\n"): + with self.subTest(reply=reply): + await self.flex.stop() + await self.flex.setup() + command = 'BGN "KF_Start_Test"' + self.io.responses[command] = reply + before = len(self.io.commands) + with self.assertRaises((FlexError, TimeoutError)): + await self.flex.start_stored_protocol("KF_Start_Test", completion_timeout=0.02) + self.assertEqual(self.io.commands[before:], ["REP 4", command]) + self.assertEqual(self.flex.state, "uncertain") diff --git a/pylabrobot/thermo_fisher/kingfisher/flex/test_helpers.py b/pylabrobot/thermo_fisher/kingfisher/flex/test_helpers.py new file mode 100644 index 00000000000..85f39ee32b7 --- /dev/null +++ b/pylabrobot/thermo_fisher/kingfisher/flex/test_helpers.py @@ -0,0 +1,65 @@ +"""Synthetic container data for offline tests; not executable laboratory methods.""" + +import gzip +import struct +import xml.etree.ElementTree as ET + +from .protocol import FLEX_TYPE_ID + + +def bdz_bytes( + script: str = 'DTL 0 "unit test"\nTIM 1\nMOR 6\nPAU 1 "plate"\nPAU 2 "plate"\n', + *, + steps: str = "", + version: str = "4.1.0.2", +) -> bytes: + """Construct minimal BDZ sections for parser and transport tests only.""" + name = "unit test" + props = ( + '' + f"{FLEX_TYPE_ID}" + '' + ).encode() + root = ET.fromstring( + "" + '' + '' + "" + '' + '' + '' + f"{steps}" + ) + ET.SubElement(root, "CommandScript").text = script + + def string(value: str) -> bytes: + """Pack a container string with its byte length.""" + encoded = value.encode() + return struct.pack("