diff --git a/docs/_static/devices.json b/docs/_static/devices.json index 1335c70215c..db01d7363cc 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -526,16 +526,26 @@ "oem": "https://www.hamiltoncompany.com/microlab-vantage" }, { - "id": "hettich-rotanta-460", + "id": "hettich-robotic-centrifuges", "vendor": "Hettich", - "name": "ROTANTA 460 Robotic", + "name": "Robotic centrifuges", + "models": [ + {"name": "MIKRO 220 Robotic", "status": "full"}, + {"name": "ROTANTA 460 Robotic", "status": "wip"}, + {"name": "ROTINA 380 Robotic", "status": "wip"}, + {"name": "ROTINA 380 R Robotic", "status": "wip"} + ], "kind": "centrifuge", "capabilities": [ "centrifuging" ], "status": "wip", + "api": "pylabrobot.hettich.HettichRoboticCentrifuge", + "api_version": "v1", + "code_slug": "hettich/centrifuge", + "doc_slug": "hettich/mikro-220/hello-world", "manager": "https://discuss.pylabrobot.org/u/rickwierenga", - "oem": "https://www.hettichlab.com/products/centrifuges/automated-centrifuges/rotanta-460-robotic/" + "notes": "Only MIKRO 220 Robotic has been hardware-validated with this driver. ROTANTA 460 Robotic, ROTINA 380 Robotic, and ROTINA 380 R Robotic have model-specific protocol configurations but have not been verified on hardware." }, { "id": "hettich-sbs-300r", diff --git a/docs/api/pylabrobot.hettich.rst b/docs/api/pylabrobot.hettich.rst new file mode 100644 index 00000000000..8575a2a1415 --- /dev/null +++ b/docs/api/pylabrobot.hettich.rst @@ -0,0 +1,25 @@ +.. currentmodule:: pylabrobot.hettich + +pylabrobot.hettich package +========================== + +.. currentmodule:: pylabrobot.hettich.centrifuge + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + HettichMikro220RoboticCentrifuge + HettichRotanta460RoboticCentrifuge + HettichRotina380RoboticCentrifuge + HettichRotina380RRoboticCentrifuge + HettichRoboticCentrifuge + HettichCooledRoboticCentrifuge + HettichCentrifugeError + HettichCommunicationError + HettichCommandError + CentrifugeStatus + HatchStatus + HettichMachineState + RotorSpecification diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index 394622fb707..225e2ef6792 100644 --- a/docs/api/pylabrobot.rst +++ b/docs/api/pylabrobot.rst @@ -28,6 +28,7 @@ Manufacturers pylabrobot.cole_parmer pylabrobot.curiox pylabrobot.hamilton + pylabrobot.hettich pylabrobot.high_res pylabrobot.inheco pylabrobot.kbioscience diff --git a/docs/contributor_guide/event-schemas.md b/docs/contributor_guide/event-schemas.md index 9f197694c48..3b181480333 100644 --- a/docs/contributor_guide/event-schemas.md +++ b/docs/contributor_guide/event-schemas.md @@ -330,7 +330,7 @@ one is assigned at operation start. | Operation | Fields | Notes | | --- | --- | --- | -| `centrifuge.spin` | `device`, `resources`, `bucket_resources`, `relative_centrifugal_force`, `duration`, `acceleration_fraction`, `deceleration_fraction` | Describes one requested spin cycle. | +| `centrifuge.spin` | `device`, `resources`, `bucket_resources`, `relative_centrifugal_force`, `duration`; optional `acceleration_fraction`, `deceleration_fraction` | Describes one requested spin cycle. | `resources` contains directly loaded resources only. Empty buckets are not represented. `bucket_resources` preserves the association between each loaded resource and its holder: @@ -342,6 +342,10 @@ one is assigned at operation start. } ``` +A frontend without a PLR rotor-resource model emits empty `resources` and `bucket_resources`. +Hettich and VSpin report the requested `relative_centrifugal_force`. VSpin also reports +`acceleration_fraction` and `deceleration_fraction`. + `relative_centrifugal_force` is the dimensionless multiple of standard gravity conventionally written as x g. Acceleration and deceleration are fractions of the device maximum. diff --git a/docs/user_guide/hettich/events.md b/docs/user_guide/hettich/events.md new file mode 100644 index 00000000000..ea0a9446a8e --- /dev/null +++ b/docs/user_guide/hettich/events.md @@ -0,0 +1,15 @@ +# Hettich robotic centrifuge events + +Each semantic operation emits `started`, `completed`, or `failed` lifecycle records. Hettich uses +the same canonical centrifuge operation name and common payload fields as other PyLabRobot +centrifuge frontends, including VSpin. + +| Operation | Primary fields | +| --- | --- | +| `centrifuge.spin` | `device`, empty `resources`, empty `bucket_resources`, `relative_centrifugal_force`, `duration` | + +The driver reports the requested `g` as `relative_centrifugal_force`, in multiples of standard +gravity (× g), including when the value was calculated with `rpm_to_g()`. Conversion to the +device's integer RPM happens internally; the event preserves the requested force without +rounding it to the achievable RPM. The Hettich frontend does not currently model rotor positions +as PLR resource holders, so both resource lists are empty. diff --git a/docs/user_guide/hettich/index.md b/docs/user_guide/hettich/index.md new file mode 100644 index 00000000000..20f69883458 --- /dev/null +++ b/docs/user_guide/hettich/index.md @@ -0,0 +1,9 @@ +# Hettich + +```{toctree} +:maxdepth: 1 + +mikro-220/hello-world +events +state +``` diff --git a/docs/user_guide/hettich/mikro-220/hello-world.ipynb b/docs/user_guide/hettich/mikro-220/hello-world.ipynb new file mode 100644 index 00000000000..ce23a7196e6 --- /dev/null +++ b/docs/user_guide/hettich/mikro-220/hello-world.ipynb @@ -0,0 +1,460 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "8c0b24bb", + "metadata": {}, + "source": [ + "# Hettich MIKRO 220 Robotic quickstart\n", + "\n", + "The MIKRO 220 Robotic is a compact centrifuge with a motorized loading hatch and rotor positioning for automated loading.\n", + "\n", + "| Property | Value |\n", + "|---|---|\n", + "| Communication | RS-232 through a serial or USB-to-serial adapter |\n", + "| Serial settings | 9600 baud, 7 data bits, even parity, 1 stop bit |\n", + "| Protocol generation | Hettich Generation 2 |\n", + "| Target-speed duration | 1-second resolution; acceleration and braking excluded |\n", + "| Rotor 2334 | 24 × 2.0 mL; 13,000 rpm; 18,327 × g |\n", + "| Rotor 2394 | 24 × 2.0 mL; 13,000 rpm; 18,516 × g |\n", + "| Power | 460 VA |\n", + "| Cooling | Air-cooled; no chamber-temperature API |\n", + "\n", + "Specifications: [MIKRO 220 Robotic product page](https://www.hettichlab.com/products/centrifuges/automated-centrifuges/mikro-220-robotic/) and [manufacturer data sheet](https://www.hettichlab.com/downloadcenter/Products/Datasheets/MIKRO220Robotic_EN.pdf).\n", + "\n", + "Follow Hettich's rotor, accessory, balancing, installation, and guarding requirements before permitting motion." + ] + }, + { + "cell_type": "markdown", + "id": "352f7ef0", + "metadata": {}, + "source": [ + "```{device-card} hettich-robotic-centrifuges\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "c5527c78", + "metadata": {}, + "source": [ + "## How it talks\n", + "\n", + "The driver sends addressed Hettich ENQUIRY frames for state and SELECT frames for commands. It verifies the address, parameter, hexadecimal value, and XOR block-check character in every reply. A rejected command automatically reads and decodes the SIOF serial-fault register." + ] + }, + { + "cell_type": "markdown", + "id": "4962d6c7", + "metadata": {}, + "source": [ + "## Physical setup\n", + "\n", + "1. Install only rotor 2334 or 2394 with approved accessories, following the Hettich operating instructions. Confirm the catalog number printed on the installed rotor.\n", + "2. Keep the rotor empty for initial hatch and positioning checks. Balance every later load by mass and position; never run an incomplete or uncertain loading pattern.\n", + "3. Close and lock the **main centrifuge lid**. The motorized **loading hatch** is a separate opening within that lid.\n", + "4. Turn the key-operated switch to `LOCK 2` for remote control.\n", + "5. Connect the serial adapter and power on the centrifuge." + ] + }, + { + "cell_type": "markdown", + "id": "cac8e407", + "metadata": {}, + "source": [ + "## Connect\n", + "\n", + "`setup()` opens the serial port, clears and reads the startup SIOF state, and verifies the exact MIKRO device code, Generation 2 identifier, and firmware version. It does not move the machine.\n", + "\n", + "Replace `` with the adapter path or COM port for this installation. Set `ROTOR_CATALOG_NUMBER` from the marking on the installed rotor." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4dd01a91", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.hettich import HettichMikro220RoboticCentrifuge, MIKRO_220_ROBOTIC_ROTORS\n", + "\n", + "ROTOR_CATALOG_NUMBER = \"2334\" # Change to \"2394\" when that rotor is installed.\n", + "centrifuge = HettichMikro220RoboticCentrifuge(\n", + " port=\"\",\n", + " rotor_catalog_number=ROTOR_CATALOG_NUMBER,\n", + ")\n", + "await centrifuge.setup()" + ] + }, + { + "cell_type": "markdown", + "id": "5bcd0e76", + "metadata": {}, + "source": [ + "## Confirm identity\n", + "\n", + "The detected model and firmware are available after setup." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "163d5613", + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Model:\", centrifuge.device_type)\n", + "print(\"Firmware:\", centrifuge.software_version)" + ] + }, + { + "cell_type": "markdown", + "id": "cf4e28e1", + "metadata": {}, + "source": [ + "## Inspect the rotor table and convert RPM and RCF\n", + "\n", + "Catalog numbers 2334 and 2394 identify two different 24-place rotors, not commands or error codes. Their radii produce slightly different relative centrifugal force (RCF). The selected rotor makes both conversion directions available programmatically." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "55760104", + "metadata": {}, + "outputs": [], + "source": [ + "for catalog_number, rotor in MIKRO_220_ROBOTIC_ROTORS.items():\n", + " print(\n", + " catalog_number,\n", + " f\"{rotor.positions} x {rotor.maximum_volume / 1000:g} mL,\",\n", + " f\"{rotor.maximum_speed:,} rpm, {rotor.maximum_rcf:,} x g\",\n", + " )\n", + "\n", + "print(\"RCF at 5,000 rpm:\", round(centrifuge.rpm_to_g(5_000)), \"x g\")\n", + "print(\"RPM for 5,000 x g:\", centrifuge.g_to_rpm(5_000), \"rpm\")" + ] + }, + { + "cell_type": "markdown", + "id": "593223a7", + "metadata": {}, + "source": [ + "## Read machine status\n", + "\n", + "`request_status()` reports the spin phase, errors, current program, rotor, main-lid state, key position, and whether centrifugation can start." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a0ef48f6", + "metadata": {}, + "outputs": [], + "source": [ + "status = await centrifuge.request_status()\n", + "print(status)" + ] + }, + { + "cell_type": "markdown", + "id": "3df7bd12", + "metadata": {}, + "source": [ + "## Read loading-hatch and positioning status\n", + "\n", + "The loading hatch has its own motion and closed signals, separate from the main lid." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c862e779", + "metadata": {}, + "outputs": [], + "source": [ + "hatch = await centrifuge.request_hatch_status()\n", + "print(hatch)" + ] + }, + { + "cell_type": "markdown", + "id": "932bb460", + "metadata": {}, + "source": [ + "## Read actual speed\n", + "\n", + "`request_speed()` returns the measured rotor speed in rpm." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "44dab35d", + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Actual speed:\", await centrifuge.request_speed(), \"rpm\")" + ] + }, + { + "cell_type": "markdown", + "id": "2eef09c5", + "metadata": {}, + "source": [ + "## Read the installed rotor's speed limit\n", + "\n", + "The live limit is checked again by `spin()` before motion." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "98f44f07", + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Rotor speed limit:\", await centrifuge.request_maximum_speed(), \"rpm\")" + ] + }, + { + "cell_type": "markdown", + "id": "4a9d1fc7", + "metadata": {}, + "source": [ + "## Read the device run timer\n", + "\n", + "`request_elapsed_time()` returns the current acceleration-inclusive device timer in seconds. `spin()` accounts for this internally so its public `duration` means time at target speed." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "688c9a38", + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Device run timer:\", await centrifuge.request_elapsed_time(), \"s\")" + ] + }, + { + "cell_type": "markdown", + "id": "d00d20e0", + "metadata": {}, + "source": [ + "## Open the loading hatch\n", + "\n", + "With an empty rotor and the main lid closed, `open_hatch()` checks the current state and sends a command only when needed. It requires standstill and `LOCK 2`, then waits for the open sensor." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5c22dbf1", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.open_hatch()" + ] + }, + { + "cell_type": "markdown", + "id": "e683c930", + "metadata": {}, + "source": [ + "## Position the empty rotor\n", + "\n", + "`move_to_position()` accepts only positions reported for the installed rotor. The main lid must be closed; the smaller loading hatch may be open, closed, or opening. Use `speed=\"slow\"` for the initial empty-rotor check and agitation-sensitive samples. The rotor remains held in positioning mode when the method returns." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8005a9b9", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.move_to_position(1, speed=\"slow\")" + ] + }, + { + "cell_type": "markdown", + "id": "fb9488eb", + "metadata": {}, + "source": [ + "## End positioning mode\n", + "\n", + "`end_positioning()` releases the positioning hold if it is active. Repeating it is safe." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4314622d", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.end_positioning()" + ] + }, + { + "cell_type": "markdown", + "id": "db50a397", + "metadata": {}, + "source": [ + "## Close the loading hatch\n", + "\n", + "`close_hatch()` is state-based and waits for both loading-hatch closed signals." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "166423b9", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.close_hatch()" + ] + }, + { + "cell_type": "markdown", + "id": "3514f9f2", + "metadata": {}, + "source": [ + "## Select a stored program\n", + "\n", + "`select_program()` selects a program from the centrifuge's stored range 1–89 without starting it. Selecting the already-active program is a no-op. The later `spin()` call replaces runtime and speed while retaining the program's other settings, such as acceleration and braking profiles." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ca563d34", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.select_program(1)" + ] + }, + { + "cell_type": "markdown", + "id": "c54970c7", + "metadata": {}, + "source": [ + "## Load, balance, and verify both closures\n", + "\n", + "After the empty-rotor checks, load and balance the rotor using the approved procedure. Before spinning, confirm that the main lid is locked and both loading-hatch closed signals are active." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "28d77e15", + "metadata": {}, + "outputs": [], + "source": [ + "status = await centrifuge.request_status()\n", + "hatch = await centrifuge.request_hatch_status()\n", + "print(\"Main lid closed:\", status.lid_closed)\n", + "print(\"Loading hatch closed:\", hatch.hatch_closed)\n", + "print(\"Loading hatch lock closed:\", hatch.lid_lock_closed)\n", + "print(\"Centrifugation possible:\", status.can_start)" + ] + }, + { + "cell_type": "markdown", + "id": "46a16499", + "metadata": {}, + "source": [ + "## Run a balanced test\n", + "\n", + "Stop here until the rotor has a fully balanced, approved load and the protected installation is cleared for motion. This example is deliberately gentle: 30 × g for 10 seconds. `g` is relative centrifugal force in multiples of standard gravity. `duration` is time at the requested force; acceleration and braking are excluded. The method validates the native timer range before START, checks all motion prerequisites, and returns only after standstill." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "dd58bbc5", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.spin(g=30, duration=10)" + ] + }, + { + "cell_type": "markdown", + "id": "6ad0c03a", + "metadata": {}, + "source": [ + "## Specify a run using RPM\n", + "\n", + "For a protocol specified in RPM, convert it to relative centrifugal force using the configured rotor. `rpm_to_g()` and `g_to_rpm()` require a supported `rotor_catalog_number`. This example requests the force corresponding to 500 rpm for 10 seconds." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "87644557", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.spin(g=centrifuge.rpm_to_g(500), duration=10)" + ] + }, + { + "cell_type": "markdown", + "id": "d2948bb0", + "metadata": {}, + "source": [ + "## Emergency-stop an active run\n", + "\n", + "The Hettich manual classifies PC STOP as an **emergency stop**. Use `stop_spin()` only when an active cycle must be aborted. It is a no-op at standstill." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ef36b59f", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.stop_spin()" + ] + }, + { + "cell_type": "markdown", + "id": "68b48b9a", + "metadata": {}, + "source": [ + "## Disconnect\n", + "\n", + "`stop()` closes the serial connection without changing the centrifuge's run state." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1ee36fc8", + "metadata": {}, + "outputs": [], + "source": [ + "await centrifuge.stop()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/hettich/state.md b/docs/user_guide/hettich/state.md new file mode 100644 index 00000000000..ad5d8af27e2 --- /dev/null +++ b/docs/user_guide/hettich/state.md @@ -0,0 +1,58 @@ +# Hettich workflow state and recovery + +`centrifuge.state` is an immutable snapshot of the workflow owned by PyLabRobot: + +```python +print(centrifuge.state.connection) +print(centrifuge.state.activity) +print(centrifuge.state.recovery_required) +``` + +`connection` tracks setup and disconnection. It is `unknown` if transport setup or +closure could not be confirmed. Call `stop()` successfully before reconnecting in +that case. `activity` describes the current operation, including hatch motion, +positioning, program selection, and the preparation, acceleration, at-speed, and +braking phases of a spin. `idle` means that no PLR workflow owns the device; it does +not mean that the rotor is physically stopped. + +Use `request_status()`, `request_hatch_status()`, and `request_speed()` for live +physical conditions. Status queries remain available while another operation is +active or recovery is required. + +## Operation ownership + +One workflow owns the device from its prerequisite checks through its final +confirmation. Conflicting motion, setup, recovery, and disconnect calls fail +immediately instead of interleaving their commands. + +`stop_spin()` can interrupt a spin owned by PLR. It asks that workflow to stop, +waits for ownership to be released, and verifies standstill, sending the emergency +STOP command if necessary. The interrupted `spin()` call raises an error. A stop +request during preparation prevents START if it has not yet been transmitted. +If no PLR spin is active, `stop_spin()` checks the hardware and stops an active run. + +## Recovery after failure + +A failed or cancelled operation that may have changed hardware or programmed +settings sets `recovery_required`. The driver records possible changes before +transmitting SELECT, because the device can accept a command whose reply is lost. +For ordinary operations, validation failures before any SELECT do not set this flag. + +While recovery is required, ordinary motion and program selection are blocked. +Status reads, emergency stopping, and disconnection remain available. Successful +spin cleanup or disconnecting and reconnecting does not clear the flag. + +Resolve the fault and inspect the samples, rotor, hatch, and program settings. +If needed, reconnect with `setup()`, then run: + +```python +await centrifuge.recover() +``` + +Recovery reads and clears the serial fault register, verifies device identity, +and requires error-free remote standstill, zero measured speed, and stable, +fault-free hatch and positioning status. A stationary positioning hold is +allowed if the target position is confirmed. Only successful checks clear the +flag; failed recovery checks leave motion blocked even if no flag was set before +the check. Recovery does not move hardware, restore program settings, or restart an +interrupted cycle. diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md index 3bbce5af02e..67c8bba32bd 100644 --- a/docs/user_guide/index.md +++ b/docs/user_guide/index.md @@ -36,6 +36,7 @@ byonoy/index cole_parmer/index curiox/index hamilton/index +hettich/index high_res/index inheco/index kbioscience/index diff --git a/docs/user_guide/machine-agnostic-features/event-bus.md b/docs/user_guide/machine-agnostic-features/event-bus.md index f78d2d5a913..e0633679ffc 100644 --- a/docs/user_guide/machine-agnostic-features/event-bus.md +++ b/docs/user_guide/machine-agnostic-features/event-bus.md @@ -130,6 +130,7 @@ events. | `legacy.centrifuge.Loader` | `centrifuge_loader.load`, `centrifuge_loader.unload` | | `agilent.vspin.VSpin` | `centrifuge.spin` | | `agilent.vspin.Access2` | `centrifuge_loader.load`, `centrifuge_loader.unload` | +| `hettich.HettichRoboticCentrifuge` | `centrifuge.spin` | | `brooks.precise_flex.PreciseFlex` | lifecycle, fault/home/freedrive, joint/cartesian/rail/gripper motion, pick/drop, park | | `manual_operator.ManualOperator` | arbitrary acknowledged manual actions; resource moves | @@ -141,6 +142,7 @@ Detailed operation references: - [Shaker and environmental controllers](event-bus/thermal-and-shaking.md) - [HighRes sample storage](../high_res/sample-storage/events.md) - [VSpin centrifuge and Access2 loader](../agilent/vspin/events.md) +- [Hettich robotic centrifuges](../hettich/events.md) - [Diagnostic transports](event-bus/diagnostic-transports.md) - [Canonical schema for every operation above](../../contributor_guide/event-schemas.md) - [Manual operator actions](manual-operator.md#eventbus-integration) diff --git a/pylabrobot/hettich/__init__.py b/pylabrobot/hettich/__init__.py new file mode 100644 index 00000000000..3a82af23a76 --- /dev/null +++ b/pylabrobot/hettich/__init__.py @@ -0,0 +1,33 @@ +from ._state import HettichMachineState +from .centrifuge import ( + MIKRO_220_ROBOTIC_ROTORS, + CentrifugeStatus, + HatchStatus, + HettichCentrifugeError, + HettichCommandError, + HettichCommunicationError, + HettichCooledRoboticCentrifuge, + HettichMikro220RoboticCentrifuge, + HettichRoboticCentrifuge, + HettichRotanta460RoboticCentrifuge, + HettichRotina380RoboticCentrifuge, + HettichRotina380RRoboticCentrifuge, + RotorSpecification, +) + +__all__ = [ + "CentrifugeStatus", + "HatchStatus", + "HettichCentrifugeError", + "HettichCommandError", + "HettichCommunicationError", + "HettichCooledRoboticCentrifuge", + "HettichMachineState", + "HettichMikro220RoboticCentrifuge", + "HettichRoboticCentrifuge", + "HettichRotanta460RoboticCentrifuge", + "HettichRotina380RRoboticCentrifuge", + "HettichRotina380RoboticCentrifuge", + "MIKRO_220_ROBOTIC_ROTORS", + "RotorSpecification", +] diff --git a/pylabrobot/hettich/_errors.py b/pylabrobot/hettich/_errors.py new file mode 100644 index 00000000000..f9b3435061e --- /dev/null +++ b/pylabrobot/hettich/_errors.py @@ -0,0 +1,32 @@ +"""Hettich protocol and workflow exceptions.""" + +_SIOF_MESSAGES = { + 0: "power on after reset or mains interruption", + 1: "serial parity error", + 2: "maximum allowed rotor cycles passed", + 3: "wrong BCC checksum", + 4: "framing error (wrong STX, ETX, ENQ, or '=')", + 5: "wrong or unknown parameter", + 6: "modification not permitted (read-only parameter)", + 7: "improper value or command not allowed", +} + + +class HettichCentrifugeError(Exception): + """Base exception raised by a Hettich robotic centrifuge.""" + + +class HettichCommunicationError(HettichCentrifugeError): + """The centrifuge returned no response or a malformed response.""" + + +class HettichCommandError(HettichCentrifugeError): + """The centrifuge rejected a parameter or command.""" + + def __init__(self, parameter: str, siof: int) -> None: + """Describe a rejected parameter using the device's SIOF fault bits.""" + self.parameter = parameter + self.siof = siof + messages = [_SIOF_MESSAGES[bit] for bit in range(8) if siof & (1 << bit)] + reason = ", ".join(messages) if messages else "no SIOF reason bit was set" + super().__init__(f"Hettich parameter {parameter} was rejected: {reason} (SIOF=0x{siof:02X})") diff --git a/pylabrobot/hettich/_state.py b/pylabrobot/hettich/_state.py new file mode 100644 index 00000000000..3aaf12a2d5c --- /dev/null +++ b/pylabrobot/hettich/_state.py @@ -0,0 +1,111 @@ +"""Hettich workflow ownership and lifecycle state, independent of serial I/O.""" + +import asyncio +from contextlib import asynccontextmanager +from dataclasses import dataclass, replace +from typing import AsyncIterator, Literal, Optional + +from ._errors import HettichCentrifugeError + +ConnectionState = Literal["disconnected", "connecting", "connected", "disconnecting", "unknown"] +Activity = Literal[ + "idle", + "setting_up", + "disconnecting", + "recovering", + "opening_hatch", + "closing_hatch", + "positioning", + "ending_positioning", + "selecting_program", + "preparing_to_spin", + "accelerating", + "at_speed", + "braking", + "stopping_spin", +] + + +@dataclass(frozen=True) +class HettichMachineState: + """PLR workflow state; physical rotor and hatch facts come from live status queries. + + ``recovery_required`` persists after a failed operation that may have changed + hardware or programmed settings. Only a successful ``recover()`` clears it. + """ + + connection: ConnectionState = "disconnected" + activity: Activity = "idle" + recovery_required: bool = False + + +@dataclass +class _Transition: + """Record whether the active operation may have changed the device.""" + + actuated: bool = False + + +class _StateMachine: + """Own workflow transitions without caching physical device status.""" + + def __init__(self) -> None: + """Create a disconnected machine with a lazily allocated operation lock.""" + self._state = HettichMachineState() + self._operation_lock: Optional[asyncio.Lock] = None + self._transition: Optional[_Transition] = None + + @property + def state(self) -> HettichMachineState: + """Return the current immutable snapshot.""" + return self._state + + def set_connection(self, connection: ConnectionState) -> None: + """Record transport lifecycle without clearing a recovery requirement.""" + self._state = replace(self.state, connection=connection) + + def set_activity(self, activity: Activity) -> None: + """Record the owning workflow's current phase.""" + self._state = replace(self.state, activity=activity) + + def require_recovery(self) -> None: + """Block ordinary motion until physical recovery checks succeed.""" + self._state = replace(self.state, recovery_required=True) + + def confirm_recovery(self) -> None: + """Record successful recovery checks performed by the device driver.""" + self._state = replace(self.state, recovery_required=False) + + def mark_actuated(self) -> None: + """Record possible device changes before transmitting a SELECT telegram.""" + if self._transition is not None: + self._transition.actuated = True + + @asynccontextmanager + async def operation( + self, activity: Activity, *, require_connected: bool = True, allow_recovery: bool = False + ) -> AsyncIterator[None]: + """Reserve a complete workflow and retain uncertainty after possible actuation.""" + if self._operation_lock is None: + self._operation_lock = asyncio.Lock() + if self._operation_lock.locked(): + raise HettichCentrifugeError(f"Cannot {activity} while {self.state.activity} is active") + async with self._operation_lock: + if require_connected and self.state.connection != "connected": + raise HettichCentrifugeError("The centrifuge is not connected; call setup() first") + if self.state.recovery_required and not allow_recovery: + raise HettichCentrifugeError( + "The centrifuge requires recovery; call recover() before motion" + ) + transition = _Transition() + self._transition = transition + self._state = replace(self.state, activity=activity) + try: + yield + except BaseException: + if transition.actuated: + self._state = replace(self.state, recovery_required=True) + raise + finally: + self._transition = None + self._state = replace(self.state, activity="idle") diff --git a/pylabrobot/hettich/centrifuge.py b/pylabrobot/hettich/centrifuge.py new file mode 100644 index 00000000000..47a5fc11439 --- /dev/null +++ b/pylabrobot/hettich/centrifuge.py @@ -0,0 +1,1198 @@ +"""Serial driver for Generation 2 Hettich robotic centrifuges.""" + +import asyncio +import logging +import math +from abc import ABC, abstractmethod +from dataclasses import dataclass +from time import monotonic +from typing import Any, Literal, Mapping, Optional + +from pylabrobot.events import device_reference, event_operation +from pylabrobot.io.serial import Serial + +from ._errors import HettichCentrifugeError, HettichCommandError, HettichCommunicationError +from ._state import HettichMachineState, _StateMachine + +logger = logging.getLogger(__name__) + +EOT = 0x04 +STX = 0x02 +ETX = 0x03 +ENQ = 0x05 +ACK = 0x06 +NAK = 0x15 + +ENQUIRY_REPLY_LENGTH = 14 +COMMAND_REPLY_LENGTH = 2 +MINIMUM_ENQUIRY_INTERVAL = 0.4 +RUN_TIME_RESOLUTION = 1.0 + +# Protocol source: Hettich document AH5680-01EN. +# https://www.hettweb.com/wp-content/uploads/2019/09/OM-ROBOTIC-CENTRIFUGE-COMMUNICATION-PARAMETERS-AH5680-01EN.pdf +SIOF_PARAMETER = "00685" +GENERATION_PARAMETER = "00600" +DEVICE_TYPE_PARAMETER = "00537" +SOFTWARE_VERSION_PARAMETER = "00636" +RUN_TIME_PARAMETER = "00601" +ACTUAL_RUN_TIME_PARAMETER = "00602" +SPEED_PARAMETER = "00603" +ACTUAL_SPEED_PARAMETER = "00604" +MAXIMUM_SPEED_PARAMETER = "00605" +MAXIMUM_RCF_PARAMETER = "00608" +MAXIMUM_RUN_UP_TIME_PARAMETER = "00614" +MAXIMUM_RUN_DOWN_TIME_PARAMETER = "00616" +ACTIVATE_PARAMETERS_COMMAND = "00522" +SPIN_COMMAND = "00521" +TARGET_POSITION_PARAMETER = "00524" +POSITION_COMMAND = "00526" +HATCH_STATUS_PARAMETER = "00528" +PROGRAM_COMMAND = "00523" +STATUS_1_PARAMETER = "00634" +STATUS_2_PARAMETER = "00635" +TEMPERATURE_PARAMETER = "00619" + +MINIMUM_SPEED = 50 +MAXIMUM_DURATION = 59_999 +DEFAULT_TIMEOUT_MARGIN = 30 +GENERATION_2_IDENTIFICATION = 0x1234 + +_VALID_ADDRESSES = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]") + +_KNOWN_DEVICE_TYPES = { + 0xE800: "MIKRO 220 POS", + 0xC901: "ROTANTA 460 R POS", + 0xC000: "ROTANTA 460", + 0x8800: "ROTINA 380 POS", + 0x8901: "ROTINA 380 R POS", + 0x8904: "ROTINA 380", + 0xE801: "MIKRO 220 POS", +} + +SpinPhase = Literal["standstill", "accelerating", "centrifuging", "braking", "unknown"] +KeyLockState = Literal["teach", "remote", "middle", "software_lock_4", "software_lock_5", "unknown"] +PositioningSpeed = Literal["slow", "fast"] + + +@dataclass(frozen=True) +class RotorSpecification: + """Manufacturer limits for a Hettich robotic centrifuge rotor. + + Attributes: + catalog_number: Hettich rotor catalog number. + positions: Number of tube positions. + maximum_volume: Maximum tube volume in microliters. + maximum_speed: Maximum rotor speed in rpm. + maximum_rcf: Maximum relative centrifugal force in multiples of gravity. + """ + + catalog_number: str + positions: int + maximum_volume: int + maximum_speed: int + maximum_rcf: int + + def rpm_to_g(self, rpm: int) -> float: + """Return the relative centrifugal force at ``rpm`` in multiples of standard gravity.""" + if not 0 <= rpm <= self.maximum_speed: + raise ValueError(f"rpm must be 0..{self.maximum_speed} rpm") + return self.maximum_rcf * (rpm / self.maximum_speed) ** 2 + + def g_to_rpm(self, g: float) -> int: + """Return the nearest integer RPM that produces ``g`` multiples of standard gravity.""" + if not 0 <= g <= self.maximum_rcf: + raise ValueError(f"g must be 0..{self.maximum_rcf} × g") + return round(self.maximum_speed * math.sqrt(g / self.maximum_rcf)) + + +# Source: https://www.hettichlab.com/products/centrifuges/automated-centrifuges/mikro-220-robotic/ +MIKRO_220_ROBOTIC_ROTORS: Mapping[str, RotorSpecification] = { + "2334": RotorSpecification( + catalog_number="2334", + positions=24, + maximum_volume=2_000, + maximum_speed=13_000, + maximum_rcf=18_327, + ), + "2394": RotorSpecification( + catalog_number="2394", + positions=24, + maximum_volume=2_000, + maximum_speed=13_000, + maximum_rcf=18_516, + ), +} + + +@dataclass(frozen=True) +class _ModelConfiguration: + """Exact protocol identity and capabilities for one centrifuge model.""" + + name: str + device_type_codes: frozenset[int] + rotor_specifications: Mapping[str, RotorSpecification] + physically_verified: bool + + +_MIKRO_220_ROBOTIC_CONFIGURATION = _ModelConfiguration( + name="MIKRO 220 POS", + device_type_codes=frozenset((0xE800, 0xE801)), + rotor_specifications=MIKRO_220_ROBOTIC_ROTORS, + physically_verified=True, +) + +_ROTANTA_460_ROBOTIC_CONFIGURATION = _ModelConfiguration( + name="ROTANTA 460 R POS", + device_type_codes=frozenset((0xC901,)), + rotor_specifications={}, + physically_verified=False, +) + +_ROTINA_380_ROBOTIC_CONFIGURATION = _ModelConfiguration( + name="ROTINA 380 POS", + device_type_codes=frozenset((0x8800,)), + rotor_specifications={}, + physically_verified=False, +) + +_ROTINA_380_R_ROBOTIC_CONFIGURATION = _ModelConfiguration( + name="ROTINA 380 R POS", + device_type_codes=frozenset((0x8901,)), + rotor_specifications={}, + physically_verified=False, +) + + +@dataclass(frozen=True) +class CentrifugeStatus: + """Decoded values from Hettich parameters 00634 and 00635.""" + + phase: SpinPhase + can_start: bool + status_changed: bool + error_number: Optional[int] + program_number: Optional[int] + rotor_number: int + key_lock: KeyLockState + key_lock_code: int + lid_closed: bool + rotor_cycle_counter_enabled: bool + maximum_rotor_cycles_exceeded: bool + rotor_cycle_limit_confirmed: bool + rotor_changed: bool + no_rotor: bool + + +@dataclass(frozen=True) +class HatchStatus: + """Decoded positioning and hatch values from Hettich parameter 00528.""" + + hatch_open: bool + hatch_closed: bool + lid_lock_closed: bool + hatch_moving: bool + hatch_opening: bool + hatch_closing: bool + hatch_timeout: bool + magnetic_brake_implemented: bool + magnetic_brake_active: bool + position_reached: bool + positioning_active: bool + rotor_moving: bool + positioning_timeout: bool + positioning_error: bool + + +class HettichRoboticCentrifuge(ABC): + """Shared Generation 2 protocol for a model-specific Hettich centrifuge. + + The protocol is shared by Hettich robotic centrifuges with the C control panel, + including the MIKRO 220 POS and ROTANTA 460 R POS. The connection uses 9600 + baud, 7 data bits, even parity, one stop bit, and no flow control. Values are + four hexadecimal ASCII digits inside addressed ENQUIRY and SELECT telegrams. + + ``setup()`` only reads the serial fault register and identification fields; it + does not move the rotor or hatch. Motion methods verify the reported machine + state before transmitting a command. + + Instantiate a concrete model class rather than this abstract protocol class. + Each subclass declares exact device-type codes and model-specific capabilities. + """ + + @classmethod + @abstractmethod + def _configuration(cls) -> _ModelConfiguration: + """Return the exact identity and capabilities declared by this model class.""" + + def __init__( + self, + port: str, + address: str = "]", + timeout: float = 0.2, + retries: int = 3, + poll_interval: float = 0.5, + rotor_catalog_number: Optional[str] = None, + name: str = "centrifuge", + ) -> None: + """Create a Hettich robotic centrifuge connection. + + Args: + port: Serial port connected to the centrifuge. + address: One-character Hettich bus address. The factory default is ``]``. + timeout: Per-read timeout in seconds. The manual specifies a maximum + response time of 150 ms. + retries: Total transmission attempts after a timeout or invalid reply. + poll_interval: Additional delay between polling iterations while waiting for + motion. All ENQUIRY telegrams have a minimum 0.4-second interval. + rotor_catalog_number: Hettich catalog number for the installed rotor. The + concrete model class defines which catalog numbers are supported. + name: Stable name used to identify this centrifuge in events. + """ + if address not in _VALID_ADDRESSES: + raise ValueError("address must be one of A-Z, [, \\, or ]") + if timeout < 0.15: + raise ValueError("timeout must be at least 0.15 seconds") + if retries < 1: + raise ValueError("retries must be at least 1") + if poll_interval < 0: + raise ValueError("poll_interval must be non-negative") + configuration = self._configuration() + if ( + rotor_catalog_number is not None + and rotor_catalog_number not in configuration.rotor_specifications + ): + supported = ", ".join(configuration.rotor_specifications) + if supported: + raise ValueError(f"unsupported rotor catalog number; expected one of {supported}") + raise ValueError(f"rotor specifications are not available for {configuration.name}") + + self.name = name + self.address = address + self.retries = retries + self.poll_interval = poll_interval + self.rotor_specification = ( + configuration.rotor_specifications[rotor_catalog_number] + if rotor_catalog_number is not None + else None + ) + self.io = Serial( + human_readable_device_name=f"Hettich {configuration.name}", + port=port, + baudrate=9600, + bytesize=7, + parity="E", + stopbits=1, + timeout=timeout, + write_timeout=timeout, + rtscts=False, + dsrdtr=False, + xonxoff=False, + ) + self._transaction_lock: Optional[asyncio.Lock] = None + self._machine = _StateMachine() + self._spin_stop_requested = False + self._spin_completion: Optional[asyncio.Event] = None + self._next_enquiry_at = 0.0 + self.device_type_code: Optional[int] = None + self.device_type: Optional[str] = None + self.software_version: Optional[str] = None + + @property + def state(self) -> HettichMachineState: + """Return an immutable snapshot of connection, active workflow, and recovery state.""" + return self._machine.state + + async def setup(self) -> None: + """Connect and identify the device, preserving any outstanding recovery requirement.""" + async with self._machine.operation("setting_up", require_connected=False, allow_recovery=True): + if self.state.connection == "connected": + return + if self.state.connection == "unknown": + raise HettichCentrifugeError("Transport closure is uncertain; call stop() before setup()") + self._machine.set_connection("connecting") + try: + await self._setup() + except BaseException: + if self.state.connection != "disconnected": + self._machine.set_connection("unknown") + raise + self._machine.set_connection("connected") + + async def _setup(self) -> None: + """Open the port and verify a Generation 2 Hettich centrifuge without moving it.""" + await self.io.setup() + try: + await self.io.reset_input_buffer() + await self.io.reset_output_buffer() + + siof = await self._enquire_parameter(SIOF_PARAMETER, allow_nak=False) + if siof: + logger.info("[Hettich %s] cleared startup SIOF=0x%02X", self.io.port, siof) + + generation = await self._enquire_parameter(GENERATION_PARAMETER) + if generation != GENERATION_2_IDENTIFICATION: + raise HettichCentrifugeError( + "The connected centrifuge did not identify as Hettich Generation 2 " + f"(received 0x{generation:04X})" + ) + + self.device_type_code = await self._enquire_parameter(DEVICE_TYPE_PARAMETER) + configuration = self._configuration() + if self.device_type_code not in configuration.device_type_codes: + connected_type = _KNOWN_DEVICE_TYPES.get( + self.device_type_code, f"unknown type 0x{self.device_type_code:04X}" + ) + expected_codes = ", ".join( + f"0x{code:04X}" for code in sorted(configuration.device_type_codes) + ) + raise HettichCentrifugeError( + f"{type(self).__name__} requires {configuration.name} ({expected_codes}), " + f"but the connected centrifuge reports {connected_type} " + f"(0x{self.device_type_code:04X})" + ) + self.device_type = configuration.name + software = await self._enquire_parameter(SOFTWARE_VERSION_PARAMETER) + software_digits = f"{software:04X}" + self.software_version = f"{software_digits[:2]}.{software_digits[2:]}" + + if not configuration.physically_verified: + logger.warning( + "%s support for %s has not been physically verified in PyLabRobot", + type(self).__name__, + self.device_type, + ) + + logger.info( + "[Hettich %s] connected: %s, software %s, address %s", + self.io.port, + self.device_type, + self.software_version, + self.address, + ) + except BaseException: + await self.io.stop() + self._machine.set_connection("disconnected") + raise + + async def stop(self) -> None: + """Close the serial connection without changing the centrifuge's run state.""" + async with self._machine.operation( + "disconnecting", require_connected=False, allow_recovery=True + ): + if self.state.connection == "disconnected": + return + self._machine.set_connection("disconnecting") + async with self._get_transaction_lock(): + try: + await self.io.stop() + except BaseException: + self._machine.set_connection("unknown") + self._machine.require_recovery() + raise + self._machine.set_connection("disconnected") + + async def recover(self) -> None: + """Clear the recovery flag after fresh, read-only checks of identity and safe standstill. + + Resolve the cause of the failure and inspect the samples first. This method + does not move the rotor or hatch, restart a cycle, or restore program settings. + Reconnect with ``setup()`` first if the transport was closed. + """ + async with self._machine.operation("recovering", allow_recovery=True): + self._machine.require_recovery() + await self._enquire_parameter(SIOF_PARAMETER, allow_nak=False) + generation = await self._enquire_parameter(GENERATION_PARAMETER) + device_type = await self._enquire_parameter(DEVICE_TYPE_PARAMETER) + if ( + generation != GENERATION_2_IDENTIFICATION + or device_type not in self._configuration().device_type_codes + ): + raise HettichCentrifugeError("The connected centrifuge identity does not match this driver") + self._require_remote_standstill(await self.request_status()) + if await self.request_speed() != 0: + raise HettichCentrifugeError("Recovery requires zero measured rotor speed") + hatch = await self.request_hatch_status() + if ( + hatch.hatch_moving + or hatch.rotor_moving + or hatch.hatch_timeout + or hatch.positioning_timeout + or hatch.positioning_error + or hatch.hatch_open == hatch.hatch_closed + or (hatch.hatch_closed and not hatch.lid_lock_closed) + or (hatch.positioning_active and not hatch.position_reached) + ): + raise HettichCentrifugeError( + "Recovery requires stable, fault-free hatch and positioning state" + ) + self._machine.confirm_recovery() + logger.info("[Hettich %s] recovery checks passed", self.io.port) + + @staticmethod + def _bcc(data: bytes) -> int: + """Return the XOR block-check character for bytes after STX through ETX.""" + checksum = 0 + for byte in data: + checksum ^= byte + return checksum + + def _build_enquiry(self, parameter: str) -> bytes: + """Build an eight-byte ENQUIRY telegram.""" + self._validate_parameter(parameter) + return bytes([EOT, ord(self.address)]) + parameter.encode("ascii") + bytes([ENQ]) + + def _build_select(self, parameter: str, value: int) -> bytes: + """Build a 15-byte SELECT telegram with its BCC.""" + self._validate_parameter(parameter) + if not 0 <= value <= 0xFFFF: + raise ValueError("parameter value must be 0..65535") + body = bytes([STX]) + parameter.encode("ascii") + f"={value:04X}".encode("ascii") + bytes([ETX]) + return bytes([EOT, ord(self.address)]) + body + bytes([self._bcc(body[1:])]) + + @staticmethod + def _validate_parameter(parameter: str) -> None: + """Require a five-digit parameter identifier accepted by the protocol.""" + if len(parameter) != 5 or not parameter.isascii() or not parameter.isdigit(): + raise ValueError("parameter must contain exactly five ASCII digits") + + async def _read_exact(self, length: int) -> bytes: + """Read exactly ``length`` bytes, raising when the serial timeout expires.""" + data = bytearray() + while len(data) < length: + chunk = await self.io.read(length - len(data)) + if not chunk: + break + data.extend(chunk) + if len(data) != length: + raise HettichCommunicationError(f"expected {length} reply bytes, received {len(data)}") + return bytes(data) + + def _parse_enquiry_reply(self, reply: bytes, parameter: str) -> int: + """Validate and decode a 14-byte ENQUIRY reply.""" + if len(reply) != ENQUIRY_REPLY_LENGTH: + raise HettichCommunicationError( + f"expected {ENQUIRY_REPLY_LENGTH} reply bytes, received {len(reply)}" + ) + if reply[0] != ord(self.address): + raise HettichCommunicationError( + f"reply address {chr(reply[0])!r} did not match {self.address!r}" + ) + if reply[1] != STX or reply[7] != ord("=") or reply[12] != ETX: + raise HettichCommunicationError(f"malformed reply for parameter {parameter}: {reply!r}") + reply_parameter = reply[2:7].decode("ascii", errors="replace") + if reply_parameter != parameter: + raise HettichCommunicationError( + f"reply parameter {reply_parameter!r} did not match {parameter!r}" + ) + expected_bcc = self._bcc(reply[2:13]) + if reply[13] != expected_bcc: + raise HettichCommunicationError( + f"wrong BCC for parameter {parameter}: received 0x{reply[13]:02X}, " + f"expected 0x{expected_bcc:02X}" + ) + raw_value = reply[8:12] + if any(byte not in b"0123456789ABCDEF" for byte in raw_value): + raise HettichCommunicationError(f"parameter {parameter} returned a non-hexadecimal value") + return int(raw_value.decode("ascii"), 16) + + async def _request_enquiry(self, parameter: str) -> Optional[int]: + """Transmit a paced ENQUIRY under the transaction lock; return ``None`` for NAK. + + The manual requires 400 ms between enquiries during centrifugation. Apply + that interval in every state because a run can also be started at the panel. + """ + frame = self._build_enquiry(parameter) + last_error: Optional[HettichCommunicationError] = None + for attempt in range(1, self.retries + 1): + delay = self._next_enquiry_at - monotonic() + if delay > 0: + await asyncio.sleep(delay) + try: + await self.io.write(frame) + prefix = await self._read_exact(COMMAND_REPLY_LENGTH) + if prefix == bytes([ord(self.address), NAK]): + return None + reply = prefix + await self._read_exact(ENQUIRY_REPLY_LENGTH - len(prefix)) + return self._parse_enquiry_reply(reply, parameter) + except HettichCommunicationError as exc: + last_error = exc + logger.warning( + "[Hettich %s] ENQUIRY %s attempt %d/%d failed: %s", + self.io.port, + parameter, + attempt, + self.retries, + exc, + ) + finally: + try: + await self.io.write(bytes([EOT])) + finally: + self._next_enquiry_at = monotonic() + MINIMUM_ENQUIRY_INTERVAL + assert last_error is not None + raise HettichCommunicationError( + f"ENQUIRY {parameter} failed after {self.retries} attempts: {last_error}" + ) from last_error + + async def _request_select(self, parameter: str, value: int) -> bool: + """Transmit a SELECT and return whether the centrifuge acknowledged it.""" + frame = self._build_select(parameter, value) + last_error: Optional[HettichCommunicationError] = None + for attempt in range(1, self.retries + 1): + try: + self._machine.mark_actuated() + await self.io.write(frame) + reply = await self._read_exact(COMMAND_REPLY_LENGTH) + if reply[0] != ord(self.address): + raise HettichCommunicationError( + f"reply address {chr(reply[0])!r} did not match {self.address!r}" + ) + if reply[1] == ACK: + return True + if reply[1] == NAK: + return False + raise HettichCommunicationError(f"invalid acknowledgement: {reply!r}") + except HettichCommunicationError as exc: + last_error = exc + logger.warning( + "[Hettich %s] SELECT %s attempt %d/%d failed: %s", + self.io.port, + parameter, + attempt, + self.retries, + exc, + ) + finally: + await self.io.write(bytes([EOT])) + assert last_error is not None + raise HettichCommunicationError( + f"SELECT {parameter} failed after {self.retries} attempts: {last_error}" + ) from last_error + + def _get_transaction_lock(self) -> asyncio.Lock: + """Return the serial transaction lock, creating it inside the active event loop.""" + if self._transaction_lock is None: + self._transaction_lock = asyncio.Lock() + return self._transaction_lock + + async def _enquire_parameter(self, parameter: str, allow_nak: bool = True) -> int: + """Read a parameter while serializing access to the bus.""" + async with self._get_transaction_lock(): + if self.state.connection not in ("connected", "connecting"): + raise HettichCentrifugeError("The centrifuge is not connected; call setup() first") + value = await self._request_enquiry(parameter) + if value is not None: + return value + if not allow_nak or parameter == SIOF_PARAMETER: + raise HettichCommunicationError(f"ENQUIRY {parameter} was rejected") + siof = await self._request_enquiry(SIOF_PARAMETER) + if siof is None: + raise HettichCommunicationError("SIOF enquiry was rejected after NAK") + raise HettichCommandError(parameter, siof) + + async def _select_parameter(self, parameter: str, value: int) -> None: + """Write a parameter and decode SIOF when the centrifuge returns NAK.""" + async with self._get_transaction_lock(): + if self.state.connection != "connected": + raise HettichCentrifugeError("The centrifuge is not connected; call setup() first") + acknowledged = await self._request_select(parameter, value) + if acknowledged: + return + siof = await self._request_enquiry(SIOF_PARAMETER) + if siof is None: + raise HettichCommunicationError("SIOF enquiry was rejected after NAK") + raise HettichCommandError(parameter, siof) + + @staticmethod + def _phase(status_byte: int) -> SpinPhase: + """Decode the motion phase bits in the low byte of parameter 00634.""" + if status_byte & 0x10: + return "braking" + if status_byte & 0x08: + return "centrifuging" + if status_byte & 0x04: + return "accelerating" + if status_byte & 0x02: + return "standstill" + return "unknown" + + @staticmethod + def _key_lock(code: int) -> KeyLockState: + """Decode the key-switch or software-lock code from parameter 00635.""" + states: dict[int, KeyLockState] = { + 1: "teach", + 2: "remote", + 3: "middle", + 4: "software_lock_4", + 5: "software_lock_5", + } + return states.get(code, "unknown") + + async def request_status(self) -> CentrifugeStatus: + """Read and decode centrifuge state, program/error, rotor, lid, and key lock.""" + state_1 = await self._enquire_parameter(STATUS_1_PARAMETER) + state_2 = await self._enquire_parameter(STATUS_2_PARAMETER) + state_byte = state_1 & 0xFF + program_or_error = state_1 >> 8 + state_2_high = state_2 >> 8 + state_2_low = state_2 & 0xFF + has_error = bool(program_or_error & 0x80) + key_lock_code = state_2_low & 0x07 + return CentrifugeStatus( + phase=self._phase(state_byte), + can_start=not bool(state_byte & 0x01), + status_changed=bool(state_byte & 0x80), + error_number=(program_or_error & 0x7F) if has_error else None, + program_number=None if has_error else program_or_error, + rotor_number=(state_2_low >> 4) & 0x0F, + key_lock=self._key_lock(key_lock_code), + key_lock_code=key_lock_code, + lid_closed=bool(state_2_high & 0x02), + rotor_cycle_counter_enabled=bool(state_2_high & 0x80), + maximum_rotor_cycles_exceeded=bool(state_2_high & 0x40), + rotor_cycle_limit_confirmed=bool(state_2_high & 0x20), + rotor_changed=bool(state_2_high & 0x08), + no_rotor=bool(state_2_high & 0x04), + ) + + async def request_hatch_status(self) -> HatchStatus: + """Read and decode hatch and positioning state.""" + value = await self._enquire_parameter(HATCH_STATUS_PARAMETER) + hatch = value >> 8 + position = value & 0xFF + return HatchStatus( + hatch_open=bool(hatch & 0x20), + hatch_closed=bool(hatch & 0x10), + lid_lock_closed=bool(hatch & 0x08), + hatch_moving=bool(hatch & 0x04), + hatch_opening=bool(hatch & 0x02), + hatch_closing=bool(hatch & 0x01), + hatch_timeout=bool(hatch & 0x40), + magnetic_brake_implemented=bool(hatch & 0x80), + magnetic_brake_active=bool(position & 0x20), + position_reached=bool(position & 0x04), + positioning_active=bool(position & 0x02), + rotor_moving=bool(position & 0x01), + positioning_timeout=bool(position & 0x08), + positioning_error=bool(position & 0x10), + ) + + async def request_speed(self) -> int: + """Return the actual rotor speed in rpm.""" + return await self._enquire_parameter(ACTUAL_SPEED_PARAMETER) + + async def request_maximum_speed(self) -> int: + """Return the installed rotor's maximum speed in rpm.""" + return await self._enquire_parameter(MAXIMUM_SPEED_PARAMETER) + + def rpm_to_g(self, rpm: int) -> float: + """Convert RPM to multiples of standard gravity using the configured rotor specification. + + Raises: + HettichCentrifugeError: No rotor catalog number was supplied at construction. + ValueError: ``rpm`` is outside the rotor's manufacturer limits. + """ + if self.rotor_specification is None: + raise HettichCentrifugeError( + "Set rotor_catalog_number to calculate RCF from the manufacturer rotor table" + ) + return self.rotor_specification.rpm_to_g(rpm) + + def g_to_rpm(self, g: float) -> int: + """Convert multiples of standard gravity to RPM using the configured rotor specification. + + Raises: + HettichCentrifugeError: No rotor catalog number was supplied at construction. + ValueError: ``g`` is outside the rotor's manufacturer limits. + """ + if self.rotor_specification is None: + raise HettichCentrifugeError( + "Set rotor_catalog_number to calculate speed from the manufacturer rotor table" + ) + return self.rotor_specification.g_to_rpm(g) + + async def request_elapsed_time(self) -> int: + """Return the current run time in seconds.""" + return await self._enquire_parameter(ACTUAL_RUN_TIME_PARAMETER) + + @staticmethod + def _require_remote_standstill(status: CentrifugeStatus) -> None: + """Require remote control, standstill, and no reported centrifuge error.""" + if status.key_lock not in ("remote", "software_lock_4", "software_lock_5"): + raise HettichCentrifugeError("The key-operated switch must be in LOCK 2 before remote motion") + if status.error_number is not None: + raise HettichCentrifugeError( + f"The centrifuge reports error {status.error_number}; resolve it before motion" + ) + if status.phase != "standstill": + raise HettichCentrifugeError( + f"The centrifuge must be at standstill, but reports {status.phase}" + ) + + @staticmethod + def _require_positioning_ready(status: CentrifugeStatus) -> None: + """Require the machine state mandated for hatch and positioning commands.""" + HettichRoboticCentrifuge._require_remote_standstill(status) + if not status.lid_closed: + raise HettichCentrifugeError( + "Close and lock the main centrifuge lid before hatch or rotor positioning motion" + ) + + async def _wait_for_hatch(self, desired: Literal["open", "closed"], timeout: float) -> None: + """Wait until the hatch reaches ``desired`` or reports a positioning fault.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + hatch = await self.request_hatch_status() + if hatch.hatch_timeout or hatch.positioning_timeout or hatch.positioning_error: + raise HettichCentrifugeError("The centrifuge reported a hatch or positioning error") + if desired == "open" and hatch.hatch_open: + return + if desired == "closed" and hatch.hatch_closed and hatch.lid_lock_closed: + return + if loop.time() >= deadline: + raise TimeoutError(f"Hettich hatch did not reach {desired} within {timeout} seconds") + await asyncio.sleep(self.poll_interval) + + async def open_hatch(self, timeout: float = 30.0) -> None: + """Move the loading hatch to the open state and wait for confirmation.""" + async with self._machine.operation("opening_hatch"): + hatch = await self.request_hatch_status() + if hatch.hatch_open: + return + self._require_positioning_ready(await self.request_status()) + logger.info("[Hettich %s] opening hatch", self.io.port) + await self._select_parameter(POSITION_COMMAND, 0x0060) + await self._wait_for_hatch("open", timeout) + + async def close_hatch(self, timeout: float = 30.0) -> None: + """Move the loading hatch to the closed state and wait for both closed switches.""" + async with self._machine.operation("closing_hatch"): + hatch = await self.request_hatch_status() + if hatch.hatch_closed and hatch.lid_lock_closed: + return + self._require_positioning_ready(await self.request_status()) + logger.info("[Hettich %s] closing hatch", self.io.port) + await self._select_parameter(POSITION_COMMAND, 0x0070) + await self._wait_for_hatch("closed", timeout) + + async def move_to_position( + self, + position: int, + speed: PositioningSpeed = "slow", + timeout: float = 320.0, + ) -> None: + """Move a rotor bucket beneath the loading hatch and hold it there. + + Args: + position: One-based rotor position. + speed: ``"slow"`` for agitation-sensitive samples or ``"fast"``. + timeout: Maximum total positioning time. Firmware may make three attempts + of up to 100 seconds each. + """ + async with self._machine.operation("positioning"): + if speed not in ("slow", "fast"): + raise ValueError('speed must be "slow" or "fast"') + current_target = await self._enquire_parameter(TARGET_POSITION_PARAMETER) + maximum_positions = current_target >> 8 + if not 1 <= position <= maximum_positions: + raise ValueError(f"position must be 1..{maximum_positions} for the installed rotor") + + hatch = await self.request_hatch_status() + self._require_positioning_ready(await self.request_status()) + if current_target & 0xFF == position and hatch.position_reached: + return + + if hatch.rotor_moving: + if current_target & 0xFF != position: + raise HettichCentrifugeError( + "A different rotor positioning move is already active; wait for it to finish" + ) + else: + if current_target & 0xFF != position: + await self._select_parameter( + TARGET_POSITION_PARAMETER, (maximum_positions << 8) | position + ) + logger.info("[Hettich %s] moving to rotor position %d (%s)", self.io.port, position, speed) + await self._select_parameter(POSITION_COMMAND, 0x0001 if speed == "slow" else 0x0002) + + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + state = await self.request_hatch_status() + if state.positioning_timeout or state.positioning_error: + raise HettichCentrifugeError(f"Positioning rotor at position {position} failed") + if state.position_reached: + return + if loop.time() >= deadline: + raise TimeoutError(f"Rotor did not reach position {position} within {timeout} seconds") + await asyncio.sleep(self.poll_interval) + + async def _wait_for_positioning_end(self, timeout: float) -> None: + """Wait until the positioning-active state clears.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + hatch = await self.request_hatch_status() + if hatch.positioning_timeout or hatch.positioning_error: + raise HettichCentrifugeError("The centrifuge reported a positioning error") + if not hatch.positioning_active: + return + if loop.time() >= deadline: + raise TimeoutError(f"Positioning mode did not end within {timeout} seconds") + await asyncio.sleep(self.poll_interval) + + async def end_positioning(self, timeout: float = 10.0) -> None: + """Leave positioning mode if it is active, readying the centrifuge for a run.""" + async with self._machine.operation("ending_positioning"): + if not (await self.request_hatch_status()).positioning_active: + return + self._require_positioning_ready(await self.request_status()) + await self._select_parameter(POSITION_COMMAND, 0x0080) + await self._wait_for_positioning_end(timeout) + + async def select_program(self, program: int) -> None: + """Select a stored program without starting centrifugation. + + Args: + program: Stored program number, from 1 through 89. + """ + async with self._machine.operation("selecting_program"): + if not 1 <= program <= 89: + raise ValueError("program must be 1..89") + status = await self.request_status() + self._require_remote_standstill(status) + if status.program_number == program: + return + await self._select_parameter(PROGRAM_COMMAND, (program << 8) | 0x04) + logger.info("[Hettich %s] selected program %d", self.io.port, program) + + async def _start_spin(self, run_time: int, speed: int) -> None: + """Start without waiting, using the device-native acceleration-inclusive run time.""" + if not 1 <= run_time <= MAXIMUM_DURATION: + raise ValueError(f"run_time must be 1..{MAXIMUM_DURATION} seconds") + if speed < MINIMUM_SPEED: + raise ValueError(f"speed must be at least {MINIMUM_SPEED} rpm") + self._raise_if_spin_stop_requested() + + status = await self.request_status() + self._require_remote_standstill(status) + if not status.lid_closed: + raise HettichCentrifugeError("Close and lock the main centrifuge lid before a run") + if status.no_rotor: + raise HettichCentrifugeError("The centrifuge does not detect an installed rotor") + if status.maximum_rotor_cycles_exceeded: + raise HettichCentrifugeError("The installed rotor's maximum cycle count is exceeded") + hatch = await self.request_hatch_status() + if not hatch.hatch_closed or not hatch.lid_lock_closed: + raise HettichCentrifugeError("The hatch and lid lock must both be closed before a run") + if hatch.positioning_active: + self._raise_if_spin_stop_requested() + await self._select_parameter(POSITION_COMMAND, 0x0080) + await self._wait_for_positioning_end(timeout=10.0) + status = await self.request_status() + self._require_remote_standstill(status) + if not status.can_start: + raise HettichCentrifugeError("The centrifuge reports that centrifugation is not possible") + + maximum_speed = await self.request_maximum_speed() + if speed > maximum_speed: + raise ValueError(f"speed must not exceed the installed rotor limit of {maximum_speed} rpm") + + self._raise_if_spin_stop_requested() + await self._select_parameter(RUN_TIME_PARAMETER, run_time) + await self._select_parameter(SPEED_PARAMETER, speed) + await self._select_parameter(ACTIVATE_PARAMETERS_COMMAND, 0x0001) + + self._raise_if_spin_stop_requested() + + logger.info( + "[Hettich %s] starting centrifugation: run_time=%d seconds, speed=%d rpm", + self.io.port, + run_time, + speed, + ) + try: + await self._select_parameter(SPIN_COMMAND, 0x0002) + except BaseException: + await self._stop_after_spin_failure() + raise + + async def _stop_after_spin_failure(self) -> None: + """Attempt to stop a possibly running rotor without masking the original failure.""" + if self.state.activity in ("preparing_to_spin", "accelerating", "at_speed", "braking"): + self._machine.set_activity("braking") + try: + await self._stop_spin() + except BaseException: + logger.exception("[Hettich %s] failed to stop after spin() failed", self.io.port) + + def _raise_if_spin_stop_requested(self) -> None: + """Let the spin owner handle an external stop request through its cleanup path.""" + if self._spin_stop_requested: + raise HettichCentrifugeError("Centrifugation was interrupted by stop_spin()") + + async def _wait_for_standstill( + self, + timeout: float, + motion_observed: bool, + expected_end_at: Optional[float] = None, + ) -> CentrifugeStatus: + """Wait for standstill and optionally reject a prematurely ended timed run. + + ``expected_end_at`` is a monotonic timestamp for the earliest expected end + of centrifugation. Allow one second for device timer rounding. Latch early + braking or standstill when observed, then report interruption at standstill. + """ + if timeout <= 0: + raise ValueError("timeout must be positive") + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + interrupted = False + while True: + if expected_end_at is not None: + self._raise_if_spin_stop_requested() + status = await self.request_status() + if expected_end_at is not None and status.phase == "braking": + self._machine.set_activity("braking") + if status.error_number is not None: + raise HettichCentrifugeError(f"The centrifuge stopped with error {status.error_number}") + if ( + expected_end_at is not None + and status.phase in ("braking", "standstill") + and monotonic() + RUN_TIME_RESOLUTION < expected_end_at + ): + interrupted = True + if status.phase != "standstill": + motion_observed = True + elif motion_observed: + if interrupted: + raise HettichCentrifugeError( + "Centrifugation was interrupted before the requested duration at target speed elapsed" + ) + return status + if loop.time() >= deadline: + raise TimeoutError(f"Centrifuge did not return to standstill within {timeout} seconds") + await asyncio.sleep(self.poll_interval) + + async def _wait_for_target_speed(self, speed: int, timeout: float) -> tuple[int, float]: + """Return elapsed run time at ``speed`` and a monotonic lower bound on its sample time.""" + if timeout <= 0: + raise ValueError("timeout must be positive") + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + motion_observed = False + while True: + self._raise_if_spin_stop_requested() + status = await self.request_status() + if status.error_number is not None: + raise HettichCentrifugeError(f"The centrifuge stopped with error {status.error_number}") + actual_speed = await self.request_speed() + if status.phase != "standstill" or actual_speed > 0: + motion_observed = True + if status.phase == "centrifuging" and actual_speed >= speed: + sample_started_at = monotonic() + return await self.request_elapsed_time(), sample_started_at + if motion_observed and status.phase in ("braking", "standstill"): + raise HettichCentrifugeError( + f"The centrifuge began {status.phase} at {actual_speed} rpm before reaching " + f"the target speed of {speed} rpm" + ) + if loop.time() >= deadline: + raise TimeoutError(f"Centrifuge did not reach {speed} rpm within {timeout} seconds") + await asyncio.sleep(self.poll_interval) + + async def spin( + self, + g: float, + duration: int, + timeout: Optional[float] = None, + ) -> None: + """Run a finite centrifugation cycle and block until standstill. + + Args: + g: Relative centrifugal force in multiples of standard gravity (× g). + duration: Time in seconds at the requested force, excluding acceleration and braking. + timeout: Total wait timeout. By default this uses the centrifuge's maximum + configured run-up and run-down times plus a communication margin. + + Force is converted to the nearest integer RPM using the configured rotor + specification, or the device-reported maximum RPM and RCF when no catalog + number is configured. The latter depends on the radius configured on the + centrifuge. The installed rotor's live speed limit is checked before START. + + The centrifuge's native timer starts during acceleration. This method first + gives that timer a bounded safety value, then replaces its normal end time + once the measured rotor speed reaches the converted target. Timing has one-second + resolution, matching the device protocol. + + If braking or standstill is observed before the expected end, allowing for + timer rounding and query delays, the method waits for standstill and raises + ``HettichCentrifugeError`` instead of reporting a completed cycle. + """ + async with self._machine.operation("preparing_to_spin"): + self._spin_completion = asyncio.Event() + self._spin_stop_requested = False + try: + operation_data: dict[str, Any] = { + "device": device_reference(self, name=self.name), + "resources": [], + "bucket_resources": [], + "relative_centrifugal_force": g, + "duration": duration, + } + with event_operation("centrifuge.spin", **operation_data): + if not 1 <= duration <= MAXIMUM_DURATION: + raise ValueError(f"duration must be 1..{MAXIMUM_DURATION} seconds") + if timeout is not None and timeout <= duration: + raise ValueError("timeout must exceed duration to allow for acceleration and braking") + if not math.isfinite(g) or g <= 0: + raise ValueError("g must be a finite, positive relative centrifugal force") + + maximum_run_up_time = await self._enquire_parameter(MAXIMUM_RUN_UP_TIME_PARAMETER) + maximum_target_duration = MAXIMUM_DURATION - maximum_run_up_time + if duration > maximum_target_duration: + raise ValueError( + f"duration must not exceed {maximum_target_duration} seconds with the centrifuge's " + f"configured maximum run-up time of {maximum_run_up_time} seconds" + ) + cycle_timeout: float + if timeout is None: + maximum_run_down_time = await self._enquire_parameter(MAXIMUM_RUN_DOWN_TIME_PARAMETER) + cycle_timeout = ( + duration + maximum_run_up_time + maximum_run_down_time + DEFAULT_TIMEOUT_MARGIN + ) + else: + cycle_timeout = timeout + + speed = await self._speed_for_spin(g) + loop = asyncio.get_running_loop() + deadline = loop.time() + cycle_timeout + initial_run_time = duration + maximum_run_up_time + await self._start_spin(run_time=initial_run_time, speed=speed) + try: + self._machine.set_activity("accelerating") + remaining = deadline - loop.time() + if remaining <= 0: + raise TimeoutError( + f"Centrifuge did not reach {speed} rpm within the {cycle_timeout}-second timeout" + ) + elapsed_at_target, sampled_at = await self._wait_for_target_speed( + speed=speed, timeout=remaining + ) + self._raise_if_spin_stop_requested() + self._machine.set_activity("at_speed") + end_time = elapsed_at_target + duration + if end_time > MAXIMUM_DURATION: + raise ValueError( + "duration is too long to exclude acceleration within the device's maximum run time" + ) + await self._select_parameter(RUN_TIME_PARAMETER, end_time) + await self._select_parameter(ACTIVATE_PARAMETERS_COMMAND, 0x0001) + + remaining = deadline - loop.time() + if remaining <= 0: + raise TimeoutError(f"Centrifuge cycle exceeded its {cycle_timeout}-second timeout") + await self._wait_for_standstill( + timeout=remaining, motion_observed=True, expected_end_at=sampled_at + duration + ) + except BaseException: + await self._stop_after_spin_failure() + raise + finally: + self._spin_stop_requested = False + self._spin_completion.set() + + async def _speed_for_spin(self, g: float) -> int: + """Convert requested RCF to firmware RPM using catalog data or live rotor limits.""" + if self.rotor_specification is not None: + speed = self.rotor_specification.g_to_rpm(g) + else: + maximum_speed = await self.request_maximum_speed() + maximum_rcf = await self._enquire_parameter(MAXIMUM_RCF_PARAMETER) + if maximum_speed < MINIMUM_SPEED or maximum_rcf == 0: + raise HettichCentrifugeError( + "The centrifuge reported invalid rotor limits for RCF conversion" + ) + if g > maximum_rcf: + raise ValueError(f"g must not exceed the installed rotor limit of {maximum_rcf} × g") + speed = round(maximum_speed * math.sqrt(g / maximum_rcf)) + if speed < MINIMUM_SPEED: + raise ValueError(f"g corresponds to a speed below the device minimum of {MINIMUM_SPEED} rpm") + return speed + + async def stop_spin(self, timeout: float = 300.0) -> None: + """Emergency-stop an active run and wait for standstill. + + The Hettich manual explicitly classifies a PC STOP command as an emergency + stop. If PLR owns a spin, ask that workflow to stop and wait for it to release + ownership. Then verify standstill, sending STOP if necessary. This operation + is available during recovery and never clears the recovery flag. + """ + if timeout <= 0: + raise ValueError("timeout must be positive") + deadline = asyncio.get_running_loop().time() + timeout + if self.state.activity in ("preparing_to_spin", "accelerating", "at_speed", "braking"): + completion = self._spin_completion + assert completion is not None + self._spin_stop_requested = True + await asyncio.wait_for(completion.wait(), timeout=timeout) + async with self._machine.operation("stopping_spin", allow_recovery=True): + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError("Timed out waiting for the spin workflow to stop") + await self._stop_spin(timeout=remaining) + + async def _stop_spin(self, timeout: float = 300.0) -> None: + """Stop under the owning operation's guard, without competing for ownership.""" + status = await self.request_status() + if status.phase == "standstill": + return + logger.warning("[Hettich %s] emergency-stopping centrifugation", self.io.port) + await self._select_parameter(SPIN_COMMAND, 0x0001) + await self._wait_for_standstill(timeout=timeout, motion_observed=True) + + +class HettichCooledRoboticCentrifuge(HettichRoboticCentrifuge, ABC): + """Abstract Generation 2 base for refrigerated Hettich robotic centrifuges.""" + + async def request_temperature(self) -> float: + """Return the actual chamber temperature in degrees Celsius.""" + raw = await self._enquire_parameter(TEMPERATURE_PARAMETER) + return raw / 2 - 25 + + +class HettichMikro220RoboticCentrifuge(HettichRoboticCentrifuge): + """Hettich MIKRO 220 Robotic centrifuge.""" + + @classmethod + def _configuration(cls) -> _ModelConfiguration: + """Return the MIKRO 220 POS protocol identity and rotor catalog.""" + return _MIKRO_220_ROBOTIC_CONFIGURATION + + +class HettichRotanta460RoboticCentrifuge(HettichCooledRoboticCentrifuge): + """Hettich ROTANTA 460 Robotic refrigerated centrifuge.""" + + @classmethod + def _configuration(cls) -> _ModelConfiguration: + """Return the ROTANTA 460 R POS protocol identity.""" + return _ROTANTA_460_ROBOTIC_CONFIGURATION + + +class HettichRotina380RoboticCentrifuge(HettichRoboticCentrifuge): + """Hettich ROTINA 380 Robotic centrifuge.""" + + @classmethod + def _configuration(cls) -> _ModelConfiguration: + """Return the ROTINA 380 POS protocol identity.""" + return _ROTINA_380_ROBOTIC_CONFIGURATION + + +class HettichRotina380RRoboticCentrifuge(HettichCooledRoboticCentrifuge): + """Hettich ROTINA 380 R Robotic refrigerated centrifuge.""" + + @classmethod + def _configuration(cls) -> _ModelConfiguration: + """Return the ROTINA 380 R POS protocol identity.""" + return _ROTINA_380_R_ROBOTIC_CONFIGURATION diff --git a/pylabrobot/hettich/centrifuge_tests.py b/pylabrobot/hettich/centrifuge_tests.py new file mode 100644 index 00000000000..3ac4ce913ea --- /dev/null +++ b/pylabrobot/hettich/centrifuge_tests.py @@ -0,0 +1,1053 @@ +import asyncio +import unittest +from inspect import isabstract +from typing import List, Optional, TypeVar, cast +from unittest.mock import AsyncMock, patch + +from pylabrobot.events import EventBus, PLREvent, use_event_bus +from pylabrobot.hettich.centrifuge import ( + ACK, + ENQ, + EOT, + ETX, + MIKRO_220_ROBOTIC_ROTORS, + NAK, + STX, + HettichCentrifugeError, + HettichCommandError, + HettichCommunicationError, + HettichCooledRoboticCentrifuge, + HettichMikro220RoboticCentrifuge, + HettichRoboticCentrifuge, + HettichRotanta460RoboticCentrifuge, + HettichRotina380RoboticCentrifuge, + HettichRotina380RRoboticCentrifuge, +) +from pylabrobot.io.serial import Serial + + +def enquiry_reply(parameter: str, value: int, address: str = "]") -> bytes: + body = bytes([STX]) + parameter.encode("ascii") + f"={value:04X}".encode("ascii") + bytes([ETX]) + return bytes([ord(address)]) + body + bytes([HettichRoboticCentrifuge._bcc(body[1:])]) + + +HettichCentrifugeT = TypeVar("HettichCentrifugeT", bound=HettichRoboticCentrifuge) + + +def make_model_device( + replies: List[bytes], + device_class: type[HettichCentrifugeT], + connected: bool = True, + **kwargs, +) -> HettichCentrifugeT: + io = AsyncMock(spec=Serial) + io.port = "FAKE" + pending = list(replies) + rx = bytearray() + + async def write(data: bytes) -> None: + if data != bytes([EOT]) and pending: + rx.extend(pending.pop(0)) + + async def read(num_bytes: int = 1) -> bytes: + output = bytes(rx[:num_bytes]) + del rx[:num_bytes] + return output + + io.write.side_effect = write + io.read.side_effect = read + with patch("pylabrobot.hettich.centrifuge.Serial", return_value=io): + device = device_class(port="FAKE", timeout=0.2, poll_interval=0, **kwargs) + if connected: + device._machine.set_connection("connected") + return device + + +def make_device( + replies: List[bytes], rotor_catalog_number: Optional[str] = "2334", **kwargs +) -> HettichMikro220RoboticCentrifuge: + return make_model_device( + replies, HettichMikro220RoboticCentrifuge, rotor_catalog_number=rotor_catalog_number, **kwargs + ) + + +def writes(device: HettichRoboticCentrifuge) -> AsyncMock: + return cast(AsyncMock, device.io.write) + + +def telegrams(device: HettichRoboticCentrifuge) -> List[bytes]: + return [call.args[0] for call in writes(device).call_args_list if call.args[0] != bytes([EOT])] + + +def telegram_parameters(device: HettichRoboticCentrifuge) -> List[bytes]: + """Return each ENQUIRY or SELECT parameter from recorded wire frames.""" + return [frame[3:8] if frame[2] == STX else frame[2:7] for frame in telegrams(device)] + + +class HettichAsyncTestCase(unittest.IsolatedAsyncioTestCase): + """Run mocked protocol exchanges without real communication delays.""" + + def setUp(self) -> None: + """Advance virtual time on sleeps without delaying the test suite.""" + self.now = 0.0 + original_sleep = asyncio.sleep + + async def sleep(delay: float) -> None: + """Advance the clock and let concurrent callers run.""" + self.now += delay + await original_sleep(0) + + self.monotonic = patch("pylabrobot.hettich.centrifuge.monotonic", side_effect=lambda: self.now) + self.sleep = patch("pylabrobot.hettich.centrifuge.asyncio.sleep", side_effect=sleep) + self.monotonic.start() + self.sleep.start() + self.addCleanup(self.monotonic.stop) + self.addCleanup(self.sleep.stop) + + def schedule_spin_states(self, device: HettichRoboticCentrifuge, times: list[float]) -> None: + """Schedule status replies relative to the elapsed-time query at target speed.""" + original_write = writes(device).side_effect + pending = list(times) + timer_requested_at = None + + async def write(data: bytes) -> None: + """Advance virtual time when a scheduled phase is observed.""" + nonlocal timer_requested_at + if data == device._build_enquiry("00602"): + timer_requested_at = self.now + elif data == device._build_enquiry("00634") and timer_requested_at is not None and pending: + self.now = max(self.now, timer_requested_at + pending.pop(0)) + await original_write(data) + + writes(device).side_effect = write + + +class HettichEnquiryTimingTests(HettichAsyncTestCase): + """Verify protocol timing with a virtual clock and a mocked serial transport.""" + + def record_transmissions(self, device: HettichRoboticCentrifuge) -> list[tuple[bytes, float]]: + """Record transmission times and simulate 100 ms spent receiving each reply.""" + transmissions: list[tuple[bytes, float]] = [] + original_write = writes(device).side_effect + + async def write(data: bytes) -> None: + """Timestamp the telegram and its terminating EOT.""" + if data == bytes([EOT]): + self.now += 0.1 + transmissions.append((data, self.now)) + await original_write(data) + + writes(device).side_effect = write + return transmissions + + def assert_enquiries_spaced(self, transmissions: list[tuple[bytes, float]]) -> None: + """Require 400 ms after the previous enquiry's completion before the next.""" + last_end = None + enquiry = False + for frame, timestamp in transmissions: + if frame == bytes([EOT]): + if enquiry: + last_end = timestamp + enquiry = False + else: + enquiry = frame[-1] == ENQ + if enquiry and last_end is not None: + self.assertGreaterEqual(timestamp - last_end, 0.4 - 1e-9) + + async def test_status_speed_and_elapsed_queries_are_spaced(self) -> None: + """Every query is spaced even when callers disable motion polling delays.""" + device = make_device( + [ + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 2000), + enquiry_reply("00602", 12), + ] + ) + transmissions = self.record_transmissions(device) + + await device.request_status() + self.assertEqual(await device.request_speed(), 2000) + self.assertEqual(await device.request_elapsed_time(), 12) + + self.assert_enquiries_spaced(transmissions) + + async def test_retries_and_fault_queries_are_spaced(self) -> None: + """Corrupt replies and NAK handling must respect the same enquiry interval.""" + corrupt = bytearray(enquiry_reply("00604", 500)) + corrupt[-1] ^= 1 + device = make_device([bytes(corrupt), bytes([ord("]"), NAK]), enquiry_reply("00685", 0x0080)]) + transmissions = self.record_transmissions(device) + + with self.assertRaises(HettichCommandError): + await device.request_speed() + + self.assertEqual(telegram_parameters(device), [b"00604", b"00604", b"00685"]) + self.assert_enquiries_spaced(transmissions) + + async def test_concurrent_queries_are_spaced_under_the_transaction_lock(self) -> None: + """Independent callers share one enquiry schedule.""" + device = make_device([enquiry_reply("00604", 500), enquiry_reply("00602", 12)]) + transmissions = self.record_transmissions(device) + + self.assertEqual( + await asyncio.gather(device.request_speed(), device.request_elapsed_time()), [500, 12] + ) + + self.assert_enquiries_spaced(transmissions) + + async def test_select_is_not_delayed_and_idle_time_counts_toward_spacing(self) -> None: + """Emergency STOP writes need no extra delay and an idle bus needs no extra sleep.""" + device = make_device( + [enquiry_reply("00604", 500), bytes([ord("]"), ACK]), enquiry_reply("00604", 0)] + ) + transmissions = self.record_transmissions(device) + + await device.request_speed() + ready_at = self.now + await device._select_parameter("00521", 1) + stop_time = next(t for frame, t in transmissions if frame == device._build_select("00521", 1)) + self.assertEqual(stop_time, ready_at) + self.now += 1 + ready_at = self.now + await device.request_speed() + self.assertEqual(transmissions[-2][1], ready_at) + + +class HettichFrameTests(unittest.TestCase): + def setUp(self) -> None: + self.device = make_device([], rotor_catalog_number=None) + + def test_build_enquiry_matches_manual_example(self) -> None: + self.assertEqual( + self.device._build_enquiry("00604"), + bytes([0x04, 0x5D, 0x30, 0x30, 0x36, 0x30, 0x34, 0x05]), + ) + + def test_transaction_lock_is_created_lazily(self) -> None: + self.assertIsNone(self.device._transaction_lock) + + def test_build_select_matches_manual_example(self) -> None: + self.assertEqual( + self.device._build_select("00603", 1500), + bytes( + [ + 0x04, + 0x5D, + 0x02, + 0x30, + 0x30, + 0x36, + 0x30, + 0x33, + 0x3D, + 0x30, + 0x35, + 0x44, + 0x43, + 0x03, + 0x09, + ] + ), + ) + + def test_parse_enquiry_matches_manual_example(self) -> None: + reply = bytes( + [0x5D, 0x02, 0x30, 0x30, 0x36, 0x30, 0x34, 0x3D, 0x30, 0x31, 0x46, 0x34, 0x03, 0x7F] + ) + self.assertEqual(self.device._parse_enquiry_reply(reply, "00604"), 500) + + def test_rejects_invalid_address_and_short_timeout(self) -> None: + with self.assertRaises(ValueError): + HettichMikro220RoboticCentrifuge(port="FAKE", address="a") + with self.assertRaises(ValueError): + HettichMikro220RoboticCentrifuge(port="FAKE", timeout=0.1) + + def test_protocol_and_cooled_bases_are_abstract(self) -> None: + self.assertTrue(isabstract(HettichRoboticCentrifuge)) + self.assertTrue(isabstract(HettichCooledRoboticCentrifuge)) + self.assertFalse(isabstract(HettichMikro220RoboticCentrifuge)) + self.assertFalse(isabstract(HettichRotanta460RoboticCentrifuge)) + self.assertFalse(isabstract(HettichRotina380RoboticCentrifuge)) + self.assertFalse(isabstract(HettichRotina380RRoboticCentrifuge)) + + def test_mikro_220_robotic_rotor_table(self) -> None: + self.assertEqual(set(MIKRO_220_ROBOTIC_ROTORS), {"2334", "2394"}) + self.assertEqual(MIKRO_220_ROBOTIC_ROTORS["2334"].maximum_speed, 13_000) + self.assertEqual(MIKRO_220_ROBOTIC_ROTORS["2334"].maximum_rcf, 18_327) + self.assertEqual(MIKRO_220_ROBOTIC_ROTORS["2334"].maximum_volume, 2_000) + self.assertEqual(MIKRO_220_ROBOTIC_ROTORS["2394"].maximum_speed, 13_000) + self.assertEqual(MIKRO_220_ROBOTIC_ROTORS["2394"].maximum_rcf, 18_516) + + def test_rotor_specification_converts_between_speed_and_rcf(self) -> None: + rotor = MIKRO_220_ROBOTIC_ROTORS["2394"] + self.assertEqual(rotor.rpm_to_g(13_000), 18_516) + self.assertEqual(rotor.rpm_to_g(6_500), 4_629) + self.assertEqual(rotor.g_to_rpm(18_516), 13_000) + self.assertEqual(rotor.g_to_rpm(4_629), 6_500) + + def test_rotor_specification_rejects_values_above_limits(self) -> None: + rotor = MIKRO_220_ROBOTIC_ROTORS["2394"] + with self.assertRaisesRegex(ValueError, "13000 rpm"): + rotor.rpm_to_g(13_001) + with self.assertRaisesRegex(ValueError, "18516"): + rotor.g_to_rpm(18_517) + + def test_device_uses_configured_rotor_specification(self) -> None: + device = make_device([], rotor_catalog_number="2334") + self.assertIs(device.rotor_specification, MIKRO_220_ROBOTIC_ROTORS["2334"]) + self.assertEqual(device.rpm_to_g(13_000), 18_327) + self.assertEqual(device.g_to_rpm(18_327), 13_000) + + def test_device_requires_known_rotor_for_rcf_conversion(self) -> None: + with self.assertRaisesRegex(ValueError, "unsupported rotor catalog"): + make_device([], rotor_catalog_number="unknown") + with self.assertRaisesRegex(HettichCentrifugeError, "rotor_catalog_number"): + self.device.rpm_to_g(1_000) + + +class HettichProtocolTests(HettichAsyncTestCase): + async def test_setup_is_read_only_and_records_identity(self) -> None: + device = make_model_device( + [ + enquiry_reply("00685", 0x0001), + enquiry_reply("00600", 0x1234), + enquiry_reply("00537", 0xC901), + enquiry_reply("00636", 0x0112), + ], + HettichRotanta460RoboticCentrifuge, + connected=False, + ) + with patch("pylabrobot.hettich.centrifuge.logger.warning") as warning: + await device.setup() + self.assertEqual(device.device_type, "ROTANTA 460 R POS") + self.assertEqual(device.software_version, "01.12") + warning.assert_called_once() + self.assertEqual( + telegram_parameters(device), + [b"00685", b"00600", b"00537", b"00636"], + ) + self.assertTrue(all(frame[-1] == ENQ for frame in telegrams(device))) + + async def test_setup_recognizes_mikro_220_hardware_code(self) -> None: + device = make_device( + [ + enquiry_reply("00685", 0x0000), + enquiry_reply("00600", 0x1234), + enquiry_reply("00537", 0xE800), + enquiry_reply("00636", 0x0121), + ], + connected=False, + ) + with patch("pylabrobot.hettich.centrifuge.logger.warning") as warning: + await device.setup() + self.assertEqual(device.device_type, "MIKRO 220 POS") + self.assertEqual(device.software_version, "01.21") + warning.assert_not_called() + + async def test_setup_rejects_unknown_e8_code_without_family_fallback(self) -> None: + device = make_device( + [ + enquiry_reply("00685", 0x0000), + enquiry_reply("00600", 0x1234), + enquiry_reply("00537", 0xE8FF), + ], + connected=False, + ) + + with self.assertRaisesRegex(HettichCentrifugeError, "unknown type 0xE8FF"): + await device.setup() + + cast(AsyncMock, device.io.stop).assert_awaited_once() + + async def test_setup_rejects_a_different_known_model(self) -> None: + device = make_device( + [ + enquiry_reply("00685", 0x0000), + enquiry_reply("00600", 0x1234), + enquiry_reply("00537", 0xC901), + ], + connected=False, + ) + + with self.assertRaisesRegex(HettichCentrifugeError, "ROTANTA 460 R POS"): + await device.setup() + + cast(AsyncMock, device.io.stop).assert_awaited_once() + + async def test_temperature_is_available_on_refrigerated_model(self) -> None: + device = make_model_device( + [enquiry_reply("00619", 70)], + HettichRotanta460RoboticCentrifuge, + ) + self.assertEqual(await device.request_temperature(), 10.0) + + async def test_enquiry_retries_after_bad_checksum(self) -> None: + corrupt = bytearray(enquiry_reply("00604", 500)) + corrupt[-1] ^= 0x01 + device = make_device([bytes(corrupt), enquiry_reply("00604", 500)]) + self.assertEqual(await device.request_speed(), 500) + self.assertEqual(len(telegrams(device)), 2) + + async def test_enquiry_fails_after_three_timeouts(self) -> None: + device = make_device([b"", b"", b""]) + with self.assertRaises(HettichCommunicationError): + await device.request_speed() + self.assertEqual(len(telegrams(device)), 3) + + async def test_nak_reads_and_decodes_siof(self) -> None: + device = make_device( + [ + bytes([ord("]"), NAK]), + enquiry_reply("00685", 0x0080), + ] + ) + with self.assertRaisesRegex(HettichCommandError, "improper value or command not allowed"): + await device._select_parameter("00603", 0xFFFF) + self.assertEqual(telegram_parameters(device), [b"00603", b"00685"]) + + async def test_request_status_decodes_both_state_words(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + ] + ) + status = await device.request_status() + self.assertEqual(status.phase, "accelerating") + self.assertTrue(status.status_changed) + self.assertTrue(status.can_start) + self.assertEqual(status.program_number, 1) + self.assertEqual(status.rotor_number, 9) + self.assertEqual(status.key_lock, "remote") + self.assertTrue(status.lid_closed) + + async def test_open_hatch_is_noop_when_already_open(self) -> None: + device = make_device([enquiry_reply("00528", 0xA000)]) + await device.open_hatch() + self.assertEqual(telegram_parameters(device), [b"00528"]) + + async def test_open_hatch_moves_and_waits_for_open_sensor(self) -> None: + device = make_device( + [ + enquiry_reply("00528", 0x1800), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + enquiry_reply("00528", 0xA000), + ] + ) + + await device.open_hatch() + + frames = telegrams(device) + self.assertEqual(frames[3], device._build_select("00526", 0x0060)) + + async def test_close_hatch_moves_and_waits_for_both_closed_sensors(self) -> None: + device = make_device( + [ + enquiry_reply("00528", 0xA000), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + enquiry_reply("00528", 0x1800), + ] + ) + + await device.close_hatch() + + frames = telegrams(device) + self.assertEqual(frames[3], device._build_select("00526", 0x0070)) + + async def test_move_to_position_requires_closed_main_lid(self) -> None: + device = make_device( + [ + enquiry_reply("00524", 0x1801), + enquiry_reply("00528", 0x2004), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA092), + ] + ) + + with self.assertRaisesRegex(HettichCentrifugeError, "main centrifuge lid"): + await device.move_to_position(2) + + self.assertTrue(all(frame[-1] == ENQ for frame in telegrams(device))) + + async def test_move_to_position_allows_closed_loading_hatch(self) -> None: + device = make_device( + [ + enquiry_reply("00524", 0x1801), + enquiry_reply("00528", 0x1804), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + enquiry_reply("00528", 0x1803), + enquiry_reply("00528", 0x1806), + ] + ) + + await device.move_to_position(2, speed="slow") + + frames = telegrams(device) + self.assertEqual(frames[4], device._build_select("00524", 0x1802)) + self.assertEqual(frames[5], device._build_select("00526", 0x0001)) + self.assertEqual(telegram_parameters(device)[-2:], [b"00528", b"00528"]) + + async def test_end_positioning_is_idempotent_and_can_end_active_mode(self) -> None: + inactive = make_device([enquiry_reply("00528", 0x1800)]) + await inactive.end_positioning() + self.assertEqual(telegram_parameters(inactive), [b"00528"]) + + active = make_device( + [ + enquiry_reply("00528", 0x1802), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + enquiry_reply("00528", 0x1800), + ] + ) + await active.end_positioning() + self.assertEqual(telegrams(active)[3], active._build_select("00526", 0x0080)) + + async def test_select_program_is_idempotent_and_selects_a_different_program(self) -> None: + current = make_device( + [ + enquiry_reply("00634", 0x0262), + enquiry_reply("00635", 0xA292), + ] + ) + await current.select_program(2) + self.assertEqual(telegram_parameters(current), [b"00634", b"00635"]) + + different = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + ] + ) + await different.select_program(2) + self.assertEqual(telegrams(different)[-1], different._build_select("00523", 0x0204)) + + async def test_live_value_requests_use_their_protocol_parameters(self) -> None: + device = make_device( + [ + enquiry_reply("00604", 500), + enquiry_reply("00605", 13_000), + enquiry_reply("00602", 17), + ] + ) + + self.assertEqual(await device.request_speed(), 500) + self.assertEqual(await device.request_maximum_speed(), 13_000) + self.assertEqual(await device.request_elapsed_time(), 17) + self.assertEqual(telegram_parameters(device), [b"00604", b"00605", b"00602"]) + + async def test_private_start_spin_checks_state_and_sets_parameters(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + ] + ) + await device._start_spin(run_time=30, speed=2000) + frames = telegrams(device) + self.assertEqual( + telegram_parameters(device), + [b"00634", b"00635", b"00528", b"00605", b"00601", b"00603", b"00522", b"00521"], + ) + self.assertEqual(frames[-4], device._build_select("00601", 30)) + self.assertEqual(frames[-3], device._build_select("00603", 2000)) + self.assertEqual(frames[-2], device._build_select("00522", 1)) + self.assertEqual(frames[-1], device._build_select("00521", 2)) + + async def test_private_start_spin_rejects_speed_above_rotor_limit(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ] + ) + with self.assertRaisesRegex(ValueError, "5000 rpm"): + await device._start_spin(run_time=30, speed=5001) + self.assertNotIn(b"00521", telegram_parameters(device)) + + async def test_private_start_spin_rejects_non_remote_key_position(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA293), + ] + ) + with self.assertRaisesRegex(HettichCentrifugeError, "LOCK 2"): + await device._start_spin(run_time=30, speed=500) + self.assertNotIn(b"00521", telegram_parameters(device)) + + async def test_private_start_spin_ends_positioning_before_start(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1802), + bytes([ord("]"), ACK]), + enquiry_reply("00528", 0x1800), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00605", 5000), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + ] + ) + await device._start_spin(run_time=30, speed=2000) + parameters = telegram_parameters(device) + self.assertLess(parameters.index(b"00526"), parameters.index(b"00521")) + self.assertEqual(parameters.count(b"00634"), 2) + + async def test_private_wait_for_standstill_observes_motion_before_returning(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + status = await device._wait_for_standstill(timeout=1, motion_observed=False) + self.assertEqual(status.phase, "standstill") + self.assertEqual(telegram_parameters(device).count(b"00634"), 3) + + async def test_spin_counts_duration_from_target_speed(self) -> None: + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 2100), + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 2000), + enquiry_reply("00602", 12), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + enquiry_reply("00634", 0x01F0), + enquiry_reply("00635", 0xA292), + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + + self.schedule_spin_states(device, [30, 40]) + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + + frames = telegrams(device) + run_time_frames = [ + frame for frame in frames if (frame[3:8] if frame[2] == STX else frame[2:7]) == b"00601" + ] + self.assertEqual( + run_time_frames, + [device._build_select("00601", 60), device._build_select("00601", 42)], + ) + self.assertEqual(telegram_parameters(device).count(b"00522"), 2) + + async def test_spin_rejects_impossible_target_duration_before_motion(self) -> None: + device = make_device([enquiry_reply("00614", 30)]) + + with self.assertRaisesRegex(ValueError, "59969 seconds"): + await device.spin(g=device.rpm_to_g(2_000), duration=59_970) + + self.assertEqual(telegram_parameters(device), [b"00614"]) + self.assertNotIn(b"00521", telegram_parameters(device)) + + async def test_spin_stops_after_lost_start_acknowledgements(self) -> None: + """A lost START reply must not leave a potentially running rotor unattended.""" + ack = bytes([ord("]"), ACK]) + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ack, + ack, + ack, + b"", + b"", + b"", + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + ack, + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + + with self.assertRaisesRegex(HettichCommunicationError, "SELECT 00521 failed"): + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + + self.assertEqual(telegrams(device).count(device._build_select("00521", 2)), 3) + self.assertIn(device._build_select("00521", 1), telegrams(device)) + self.assertEqual(telegram_parameters(device)[-2:], [b"00634", b"00635"]) + + async def test_spin_stops_when_cancelled_while_awaiting_start_reply(self) -> None: + """Cancellation during START must stop the rotor and still propagate cancellation.""" + ack = bytes([ord("]"), ACK]) + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ack, + ack, + ack, + ack, + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + ack, + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + start_sent = asyncio.Event() + original_read = cast(AsyncMock, device.io.read).side_effect + + async def read(num_bytes: int = 1) -> bytes: + """Suspend delivery of the START reply until the spin task is cancelled.""" + reply = cast(bytes, await original_read(num_bytes)) + if telegrams(device)[-1] == device._build_select("00521", 2): + start_sent.set() + await asyncio.Event().wait() + return reply + + cast(AsyncMock, device.io.read).side_effect = read + task = asyncio.create_task(device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60)) + await asyncio.wait_for(start_sent.wait(), timeout=1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + + self.assertIn(device._build_select("00521", 1), telegrams(device)) + self.assertEqual(telegram_parameters(device)[-2:], [b"00634", b"00635"]) + + async def test_spin_preflight_failure_does_not_stop_an_existing_run(self) -> None: + """Rejecting an already running machine must not interrupt its existing cycle.""" + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + ] + ) + with self.assertRaisesRegex(HettichCentrifugeError, "standstill"): + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + + self.assertEqual(telegram_parameters(device), [b"00614", b"00634", b"00635"]) + + async def test_stop_spin_is_noop_at_standstill(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + ] + ) + await device.stop_spin() + self.assertNotIn(b"00521", telegram_parameters(device)) + + async def test_stop_spin_sends_emergency_stop_and_waits_for_standstill(self) -> None: + device = make_device( + [ + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + bytes([ord("]"), ACK]), + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + + await device.stop_spin(timeout=1) + + self.assertEqual(telegrams(device)[2], device._build_select("00521", 0x0001)) + + +class HettichSpinCompletionTests(HettichAsyncTestCase): + """Distinguish completed timed runs from early stops without operating hardware.""" + + async def check_cycle(self, phases: list[tuple[int, float]], interrupted: bool) -> None: + """Run a complete mocked spin and check its result and lifecycle events.""" + ack = bytes([ord("]"), ACK]) + replies = [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ack, + ack, + ack, + ack, + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 2000), + enquiry_reply("00602", 12), + ack, + ack, + ] + for phase, _ in phases: + replies.extend([enquiry_reply("00634", phase), enquiry_reply("00635", 0xA292)]) + # Failure cleanup verifies that the rotor is already stopped. + replies.extend([enquiry_reply("00634", 0x01E2), enquiry_reply("00635", 0xA292)]) + device = make_device(replies) + self.schedule_spin_states(device, [when for _, when in phases]) + events: list[PLREvent] = [] + bus = EventBus() + bus.subscribe(events.append) + + with use_event_bus(bus): + if interrupted: + with self.assertRaisesRegex(HettichCentrifugeError, "interrupted"): + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + else: + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + + terminal = "failed" if interrupted else "completed" + self.assertEqual( + [event.name for event in events], ["centrifuge.spin.started", f"centrifuge.spin.{terminal}"] + ) + self.assertNotIn(device._build_select("00521", 1), telegrams(device)) + self.assertEqual( + telegram_parameters(device).count(b"00634"), 2 + len(phases) + int(interrupted) + ) + self.assertGreaterEqual(self.now, phases[-1][1]) + + async def test_early_braking_stays_interrupted_after_a_long_run_down(self) -> None: + """Braking time cannot satisfy the requested time at speed.""" + await self.check_cycle([(0x01F0, 5), (0x01F0, 35), (0x01E2, 40)], interrupted=True) + + async def test_early_standstill_is_interrupted_when_braking_was_not_polled(self) -> None: + """A short run-down between polls must also report interruption.""" + await self.check_cycle([(0x01E2, 5)], interrupted=True) + + async def test_complete_cycle_succeeds_with_delayed_polling(self) -> None: + """Observing braking after the programmed end remains a successful cycle.""" + await self.check_cycle([(0x01E8, 20), (0x01F0, 32), (0x01E2, 40)], interrupted=False) + + async def test_timer_rounding_near_the_requested_end_is_allowed(self) -> None: + """A subsecond difference due to the integer device timer is not an interruption.""" + await self.check_cycle([(0x01F0, 29.1), (0x01E2, 40)], interrupted=False) + + +class HettichForceTests(HettichAsyncTestCase): + """Verify force-based spin requests against the RPM sent to the controller.""" + + async def test_spin_converts_force_and_preserves_requested_event_value(self) -> None: + """Catalog and live-limit conversions both keep the requested RCF in events.""" + for catalog, g, rpm in (("2394", 4629.25, 6500), (None, 500.0, 2500)): + with self.subTest(catalog=catalog): + replies = [enquiry_reply("00614", 30)] + if catalog is None: + replies.extend([enquiry_reply("00605", 5000), enquiry_reply("00608", 2000)]) + replies.extend( + [ + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 13000 if catalog is not None else 5000), + b"]\x06", + b"]\x06", + b"]\x06", + b"]\x06", + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", rpm), + enquiry_reply("00602", 12), + b"]\x06", + b"]\x06", + enquiry_reply("00634", 0x01F0), + enquiry_reply("00635", 0xA292), + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ] + ) + device = make_device(replies, rotor_catalog_number=catalog) + self.schedule_spin_states(device, [30, 40]) + events: list[PLREvent] = [] + bus = EventBus() + bus.subscribe(events.append) + with use_event_bus(bus): + await device.spin(g=g, duration=30, timeout=60) + + self.assertIn(device._build_select("00603", rpm), telegrams(device)) + self.assertEqual( + [event.name for event in events], ["centrifuge.spin.started", "centrifuge.spin.completed"] + ) + self.assertEqual(events[0].data["relative_centrifugal_force"], g) + self.assertNotIn("speed_rpm", events[0].data) + self.assertFalse(device.state.recovery_required) + + async def test_spin_rejects_invalid_force_without_communicating(self) -> None: + """Nonpositive and nonfinite RCF cannot cause device changes.""" + for g in (0, -1, float("nan"), float("inf"), -float("inf")): + with self.subTest(g=g): + device = make_device([]) + with self.assertRaisesRegex(ValueError, "finite, positive"): + await device.spin(g=g, duration=30) + self.assertEqual(telegrams(device), []) + self.assertFalse(device.state.recovery_required) + + async def test_spin_checks_catalog_force_limits_before_actuation(self) -> None: + """Excessive force and force below the minimum RPM are rejected without SELECT.""" + for g in (18516.1, 0.0001): + with self.subTest(g=g): + device = make_device([enquiry_reply("00614", 30)], rotor_catalog_number="2394") + with self.assertRaises(ValueError): + await device.spin(g=g, duration=30, timeout=60) + self.assertEqual(telegram_parameters(device), [b"00614"]) + self.assertFalse(device.state.recovery_required) + + async def test_spin_checks_live_force_limits_without_a_catalog(self) -> None: + """Models without a catalog must still enforce the controller's RCF limit.""" + device = make_model_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00605", 5000), + enquiry_reply("00608", 2000), + ], + HettichRotanta460RoboticCentrifuge, + ) + with self.assertRaisesRegex(ValueError, "2000 × g"): + await device.spin(g=2001, duration=30, timeout=60) + self.assertTrue(all(frame[-1] == ENQ for frame in telegrams(device))) + + async def test_spin_rejects_invalid_live_conversion_limits(self) -> None: + """A zero RCF limit cannot be used to derive an RPM.""" + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00605", 5000), + enquiry_reply("00608", 0), + ], + rotor_catalog_number=None, + ) + with self.assertRaisesRegex(HettichCentrifugeError, "invalid rotor limits"): + await device.spin(g=500, duration=30, timeout=60) + self.assertTrue(all(frame[-1] == ENQ for frame in telegrams(device))) + + async def test_catalog_conversion_still_checks_the_live_speed_limit(self) -> None: + """A configured catalog does not override the installed rotor's speed limit.""" + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ], + rotor_catalog_number="2394", + ) + with self.assertRaisesRegex(ValueError, "5000 rpm"): + await device.spin(g=4629, duration=30, timeout=60) + self.assertTrue(all(frame[-1] == ENQ for frame in telegrams(device))) + + +class HettichEventTests(HettichAsyncTestCase): + async def test_spin_uses_vspin_event_name_and_field_conventions(self) -> None: + device = make_device( + [ + enquiry_reply("00614", 30), + enquiry_reply("00634", 0x0162), + enquiry_reply("00635", 0xA292), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 2000), + enquiry_reply("00602", 12), + bytes([ord("]"), ACK]), + bytes([ord("]"), ACK]), + enquiry_reply("00634", 0x01F0), + enquiry_reply("00635", 0xA292), + enquiry_reply("00634", 0x01E2), + enquiry_reply("00635", 0xA292), + ], + name="hettich_centrifuge", + rotor_catalog_number="2334", + ) + events: list[PLREvent] = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + self.schedule_spin_states(device, [30, 40]) + with use_event_bus(event_bus): + await device.spin(g=device.rpm_to_g(2000), duration=30, timeout=60) + + self.assertEqual( + [event.name for event in events], + ["centrifuge.spin.started", "centrifuge.spin.completed"], + ) + started, completed = events + self.assertEqual(started.context["operation_id"], completed.context["operation_id"]) + self.assertEqual(started.data["device"]["name"], "hettich_centrifuge") + self.assertEqual(started.data["resources"], []) + self.assertEqual(started.data["bucket_resources"], []) + self.assertNotIn("speed_rpm", started.data) + self.assertEqual(started.data["duration"], 30) + self.assertAlmostEqual( + cast(float, started.data["relative_centrifugal_force"]), + device.rpm_to_g(2000), + ) + self.assertNotIn("speed", started.data) + self.assertNotIn("duration_seconds", started.data) + + async def test_spin_failure_emits_requested_parameters(self) -> None: + device = make_device([], name="hettich_centrifuge") + events: list[PLREvent] = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + with self.assertRaisesRegex(ValueError, "duration"): + await device.spin(g=device.rpm_to_g(2000), duration=0) + + self.assertEqual( + [event.name for event in events], + ["centrifuge.spin.started", "centrifuge.spin.failed"], + ) + started, failed = events + self.assertEqual(started.context["operation_id"], failed.context["operation_id"]) + self.assertNotIn("speed_rpm", started.data) + self.assertEqual(started.data["duration"], 0) + self.assertEqual(started.data["relative_centrifugal_force"], device.rpm_to_g(2000)) + self.assertEqual(failed.data["error_type"], "ValueError") + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/hettich/state_tests.py b/pylabrobot/hettich/state_tests.py new file mode 100644 index 00000000000..f39bd725678 --- /dev/null +++ b/pylabrobot/hettich/state_tests.py @@ -0,0 +1,365 @@ +"""Mocked integration tests for Hettich workflow ownership and recovery.""" + +import asyncio +from unittest.mock import AsyncMock, patch + +from pylabrobot.hettich._errors import HettichCentrifugeError, HettichCommunicationError +from pylabrobot.hettich._state import HettichMachineState +from pylabrobot.hettich.centrifuge_tests import ( + HettichAsyncTestCase, + enquiry_reply, + make_device, + telegram_parameters, + telegrams, + writes, +) + +ACK_REPLY = b"]\x06" + + +def identity_replies() -> list[bytes]: + """Return a valid MIKRO startup identity exchange.""" + return [ + enquiry_reply("00685", 0), + enquiry_reply("00600", 0x1234), + enquiry_reply("00537", 0xE800), + enquiry_reply("00636", 0x0121), + ] + + +def standstill_replies() -> list[bytes]: + """Return an error-free, remotely controlled standstill status.""" + return [enquiry_reply("00634", 0x0162), enquiry_reply("00635", 0xA292)] + + +def start_replies() -> list[bytes]: + """Return the complete preflight and START exchange for a finite spin.""" + return [ + enquiry_reply("00614", 30), + *standstill_replies(), + enquiry_reply("00528", 0x1800), + enquiry_reply("00605", 5000), + ACK_REPLY, + ACK_REPLY, + ACK_REPLY, + ACK_REPLY, + ] + + +def recovery_replies(speed: int = 0, hatch: int = 0x1800) -> list[bytes]: + """Return recovery observations with configurable physical speed and hatch state.""" + return [ + *identity_replies()[:3], + *standstill_replies(), + enquiry_reply("00604", speed), + enquiry_reply("00528", hatch), + ] + + +class HettichStateTests(HettichAsyncTestCase): + """Exercise state transitions through public device workflows.""" + + async def test_connection_lifecycle_and_snapshots(self) -> None: + """Setup and stop are idempotent and leave earlier state snapshots unchanged.""" + device = make_device(identity_replies(), connected=False) + initial = device.state + self.assertEqual(initial, HettichMachineState()) + with self.assertRaisesRegex(HettichCentrifugeError, "not connected"): + await device.open_hatch() + with self.assertRaisesRegex(HettichCentrifugeError, "not connected"): + await device.request_speed() + self.assertEqual(telegrams(device), []) + + async def connect() -> None: + """Observe the in-progress lifecycle at the transport boundary.""" + self.assertEqual(device.state.connection, "connecting") + self.assertEqual(device.state.activity, "setting_up") + + with patch.object(device.io, "setup", side_effect=connect): + await device.setup() + connected = device.state + self.assertEqual(connected, HettichMachineState(connection="connected")) + await device.setup() + self.assertEqual(len(telegrams(device)), 4) + with patch.object(device.io, "stop", new_callable=AsyncMock) as close: + await device.stop() + await device.stop() + close.assert_awaited_once() + self.assertEqual(device.state.connection, "disconnected") + self.assertEqual(initial.connection, "disconnected") + self.assertEqual(connected.connection, "connected") + + async def test_preflight_failure_does_not_require_recovery(self) -> None: + """A rejected request before actuation leaves the device ready for another operation.""" + device = make_device([enquiry_reply("00528", 0xA000)]) + with self.assertRaises(ValueError): + await device.spin(g=device.rpm_to_g(500), duration=0) + self.assertEqual(device.state, HettichMachineState(connection="connected")) + await device.open_hatch() + self.assertEqual(telegram_parameters(device), [b"00528"]) + + async def test_failed_hatch_actuation_blocks_motion_but_allows_status(self) -> None: + """A command with lost replies leaves a persistent recovery requirement.""" + device = make_device( + [ + enquiry_reply("00528", 0x1800), + *standstill_replies(), + b"", + b"", + b"", + enquiry_reply("00604", 0), + ] + ) + with self.assertRaises(HettichCommunicationError): + await device.open_hatch() + self.assertTrue(device.state.recovery_required) + self.assertEqual(device.state.activity, "idle") + self.assertEqual(await device.request_speed(), 0) + before = len(telegrams(device)) + for operation in ( + device.open_hatch(), + device.close_hatch(), + device.move_to_position(1), + device.end_positioning(), + device.select_program(1), + device.spin(g=device.rpm_to_g(500), duration=30), + ): + with self.assertRaisesRegex(HettichCentrifugeError, "requires recovery"): + await operation + self.assertEqual(len(telegrams(device)), before) + + async def test_cancellation_after_hatch_command_requires_recovery(self) -> None: + """Cancellation after possible hatch movement must not make the device ready.""" + device = make_device([enquiry_reply("00528", 0x1800), *standstill_replies(), ACK_REPLY]) + waiting = asyncio.Event() + + async def wait_for_hatch(desired: str, timeout: float) -> None: + """Hold the operation after its command has been acknowledged.""" + waiting.set() + await asyncio.Event().wait() + + with patch.object(device, "_wait_for_hatch", side_effect=wait_for_hatch): + task = asyncio.create_task(device.open_hatch()) + await asyncio.wait_for(waiting.wait(), 1) + self.assertEqual(device.state.activity, "opening_hatch") + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertTrue(device.state.recovery_required) + self.assertEqual(device.state.activity, "idle") + + async def test_spin_owns_operations_while_status_remains_available(self) -> None: + """A second spin or lifecycle/motion operation cannot interleave with an active spin.""" + device = make_device([*start_replies(), enquiry_reply("00604", 500), *standstill_replies()]) + waiting = asyncio.Event() + + async def wait_for_speed(speed: int, timeout: float) -> tuple[int, float]: + """Suspend a spin after its START exchange.""" + waiting.set() + await asyncio.Event().wait() + return 0, 0.0 + + with patch.object(device, "_wait_for_target_speed", side_effect=wait_for_speed): + task = asyncio.create_task(device.spin(g=device.rpm_to_g(500), duration=30, timeout=60)) + await asyncio.wait_for(waiting.wait(), 1) + self.assertEqual(device.state.activity, "accelerating") + before = len(telegrams(device)) + for operation in ( + device.spin(g=device.rpm_to_g(500), duration=30), + device.open_hatch(), + device.close_hatch(), + device.move_to_position(1), + device.end_positioning(), + device.select_program(1), + device.setup(), + device.stop(), + device.recover(), + ): + with self.assertRaisesRegex(HettichCentrifugeError, "is active"): + await operation + self.assertEqual(len(telegrams(device)), before) + self.assertEqual(await device.request_speed(), 500) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertTrue(device.state.recovery_required) + + async def test_stop_request_uses_spin_owner_and_waits_for_standstill(self) -> None: + """stop_spin must interrupt the owning workflow without deadlock or duplicate STOP.""" + device = make_device( + [ + *start_replies(), + enquiry_reply("00634", 0x01E4), + enquiry_reply("00635", 0xA292), + ACK_REPLY, + *standstill_replies(), + *standstill_replies(), + ] + ) + waiting, release = asyncio.Event(), asyncio.Event() + original_wait = device._wait_for_target_speed + + async def wait_for_speed(speed: int, timeout: float) -> tuple[int, float]: + """Allow a stop request to arrive before the owner's next polling iteration.""" + waiting.set() + await release.wait() + return await original_wait(speed, timeout) + + with patch.object(device, "_wait_for_target_speed", side_effect=wait_for_speed): + spin = asyncio.create_task(device.spin(g=device.rpm_to_g(500), duration=30, timeout=60)) + await asyncio.wait_for(waiting.wait(), 1) + stop = asyncio.create_task(device.stop_spin()) + await asyncio.sleep(0) + self.assertFalse(stop.done()) + release.set() + with self.assertRaisesRegex(HettichCentrifugeError, "interrupted by stop_spin"): + await asyncio.wait_for(spin, 1) + await asyncio.wait_for(stop, 1) + self.assertEqual(telegrams(device).count(device._build_select("00521", 1)), 1) + self.assertEqual(device.state.activity, "idle") + self.assertTrue(device.state.recovery_required) + + async def test_stop_during_preflight_prevents_start_without_recovery(self) -> None: + """A stop request before any SELECT prevents actuation and preserves readiness.""" + device = make_device([enquiry_reply("00614", 30), *standstill_replies()]) + waiting, release = asyncio.Event(), asyncio.Event() + original_write = writes(device).side_effect + + async def write(data: bytes) -> None: + """Pause a preparation query while the caller requests a stop.""" + await original_write(data) + if data == device._build_enquiry("00614"): + waiting.set() + await release.wait() + + writes(device).side_effect = write + spin = asyncio.create_task(device.spin(g=device.rpm_to_g(500), duration=30, timeout=60)) + await asyncio.wait_for(waiting.wait(), 1) + self.assertEqual(device.state.activity, "preparing_to_spin") + stop = asyncio.create_task(device.stop_spin()) + await asyncio.sleep(0) + release.set() + with self.assertRaisesRegex(HettichCentrifugeError, "interrupted by stop_spin"): + await asyncio.wait_for(spin, 1) + await asyncio.wait_for(stop, 1) + self.assertEqual(device.state, HettichMachineState(connection="connected")) + self.assertEqual(telegram_parameters(device), [b"00614", b"00634", b"00635"]) + + async def test_failed_spin_cleanup_preserves_failure_and_requires_recovery(self) -> None: + """Failed emergency stopping must not mask the original error or enable more motion.""" + device = make_device([*start_replies(), b"", b"", b"", enquiry_reply("00604", 0)]) + failure = TimeoutError("target speed timed out") + with patch.object(device, "_wait_for_target_speed", side_effect=failure): + with self.assertLogs("pylabrobot.hettich.centrifuge", level="ERROR"): + with self.assertRaises(TimeoutError) as raised: + await device.spin(g=device.rpm_to_g(500), duration=30, timeout=60) + self.assertIs(raised.exception, failure) + self.assertTrue(device.state.recovery_required) + self.assertEqual(device.state.activity, "idle") + self.assertEqual(await device.request_speed(), 0) + with self.assertRaisesRegex(HettichCentrifugeError, "requires recovery"): + await device.spin(g=device.rpm_to_g(500), duration=30) + + async def test_successful_spin_records_each_workflow_phase(self) -> None: + """State follows confirmed spin phases and returns to idle on success.""" + device = make_device( + [ + *start_replies(), + enquiry_reply("00634", 0x01E8), + enquiry_reply("00635", 0xA292), + enquiry_reply("00604", 500), + enquiry_reply("00602", 12), + ACK_REPLY, + ACK_REPLY, + enquiry_reply("00634", 0x01F0), + enquiry_reply("00635", 0xA292), + *standstill_replies(), + ] + ) + self.schedule_spin_states(device, [30, 40]) + original_write = writes(device).side_effect + snapshots: list[HettichMachineState] = [] + + async def write(data: bytes) -> None: + """Observe workflow phases at serial boundaries without modifying them.""" + snapshots.append(device.state) + await original_write(data) + + writes(device).side_effect = write + await device.spin(g=device.rpm_to_g(500), duration=30, timeout=60) + activities = list(dict.fromkeys(snapshot.activity for snapshot in snapshots)) + self.assertEqual(activities, ["preparing_to_spin", "accelerating", "at_speed", "braking"]) + self.assertEqual(device.state, HettichMachineState(connection="connected")) + + async def test_recovery_is_explicit_and_read_only(self) -> None: + """Closing, reconnecting, and stopping preserve the flag until fresh checks pass.""" + device = make_device([*identity_replies(), *standstill_replies(), *recovery_replies()]) + device._machine.require_recovery() + await device.stop() + await device.setup() + self.assertTrue(device.state.recovery_required) + await device.stop_spin() + self.assertTrue(device.state.recovery_required) + await device.recover() + self.assertEqual(device.state, HettichMachineState(connection="connected")) + self.assertTrue(all(frame[-1] == 5 for frame in telegrams(device))) + + async def test_recovery_rejects_moving_or_faulted_hardware(self) -> None: + """Neither a live speed nor an unstable/faulted hatch can clear recovery.""" + for speed, hatch in ( + (10, 0x1800), + (0, 0x1801), + (0, 0x1C00), + (0, 0x5800), + (0, 0x1810), + (0, 0x3800), + ): + with self.subTest(speed=speed, hatch=hatch): + device = make_device(recovery_replies(speed=speed, hatch=hatch)) + device._machine.require_recovery() + with self.assertRaises(HettichCentrifugeError): + await device.recover() + self.assertTrue(device.state.recovery_required) + self.assertEqual(device.state.activity, "idle") + + async def test_recovery_rejects_identity_changes_and_communication_failures(self) -> None: + """Unverified identity or missing replies cannot clear uncertainty.""" + for replies in ( + [enquiry_reply("00685", 0), enquiry_reply("00600", 0x1234), enquiry_reply("00537", 0xC901)], + [b"", b"", b""], + ): + with self.subTest(replies=replies): + device = make_device(replies) + device._machine.require_recovery() + with self.assertRaises(HettichCentrifugeError): + await device.recover() + self.assertTrue(device.state.recovery_required) + + async def test_failed_recovery_checks_block_motion_even_without_an_existing_flag(self) -> None: + """A failed explicit check cannot leave an unverified machine ready for motion.""" + device = make_device(recovery_replies(speed=100)) + with self.assertRaisesRegex(HettichCentrifugeError, "zero measured rotor speed"): + await device.recover() + with self.assertRaisesRegex(HettichCentrifugeError, "requires recovery"): + await device.open_hatch() + + async def test_recovery_allows_a_confirmed_stationary_positioning_hold(self) -> None: + """A held rotor at its target position is a known state suitable for recovery.""" + device = make_device(recovery_replies(hatch=0x2006)) + device._machine.require_recovery() + await device.recover() + self.assertFalse(device.state.recovery_required) + + async def test_failed_close_does_not_claim_disconnection_or_clear_recovery(self) -> None: + """An uncertain transport close requires another successful close before setup.""" + device = make_device([]) + with patch.object(device.io, "stop", side_effect=OSError("close failed")): + with self.assertRaises(OSError): + await device.stop() + self.assertEqual(device.state.connection, "unknown") + self.assertTrue(device.state.recovery_required) + with self.assertRaisesRegex(HettichCentrifugeError, "closure is uncertain"): + await device.setup() + await device.stop() + self.assertEqual(device.state.connection, "disconnected") + self.assertTrue(device.state.recovery_required)