diff --git a/docs/_static/devices.json b/docs/_static/devices.json index c6ec378ff72..cbf5b07cc70 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -517,6 +517,22 @@ "manager": "https://discuss.pylabrobot.org/u/rickwierenga", "oem": "https://www.hamiltoncompany.com/microlab-vantage" }, + { + "id": "hettich-mikro-220-robotic", + "vendor": "Hettich", + "name": "MIKRO 220 Robotic", + "kind": "centrifuge", + "capabilities": [ + "centrifuging" + ], + "status": "full", + "api": "pylabrobot.hettich.HettichMikro220RoboticCentrifuge", + "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/mikro-220-robotic/" + }, { "id": "hettich-rotanta-460", "vendor": "Hettich", diff --git a/docs/api/pylabrobot.hettich.rst b/docs/api/pylabrobot.hettich.rst new file mode 100644 index 00000000000..c8f94b2f2ee --- /dev/null +++ b/docs/api/pylabrobot.hettich.rst @@ -0,0 +1,24 @@ +.. 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 + RotorSpecification diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index 67dd588713c..57693cd1b43 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 337d48d9d52..1c4ed3f76c4 100644 --- a/docs/contributor_guide/event-schemas.md +++ b/docs/contributor_guide/event-schemas.md @@ -219,7 +219,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`, `duration`; at least one of `relative_centrifugal_force` or `speed_rpm`; optional `acceleration_fraction`, `deceleration_fraction` | Describes one requested spin cycle using the frontend's requested force or speed. | `resources` contains directly loaded resources only. Empty buckets are not represented. `bucket_resources` preserves the association between each loaded resource and its holder: @@ -231,6 +231,11 @@ one is assigned at operation start. } ``` +A frontend without a PLR rotor-resource model emits empty `resources` and `bucket_resources`. +Hettich centrifuges report the requested `speed_rpm`; when a rotor catalog number is configured, +they also report the corresponding calculated `relative_centrifugal_force`. VSpin reports the +requested `relative_centrifugal_force`, `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..4c0e5aa05d4 --- /dev/null +++ b/docs/user_guide/hettich/events.md @@ -0,0 +1,14 @@ +# 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`, `speed_rpm`, `duration`, optional `relative_centrifugal_force` | + +The driver reports the requested rotor speed as `speed_rpm`. If it was constructed with a supported +`rotor_catalog_number`, it also calculates and reports `relative_centrifugal_force`. 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..16d22625c98 --- /dev/null +++ b/docs/user_guide/hettich/index.md @@ -0,0 +1,8 @@ +# Hettich + +```{toctree} +:maxdepth: 1 + +mikro-220/hello-world +events +``` 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..b84b385e94a --- /dev/null +++ b/docs/user_guide/hettich/mikro-220/hello-world.ipynb @@ -0,0 +1,440 @@ +{ + "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-mikro-220-robotic\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.rcf_at_speed(5_000)), \"x g\")\n", + "print(\"RPM for 5,000 x g:\", centrifuge.speed_for_rcf(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: 500 rpm for 10 seconds. `duration` is time at target speed; 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(duration=10, speed=500)" + ] + }, + { + "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/index.md b/docs/user_guide/index.md index b77381922a3..cbf9a497283 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 59d484df5f1..a362108b06e 100644 --- a/docs/user_guide/machine-agnostic-features/event-bus.md +++ b/docs/user_guide/machine-agnostic-features/event-bus.md @@ -125,6 +125,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 | Detailed operation references: @@ -134,6 +135,7 @@ Detailed operation references: - [LiquidHandler](event-bus/liquid-handler.md) - [Shaker and temperature controller](event-bus/thermal-and-shaking.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) diff --git a/pylabrobot/hettich/__init__.py b/pylabrobot/hettich/__init__.py new file mode 100644 index 00000000000..cbb24ad2346 --- /dev/null +++ b/pylabrobot/hettich/__init__.py @@ -0,0 +1,31 @@ +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", + "HettichMikro220RoboticCentrifuge", + "HettichRoboticCentrifuge", + "HettichRotanta460RoboticCentrifuge", + "HettichRotina380RRoboticCentrifuge", + "HettichRotina380RoboticCentrifuge", + "MIKRO_220_ROBOTIC_ROTORS", + "RotorSpecification", +] diff --git a/pylabrobot/hettich/centrifuge.py b/pylabrobot/hettich/centrifuge.py new file mode 100644 index 00000000000..92414b15f8b --- /dev/null +++ b/pylabrobot/hettich/centrifuge.py @@ -0,0 +1,1021 @@ +"""Serial driver for Generation 2 Hettich robotic centrifuges.""" + +import asyncio +import logging +import math +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Literal, Mapping, Optional + +from pylabrobot.events import device_reference, event_operation +from pylabrobot.io.serial import Serial + +logger = logging.getLogger(__name__) + +EOT = 0x04 +STX = 0x02 +ETX = 0x03 +ENQ = 0x05 +ACK = 0x06 +NAK = 0x15 + +ENQUIRY_REPLY_LENGTH = 14 +COMMAND_REPLY_LENGTH = 2 + +# 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_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[\\]") + +_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", +} + +_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 rcf_at_speed(self, speed: int) -> float: + """Return the RCF produced at ``speed`` in rpm.""" + if not 0 <= speed <= self.maximum_speed: + raise ValueError(f"speed must be 0..{self.maximum_speed} rpm") + return self.maximum_rcf * (speed / self.maximum_speed) ** 2 + + def speed_for_rcf(self, rcf: float) -> int: + """Return the nearest speed in rpm that produces ``rcf``.""" + if not 0 <= rcf <= self.maximum_rcf: + raise ValueError(f"rcf must be 0..{self.maximum_rcf}") + return round(self.maximum_speed * math.sqrt(rcf / 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 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})") + + +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: Delay between state requests while waiting for motion. + 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.device_type_code: Optional[int] = None + self.device_type: Optional[str] = None + self.software_version: Optional[str] = None + + 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() + raise + + async def stop(self) -> None: + """Close the serial connection without changing the centrifuge's run state.""" + await self.io.stop() + + @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 an ENQUIRY and return its value, or ``None`` for NAK.""" + frame = self._build_enquiry(parameter) + last_error: Optional[HettichCommunicationError] = None + for attempt in range(1, self.retries + 1): + 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: + await self.io.write(bytes([EOT])) + 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: + 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(): + 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(): + 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) + if self.poll_interval: + await asyncio.sleep(min(self.poll_interval, 0.4)) + 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 rcf_at_speed(self, speed: int) -> float: + """Return RCF at ``speed`` using the configured rotor specification. + + Raises: + HettichCentrifugeError: No rotor catalog number was supplied at construction. + ValueError: ``speed`` 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.rcf_at_speed(speed) + + def speed_for_rcf(self, rcf: float) -> int: + """Return rpm for ``rcf`` using the configured rotor specification. + + Raises: + HettichCentrifugeError: No rotor catalog number was supplied at construction. + ValueError: ``rcf`` 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.speed_for_rcf(rcf) + + 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.""" + 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.""" + 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. + """ + 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.""" + 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. + """ + 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") + + 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: + 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") + + await self._select_parameter(RUN_TIME_PARAMETER, run_time) + await self._select_parameter(SPEED_PARAMETER, speed) + await self._select_parameter(ACTIVATE_PARAMETERS_COMMAND, 0x0001) + + logger.info( + "[Hettich %s] starting centrifugation: run_time=%d seconds, speed=%d rpm", + self.io.port, + run_time, + speed, + ) + await self._select_parameter(SPIN_COMMAND, 0x0002) + + async def _wait_for_standstill(self, timeout: float, motion_observed: bool) -> CentrifugeStatus: + """Wait for standstill, optionally accepting that motion was observed by the caller.""" + if timeout <= 0: + raise ValueError("timeout must be positive") + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + status = await self.request_status() + if status.error_number is not None: + raise HettichCentrifugeError(f"The centrifuge stopped with error {status.error_number}") + if status.phase != "standstill": + motion_observed = True + elif motion_observed: + 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) -> int: + """Wait for ``speed`` and return the device's elapsed run time at that point.""" + if timeout <= 0: + raise ValueError("timeout must be positive") + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + motion_observed = False + while True: + 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: + return await self.request_elapsed_time() + 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, + duration: int, + speed: int, + timeout: Optional[float] = None, + ) -> None: + """Run a finite centrifugation cycle and block until standstill. + + Args: + duration: Time in seconds at the target speed, excluding acceleration and braking. + speed: Rotor speed in rpm, bounded at runtime by the installed rotor. + timeout: Total wait timeout. By default this uses the centrifuge's maximum + configured run-up and run-down times plus a communication margin. + + 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 ``speed``. Timing has one-second + resolution, matching the device protocol. + """ + operation_data: dict[str, Any] = { + "device": device_reference(self, name=self.name), + "resources": [], + "bucket_resources": [], + "speed_rpm": speed, + "duration": duration, + } + if ( + self.rotor_specification is not None and 0 <= speed <= self.rotor_specification.maximum_speed + ): + operation_data["relative_centrifugal_force"] = self.rotor_specification.rcf_at_speed(speed) + + 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") + + 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 + + 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: + 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 = await self._wait_for_target_speed(speed=speed, timeout=remaining) + 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) + except BaseException: + try: + await self.stop_spin() + except BaseException: + logger.exception("[Hettich %s] failed to stop after spin() failed", self.io.port) + raise + + 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 the rotor is already at standstill, this method is a no-op. + """ + 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..e3be9b04efc --- /dev/null +++ b/pylabrobot/hettich/centrifuge_tests.py @@ -0,0 +1,642 @@ +import unittest +from inspect import isabstract +from typing import List, 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], + **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) + return device + + +def make_device(replies: List[bytes], **kwargs) -> HettichMikro220RoboticCentrifuge: + return make_model_device(replies, HettichMikro220RoboticCentrifuge, **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 HettichFrameTests(unittest.TestCase): + def setUp(self) -> None: + self.device = make_device([]) + + 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.rcf_at_speed(13_000), 18_516) + self.assertEqual(rotor.rcf_at_speed(6_500), 4_629) + self.assertEqual(rotor.speed_for_rcf(18_516), 13_000) + self.assertEqual(rotor.speed_for_rcf(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.rcf_at_speed(13_001) + with self.assertRaisesRegex(ValueError, "18516"): + rotor.speed_for_rcf(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.rcf_at_speed(13_000), 18_327) + self.assertEqual(device.speed_for_rcf(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.rcf_at_speed(1_000) + + +class HettichProtocolTests(unittest.IsolatedAsyncioTestCase): + 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, + ) + 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), + ] + ) + 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), + ] + ) + + 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), + ] + ) + + 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), + ] + ) + + await device.spin(duration=30, speed=2000, 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(duration=59_970, speed=2_000) + + self.assertEqual(telegram_parameters(device), [b"00614"]) + self.assertNotIn(b"00521", telegram_parameters(device)) + + 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 HettichEventTests(unittest.IsolatedAsyncioTestCase): + 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) + + with use_event_bus(event_bus): + await device.spin(duration=30, speed=2000, 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.assertEqual(started.data["speed_rpm"], 2000) + self.assertEqual(started.data["duration"], 30) + self.assertAlmostEqual( + cast(float, started.data["relative_centrifugal_force"]), + device.rcf_at_speed(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(0, 2000) + + 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.assertEqual(started.data["speed_rpm"], 2000) + self.assertEqual(started.data["duration"], 0) + self.assertNotIn("relative_centrifugal_force", started.data) + self.assertEqual(failed.data["error_type"], "ValueError") + + +if __name__ == "__main__": + unittest.main()