diff --git a/docs/_static/devices.json b/docs/_static/devices.json
index 901e2ca823c..2dd5dc3d058 100644
--- a/docs/_static/devices.json
+++ b/docs/_static/devices.json
@@ -707,11 +707,26 @@
"heating",
"active cooling"
],
- "status": "mostly",
- "api_version": "v0",
+ "status": "wip",
+ "api_version": "v1",
"doc_slug": "inheco/odtc/hello-world",
"manager": "https://discuss.pylabrobot.org/u/rickwierenga",
- "oem": "https://www.inheco.com/odtc.html"
+ "oem": "https://www.inheco.com/odtc.html",
+ "api": "pylabrobot.inheco.ODTC",
+ "code_slug": "inheco/odtc",
+ "needs_hardware_testing": true,
+ "models": [
+ {
+ "name": "96-well",
+ "status": "wip",
+ "needs_hardware_testing": true
+ },
+ {
+ "name": "384-well",
+ "status": "wip",
+ "needs_hardware_testing": true
+ }
+ ]
},
{
"id": "inheco-scila",
diff --git a/docs/api/pylabrobot.inheco.rst b/docs/api/pylabrobot.inheco.rst
index c149c30299e..290dc3cea4e 100644
--- a/docs/api/pylabrobot.inheco.rst
+++ b/docs/api/pylabrobot.inheco.rst
@@ -59,3 +59,29 @@ SCILA
SCILA
SCILADrawerLoadingTray
DrawerStatus
+
+ODTC
+----
+
+.. currentmodule:: pylabrobot.inheco.odtc
+
+.. autosummary::
+ :toctree: _autosummary
+ :nosignatures:
+
+ ODTC
+ ODTCProtocolStopped
+ ODTCProtocol
+ ODTCStage
+ ODTCStep
+ ODTCOvershoot
+ ODTCPID
+ ODTCMethod
+ ODTCMethodStep
+ ODTCPreMethod
+ ODTCMethodSet
+ ODTCTemperatures
+ ODTCProgress
+ compile_protocol
+ method_set_to_xml
+ parse_method_set
diff --git a/docs/user_guide/inheco/odtc/hello-world.ipynb b/docs/user_guide/inheco/odtc/hello-world.ipynb
index f0828e0ef45..2e11acbcfc8 100644
--- a/docs/user_guide/inheco/odtc/hello-world.ipynb
+++ b/docs/user_guide/inheco/odtc/hello-world.ipynb
@@ -2,42 +2,430 @@
"cells": [
{
"cell_type": "markdown",
- "id": "fa0retc47os",
+ "id": "6c009112",
"metadata": {},
"source": [
"# Inheco ODTC\n",
"\n",
- "The Inheco ODTC (On Deck Thermal Cycler) is a thermocycler designed for automated PCR workflows. It features:\n",
+ "```{device-card} inheco-odtc\n",
+ "```\n",
+ "\n",
+ "| Property | Value |\n",
+ "|---|---|\n",
+ "| Connection | Ethernet, SiLA 1 SOAP; device port 8080 |\n",
+ "| Models | 96-well and 384-well |\n",
+ "| Block temperature | 4–99 °C |\n",
+ "| Lid temperature | 30–110 °C (96); 30–115 °C (384) |\n",
+ "| Units | °C, seconds, microliters |\n",
+ "\n",
+ "**Hardware verification required for this implementation on both variants.** Offline tests cover\n",
+ "protocol compilation and simulated communication; they do not establish thermal accuracy or safe\n",
+ "robotic placement. `setup()` logs a warning until a verified hardware run is reported."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "cd8301ff",
+ "metadata": {},
+ "source": [
+ "## Connection and physical setup\n",
"\n",
- "- Precise temperature control (4 -- 99 °C, up to 4.4 °C/s ramp rate)\n",
- "- Heated lid to prevent condensation\n",
- "- Motorized door for automated plate handling\n",
- "- 96-well plate format\n",
+ "Connect the ODTC controller to Ethernet and power, following the manufacturer's operating manual.\n",
+ "Configure an address reachable from this computer. The instrument sends completion and temperature\n",
+ "callbacks to the computer: allow the selected `client_ip` and `callback_port` through its firewall.\n",
+ "The default callback port is chosen automatically. Supply a fixed port if firewall rules require it.\n",
"\n",
- "The ODTC communicates over Ethernet using the SiLA 2 protocol. You will need the IP address of the device and network connectivity between your computer and the ODTC.\n",
+ "The driver opens no connection during construction. `plate_holder` represents the plate location;\n",
+ "its zero child offset is **not a measured loading position**. Calibrate its placement before robotic\n",
+ "transfers. Load compatible labware manually for an initial hardware check.\n",
+ "\n",
+ "The controls below move and heat real hardware. Run them only when the instrument and labware are ready."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "d2704d01",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "from pylabrobot.inheco import ODTC\n",
+ "from pylabrobot.inheco.odtc import (\n",
+ " ODTCProtocol, ODTCStage, ODTCStep, compile_protocol, method_set_to_xml, ODTCMethodSet,\n",
+ ")\n",
"\n",
- "See the [Inheco ODTC product page](https://www.inheco.com/odtc.html) for hardware details."
+ "odtc = ODTC(\n",
+ " name=\"odtc\",\n",
+ " odtc_ip=\"192.168.1.50\", # use the configured instrument address\n",
+ " variant=96,\n",
+ " # client_ip=\"192.168.1.10\", # computer interface reachable from the ODTC\n",
+ " # callback_port=8081,\n",
+ ")"
]
},
{
"cell_type": "markdown",
+ "id": "2d94097e",
"metadata": {},
"source": [
- "```{device-card} inheco-odtc\n",
- "```"
- ],
- "id": "device-card"
+ "## Set up\n",
+ "\n",
+ "Set up the HTTP connection and callback listener, register the listener, and initialize the device.\n",
+ "Setup refuses to reset a busy or faulted instrument. Calling it again on an established session leaves\n",
+ "the session intact. `simulation_mode=True` selects the instrument firmware's simulation mode; it is\n",
+ "not an offline simulator."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "a00c5aac",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.setup()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "2123ea1f",
+ "metadata": {},
+ "source": [
+ "## Open the drawer\n",
+ "\n",
+ "Open and wait for command completion. Repeating an already confirmed open or close does not issue\n",
+ "another movement. The firmware provides no door-position query: `door_open` is the last\n",
+ "command-confirmed state, or `None` after connection, a failed movement, or a fault."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "3b62fd1f",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.open_door()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "3dd1f560",
+ "metadata": {},
+ "source": [
+ "## Load the plate, then close\n",
+ "\n",
+ "Place the plate on the ODTC mount before running this cell."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "69a37a38",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.close_door()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "0fc5f629",
+ "metadata": {},
+ "source": [
+ "## Read temperatures\n",
+ "\n",
+ "Read all eight sensors in degrees C. This operation is also available while a method is running."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "b699d40b",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "temperatures = await odtc.request_temperatures()\n",
+ "print(temperatures)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "e96c4690",
+ "metadata": {},
+ "source": [
+ "## Define a temperature profile\n",
+ "\n",
+ "`ODTCStep` holds a target temperature and duration. `ODTCStage` repeats an ordered sequence of steps\n",
+ "or nested stages. Temperatures, durations, and ramp rates are checked before upload. Native methods\n",
+ "are limited to 100 steps; deeply nested loops sharing an endpoint may need expansion.\n",
+ "\n",
+ "Lid temperature and maximum sample volume are supplied per run. An explicit step lid temperature\n",
+ "or ramp rate overrides the run's lid default or the variant's transition rate."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "854aa8f4",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "protocol = ODTCProtocol(stages=[\n",
+ " ODTCStage([ODTCStep(95, duration=300)]),\n",
+ " ODTCStage([\n",
+ " ODTCStep(95, duration=30),\n",
+ " ODTCStep(55, duration=30),\n",
+ " ODTCStep(72, duration=60),\n",
+ " ], repeats=30),\n",
+ " ODTCStage([ODTCStep(72, duration=600), ODTCStep(4, duration=30)]),\n",
+ "])"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "f7e00de3",
+ "metadata": {},
+ "source": [
+ "## Inspect the native method without hardware\n",
+ "\n",
+ "Compilation is pure Python. The volume selects the device's compensation range: below 30 µL,\n",
+ "30–below 75 µL, or 75–100 µL. The volume must be positive and no greater than 100 µL.\n",
+ "\n",
+ "Automatic empirical overshoot is **disabled by default**: the coefficients have not been validated\n",
+ "against a manufacturer reference or this implementation's hardware runs. `apply_overshoot=True`\n",
+ "opts in with a warning. A loop entry with different incoming temperatures requires explicit\n",
+ "`ODTCOvershoot` settings because a single calculated excursion would be ambiguous. Its `temperature`\n",
+ "is an excursion magnitude, not an absolute target. Explicit overshoot and PID settings are retained\n",
+ "in native XML round trips. Validate thermal compensation on the actual model and labware."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "e6d8ea5e",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "method = compile_protocol(protocol, variant=96, volume=25, lid_temperature=105)\n",
+ "print(method_set_to_xml(ODTCMethodSet(methods=[method])))"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "cf7824df",
+ "metadata": {},
+ "source": [
+ "## Run: prepare, execute, and wait\n",
+ "\n",
+ "`run_protocol` uploads the method, runs a matching block/lid premethod when necessary, and waits for\n",
+ "completion. Preparation can take several minutes. It waits for firmware completion rather than\n",
+ "merely observing the target on a temperature sensor. Compatible conditions retained from a completed\n",
+ "premethod or post-heated method can be reused within the session.\n",
+ "\n",
+ "The firmware executes timed temperature steps because its command interface has no direct\n",
+ "block-setpoint primitive. PLR performs compilation and sequencing.\n",
+ "\n",
+ "`post_heating=True` keeps regulation at the final step after completion. Pass `False` to disable it.\n",
+ "`timeout` limits each execution wait; a timeout **does not stop the instrument**."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "816c7a8c",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.run_protocol(protocol, volume=25, lid_temperature=105, post_heating=True)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "d32ef9fa",
+ "metadata": {},
+ "source": [
+ "## Read measured progress\n",
+ "\n",
+ "The latest callback provides elapsed time and target, block, and lid temperatures. `None` means no\n",
+ "sample has arrived for this execution. Step numbers, cycle counts, and estimated remaining time are\n",
+ "not reported as measured device state."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "27277702",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "progress = await odtc.request_progress()\n",
+ "print(progress)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "aa379460",
+ "metadata": {},
+ "source": [
+ "## Stop temperature control\n",
+ "\n",
+ "This stops an active method or regulation remaining after a method or premethod. It remains available\n",
+ "while `run_protocol()` is awaiting completion. An interrupted run raises `ODTCProtocolStopped`.\n",
+ "To run concurrently with other work, create an asyncio task for `run_protocol`; canceling that task\n",
+ "only cancels the wait. Call `stop_protocol()` to stop the hardware. A later `wait_for_completion()`\n",
+ "still observes completion errors."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "27e80f0f",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.stop_protocol()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "f35791b3",
+ "metadata": {},
+ "source": [
+ "## Prepare independently\n",
+ "\n",
+ "Use explicit preparation when the block and lid should be ready before loading a plate. The same\n",
+ "preparation is included in `run_protocol`. Temperatures are per-operation arguments."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "408a7ef7",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.prepare(temperature=95, lid_temperature=105)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "6fd596be",
+ "metadata": {},
+ "source": [
+ "## Store a named method\n",
+ "\n",
+ "Compile with a persistent name, then upload. Existing names require `overwrite=True`. The\n",
+ "`plr_currentProtocol` and `plr_currentPreMethod` slots are reserved for the driver's working methods;\n",
+ "unnamed runs replace only these slots. Uploads never clear the instrument's method library."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "c1cd96cf",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "stored_method = compile_protocol(\n",
+ " protocol, name=\"StandardPCR\", variant=96, volume=25, lid_temperature=105,\n",
+ ")\n",
+ "await odtc.upload_protocol(stored_method)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "35cdf928",
+ "metadata": {},
+ "source": [
+ "## Retrieve methods\n",
+ "\n",
+ "Downloaded `ODTCMethod` objects retain native step numbers, loop targets/counts, independent overshoot\n",
+ "rates, PID sets, and metadata. They can be inspected or edited using `dataclasses.replace`, then\n",
+ "uploaded explicitly. `request_method_set()` includes premethods too."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "467d4b2c",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "stored = await odtc.request_protocol(\"StandardPCR\")\n",
+ "print(stored)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "023dd84f",
+ "metadata": {},
+ "source": [
+ "## Run a stored method\n",
+ "\n",
+ "Use its stored volume, lid, and ramp settings. PLR prepares matching initial thermal conditions\n",
+ "before executing it."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "7d2780db",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.run_stored_protocol(\"StandardPCR\")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "982ed168",
+ "metadata": {},
+ "source": [
+ "## Disconnect\n",
+ "\n",
+ "`stop()` stops temperature control and closes the HTTP connection and callback server. Transport\n",
+ "cleanup also runs if the stop command fails; the exception still reports that the device did not\n",
+ "confirm stopping."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "8d38a4a9",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "await odtc.stop()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "3cff7c6b",
+ "metadata": {},
+ "source": [
+ "## Variant verification and protocol references\n",
+ "\n",
+ "Choose `variant=384` for a 384-well unit. Its current compiler limits are 5 °C/s heating and 115 °C lid,\n",
+ "compared with 4.4 °C/s and 110 °C for the 96-well unit. Both use a 2.2 °C/s cooling limit. The variant\n",
+ "codes, plate types, volume compensation, and limits need confirmation on each hardware variant;\n",
+ "passing mock tests is not hardware validation.\n",
+ "\n",
+ "The wire format, method storage, premethod sequence, callback sample units, and return codes are\n",
+ "covered by the [INHECO Firmware Command Set v1.06](https://github.com/user-attachments/files/26010799/TD_SILA_FWCommandSet.25210.pdf)\n",
+ "(sections 3, 5, 6, 7.3, 9, and 10). This 2016 document does not establish which firmware revision is\n",
+ "installed on a particular device. Record the instrument and firmware version when reporting a\n",
+ "[hardware verification run](../../needs-testing.md). The tests use its example method and data series\n",
+ "as fixtures, alongside simulated SOAP exchanges. Thermal overshoot coefficients are not established\n",
+ "by that command-set document."
+ ]
}
],
"metadata": {
"kernelspec": {
- "display_name": "Python 3 (ipykernel)",
+ "display_name": "Python 3",
"language": "python",
"name": "python3"
- },
- "language_info": {
- "name": "python",
- "version": "3.12.0"
}
},
"nbformat": 4,
diff --git a/pylabrobot/inheco/__init__.py b/pylabrobot/inheco/__init__.py
index 95af4f51c0e..b66f5508fe7 100644
--- a/pylabrobot/inheco/__init__.py
+++ b/pylabrobot/inheco/__init__.py
@@ -1,5 +1,6 @@
from .control_box import InhecoTECControlBox
from .cpac import InhecoCPAC, inheco_cpac_ultraflat
+from .odtc import ODTC
from .temperature_controller import InhecoTemperatureController
from .thermoshake import (
InhecoThermoShake,
diff --git a/pylabrobot/inheco/odtc/__init__.py b/pylabrobot/inheco/odtc/__init__.py
new file mode 100644
index 00000000000..b159e8b9131
--- /dev/null
+++ b/pylabrobot/inheco/odtc/__init__.py
@@ -0,0 +1,19 @@
+"""Inheco ODTC thermocycler and native protocol data."""
+
+from .odtc import ODTC, ODTCProtocolStopped
+from .protocol import (
+ ODTCPID,
+ ODTCMethod,
+ ODTCMethodSet,
+ ODTCMethodStep,
+ ODTCOvershoot,
+ ODTCPreMethod,
+ ODTCProgress,
+ ODTCProtocol,
+ ODTCStage,
+ ODTCStep,
+ ODTCTemperatures,
+ ODTCVariant,
+ compile_protocol,
+)
+from .xml import method_set_to_xml, parse_method_set
diff --git a/pylabrobot/inheco/odtc/odtc.py b/pylabrobot/inheco/odtc/odtc.py
new file mode 100644
index 00000000000..d03c423d769
--- /dev/null
+++ b/pylabrobot/inheco/odtc/odtc.py
@@ -0,0 +1,456 @@
+"""Inheco ODTC thermocycler operations over SiLA 1."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import math
+import xml.etree.ElementTree as ET
+
+from pylabrobot.inheco.odtc.protocol import (
+ DEFAULT_APPLY_OVERSHOOT,
+ DEFAULT_PLATE_TYPE,
+ DEFAULT_POST_HEATING,
+ ODTCPID,
+ ODTCMethod,
+ ODTCMethodSet,
+ ODTCPreMethod,
+ ODTCProgress,
+ ODTCProtocol,
+ ODTCTemperatures,
+ ODTCVariant,
+ compile_protocol,
+ validate_method,
+ validate_temperature,
+ validate_variant,
+)
+from pylabrobot.inheco.odtc.xml import (
+ method_set_to_xml,
+ parse_method_set,
+ parse_progress,
+ parse_sensor_values,
+)
+from pylabrobot.inheco.transport.sila import (
+ InhecoSiLAInterface,
+ SiLACommand,
+ SiLAError,
+ get_params,
+ required_text,
+)
+from pylabrobot.resources import Coordinate, ResourceHolder
+
+logger = logging.getLogger(__name__)
+
+
+class ODTCProtocolStopped(RuntimeError):
+ """The user stopped a protocol before it completed."""
+
+
+class ODTC:
+ """An ODTC, its plate location, and its connection and thermal execution state.
+
+ Temperatures use degrees C, durations seconds, and sample volumes microliters.
+ The plate holder's position must be calibrated before using it for robotic transfers.
+ Both 96- and 384-well implementations require hardware verification.
+ """
+
+ _door_open: bool | None
+ _prepared: tuple[float, float] | None
+ _execution: SiLACommand | None
+ _progress_by_request: dict[int, ODTCProgress]
+ _workflow_lock: asyncio.Lock | None
+
+ @property
+ def door_open(self) -> bool | None:
+ """Last command-confirmed door state; None means unknown, not a sensor reading."""
+ return self._door_open
+
+ def _on_data(self, request_id: int, data: str) -> None:
+ """Store one measured sample, including callbacks that precede HTTP acceptance."""
+ current = self._execution
+ if self._sila.pending_command_name(request_id) != "ExecuteMethod" and (
+ current is None or current.request_id != request_id
+ ):
+ return
+ self._progress_by_request[request_id] = parse_progress(data)
+
+ def _on_status(self, event: ET.Element) -> None:
+ """Surface device-wide faults and invalidate state that can no longer be trusted."""
+ description = event.find("{*}eventDescription")
+ if description is None:
+ raise ValueError("StatusEvent is missing eventDescription")
+ if description.text and description.text.strip():
+ description = ET.fromstring(description.text)
+ state = (
+ description.text
+ if description.tag.rsplit("}", 1)[-1] == "DeviceState"
+ else required_text(description, "DeviceState")
+ )
+ logger.debug("ODTC %s state: %s", self.name, state)
+ if state is None or state.lower() not in ("errorhandling", "inerror"):
+ return
+ self._prepared = None
+ self._door_open = None
+ extensions = description.find("{*}Extensions")
+ values = [child.text or "" for child in extensions] if extensions is not None else []
+ code = int(values[1]) if len(values) >= 2 and values[1] else 9
+ message = values[4] if len(values) >= 5 else f"Device entered {state}"
+ logger.error("ODTC %s fault %d: %s", self.name, code, message)
+ self._sila.fail_pending(SiLAError(code, message, "StatusEvent"))
+
+ def __init__(
+ self,
+ name: str,
+ odtc_ip: str,
+ variant: ODTCVariant = 96,
+ client_ip: str | None = None,
+ *,
+ odtc_port: int = 8080,
+ callback_port: int = 0,
+ ) -> None:
+ self.name = name
+ self.variant = validate_variant(variant)
+ self.plate_holder = ResourceHolder(
+ name=f"{name}_plate_holder",
+ size_x=127.76,
+ size_y=85.48,
+ size_z=0,
+ child_location=Coordinate.zero(),
+ )
+ self._sila = InhecoSiLAInterface(
+ machine_ip=odtc_ip,
+ client_ip=client_ip,
+ machine_port=odtc_port,
+ callback_port=callback_port,
+ on_data=self._on_data,
+ on_status=self._on_status,
+ )
+ self._connected = False
+ self._door_open = None
+ self._prepared = None
+ self._execution = None
+ self._progress_by_request = {}
+ self._workflow_lock = None
+ self._stop_generation = 0
+
+ def _require_connection(self) -> None:
+ """Require a successfully initialized session."""
+ if not self._connected:
+ raise RuntimeError("ODTC is not set up")
+
+ def _require_workflow_lock(self) -> asyncio.Lock:
+ """Protect multi-command preparation/upload sequences, leaving reads and stop available."""
+ self._require_connection()
+ assert self._workflow_lock is not None
+ return self._workflow_lock
+
+ async def request_status(self) -> str:
+ """Read the current SiLA state from the device."""
+ response = await self._sila.send_command("GetStatus")
+ state = response["GetStatusResponse"]["state"]
+ if not isinstance(state, str):
+ raise ValueError("GetStatus returned a non-string state")
+ return state
+
+ async def setup(self, *, simulation_mode: bool = False) -> None:
+ """Connect, register callbacks, and initialize an idle or standby instrument.
+
+ A busy device is left running and setup raises. Repeated setup on an established
+ session is a no-op. Simulation mode is the firmware's simulation mode.
+ """
+ if self._connected:
+ return
+ logger.warning(
+ "ODTC %d driver needs hardware verification; please report a verified run", self.variant
+ )
+ await self._sila.setup()
+ try:
+ state = await self.request_status()
+ if state.lower() not in ("standby", "idle"):
+ raise RuntimeError(f"ODTC is {state}; setup will not reset an active or faulted device")
+ await self._sila.send_command(
+ "Reset",
+ deviceId=self.name,
+ eventReceiverURI=self._sila.event_receiver_uri,
+ simulationMode=simulation_mode,
+ )
+ state = await self.request_status()
+ if state.lower() == "standby":
+ await self._sila.send_command("Initialize")
+ state = await self.request_status()
+ if state.lower() != "idle":
+ raise RuntimeError(f"ODTC did not initialize: {state}")
+ self._door_open = None
+ self._prepared = None
+ self._execution = None
+ self._progress_by_request = {}
+ self._workflow_lock = asyncio.Lock()
+ self._connected = True
+ logger.info("ODTC %s connected", self.name)
+ except BaseException:
+ await self._sila.close()
+ raise
+
+ async def open_door(self) -> None:
+ """Open the drawer and await command completion; remember only confirmed state."""
+ self._require_connection()
+ if self._door_open is True:
+ return
+ self._door_open = None
+ await self._sila.send_command("OpenDoor")
+ self._door_open = True
+ logger.info("ODTC %s door opened", self.name)
+
+ async def close_door(self) -> None:
+ """Close the drawer and await command completion."""
+ self._require_connection()
+ if self._door_open is False:
+ return
+ self._door_open = None
+ await self._sila.send_command("CloseDoor")
+ self._door_open = False
+ logger.info("ODTC %s door closed", self.name)
+
+ async def request_temperatures(self) -> ODTCTemperatures:
+ """Read all physical temperature sensors, including during a run."""
+ self._require_connection()
+ response = await self._sila.send_command("ReadActualTemperature")
+ if not isinstance(response, ET.Element):
+ raise ValueError("ReadActualTemperature did not return ResponseData")
+ value = get_params(response, ["SensorValues"])["SensorValues"]
+ if not isinstance(value, str):
+ raise ValueError("SensorValues must contain XML text")
+ return parse_sensor_values(value)
+
+ async def request_block_temperature(self) -> float:
+ """Read the block's mount temperature in degrees C."""
+ return (await self.request_temperatures()).mount
+
+ async def request_lid_temperature(self) -> float:
+ """Read the heated lid temperature in degrees C."""
+ return (await self.request_temperatures()).lid
+
+ async def request_progress(self) -> ODTCProgress | None:
+ """Return the latest measured sample for this execution, or None before any data arrives."""
+ if self._execution is None:
+ return None
+ return self._progress_by_request.get(self._execution.request_id)
+
+ async def request_method_set(self) -> ODTCMethodSet:
+ """Retrieve the full stored methods, including native loops and PID settings."""
+ self._require_connection()
+ response = await self._sila.send_command("GetParameters")
+ if not isinstance(response, ET.Element):
+ raise ValueError("GetParameters did not return ResponseData")
+ xml = get_params(response, ["MethodsXML"])["MethodsXML"]
+ if not isinstance(xml, str):
+ raise ValueError("MethodsXML must contain XML text")
+ return parse_method_set(xml)
+
+ async def request_protocol(self, name: str) -> ODTCMethod | None:
+ """Retrieve a named compiled method, or None if it is absent."""
+ methods = await self.request_method_set()
+ return next((method for method in methods.methods if method.name == name), None)
+
+ async def _upload(self, methods: ODTCMethodSet, *, dynamic_pre_method_duration: bool) -> None:
+ """Upload explicit settings and methods without deleting unrelated device storage."""
+ for method in methods.methods:
+ if method.variant != self.variant:
+ raise ValueError("Method variant does not match the ODTC")
+ root = ET.Element("ParameterSet")
+ param = ET.SubElement(root, "Parameter", name="MethodsXML", parameterType="String")
+ ET.SubElement(param, "String").text = method_set_to_xml(methods)
+ for name, value in {
+ "DynamicPreMethodDuration": dynamic_pre_method_duration,
+ "ExecuteMethodDataEvent": True,
+ "MatchLidTemperatures": True,
+ }.items():
+ param = ET.SubElement(root, "Parameter", name=name, parameterType="Boolean")
+ ET.SubElement(param, "Boolean").text = str(value).lower()
+ await self._sila.send_command("SetParameters", paramsXML=ET.tostring(root, encoding="unicode"))
+
+ async def _check_name(self, name: str, overwrite: bool) -> None:
+ """Require an explicit overwrite for persistent methods, including premethod collisions."""
+ if overwrite:
+ return
+ existing = await self.request_method_set()
+ names = {method.name for method in existing.methods} | {
+ method.name for method in existing.premethods
+ }
+ if name in names:
+ raise ValueError(f"Method {name!r} already exists; pass overwrite=True to replace it")
+
+ async def upload_protocol(self, method: ODTCMethod, *, overwrite: bool = False) -> None:
+ """Persist a compiled method. Use compile_protocol to inspect the method before upload."""
+ validate_method(method)
+ async with self._require_workflow_lock():
+ await self._check_name(method.name, overwrite)
+ await self._upload(ODTCMethodSet(methods=[method]), dynamic_pre_method_duration=True)
+
+ async def wait_for_completion(self, *, timeout: float | None = None) -> None:
+ """Wait for the latest execution, retaining completed failures and shielding the device task.
+
+ Cancellation or timeout stops only this wait. Call stop_protocol to stop the instrument.
+ """
+ execution = self._execution
+ if execution is not None:
+ await execution.wait(timeout)
+
+ def _validate_timeout(self, timeout: float | None) -> None:
+ """Reject invalid waits before starting a hardware operation."""
+ if timeout is not None and (not math.isfinite(timeout) or timeout <= 0):
+ raise ValueError("timeout must be finite and positive")
+
+ def _check_stopped(self, generation: int) -> None:
+ """Prevent a stopped preparation workflow from starting its following method."""
+ if generation != self._stop_generation:
+ raise ODTCProtocolStopped("ODTC workflow was stopped")
+
+ async def _execute(self, name: str, timeout: float | None, generation: int) -> None:
+ """Execute a named method and retain its result for subsequent waiters."""
+ self._progress_by_request = {}
+ self._check_stopped(generation)
+ self._execution = await self._sila.start_command("ExecuteMethod", methodName=name)
+ if generation != self._stop_generation and not self._execution.future.done():
+ self._execution.future.set_exception(ODTCProtocolStopped("ODTC protocol was stopped"))
+ self._check_stopped(generation)
+ logger.info("ODTC %s executing %s", self.name, name)
+ await self.wait_for_completion(timeout=timeout)
+ self._check_stopped(generation)
+ logger.info("ODTC %s completed %s", self.name, name)
+
+ async def _prepare(
+ self,
+ temperature: float,
+ lid_temperature: float,
+ timeout: float | None,
+ dynamic_pre_method_duration: bool,
+ generation: int,
+ ) -> None:
+ """Establish homogeneous temperatures by waiting for a completed premethod."""
+ validate_temperature(temperature, lid=False, variant=self.variant)
+ validate_temperature(lid_temperature, lid=True, variant=self.variant)
+ targets = (temperature, lid_temperature)
+ self._check_stopped(generation)
+ if self._prepared == targets:
+ return
+ self._prepared = None
+ premethod = ODTCPreMethod("plr_currentPreMethod", temperature, lid_temperature)
+ await self._upload(
+ ODTCMethodSet(premethods=[premethod]), dynamic_pre_method_duration=dynamic_pre_method_duration
+ )
+ await self._execute(premethod.name, timeout, generation)
+ self._prepared = targets
+
+ async def prepare(
+ self,
+ temperature: float,
+ lid_temperature: float,
+ *,
+ timeout: float | None = None,
+ dynamic_pre_method_duration: bool = True,
+ ) -> None:
+ """Preheat before loading or running; return only after thermal preparation completes."""
+ self._validate_timeout(timeout)
+ generation = self._stop_generation
+ async with self._require_workflow_lock():
+ try:
+ await self._prepare(
+ temperature, lid_temperature, timeout, dynamic_pre_method_duration, generation
+ )
+ except BaseException:
+ self._prepared = None
+ raise
+
+ async def _run(
+ self, method: ODTCMethod, timeout: float | None, dynamic: bool, generation: int
+ ) -> None:
+ """Prepare matching thermal conditions and run an uploaded method."""
+ try:
+ await self._prepare(
+ method.start_temperature, method.start_lid_temperature, timeout, dynamic, generation
+ )
+ self._prepared = None
+ await self._execute(method.name, timeout, generation)
+ if method.post_heating:
+ final_step = method.steps[-1]
+ self._prepared = (final_step.temperature, final_step.lid_temperature)
+ except BaseException:
+ self._prepared = None
+ raise
+
+ async def run_protocol(
+ self,
+ protocol: ODTCProtocol,
+ *,
+ volume: float,
+ lid_temperature: float,
+ plate_type: int = DEFAULT_PLATE_TYPE,
+ post_heating: bool = DEFAULT_POST_HEATING,
+ apply_overshoot: bool = DEFAULT_APPLY_OVERSHOOT,
+ pid_set: list[ODTCPID] | None = None,
+ overwrite: bool = False,
+ timeout: float | None = None,
+ dynamic_pre_method_duration: bool = True,
+ ) -> None:
+ """Compile, upload, prepare, and await a complete temperature protocol.
+
+ A name in the profile selects persistent storage; otherwise the reserved
+ plr_currentProtocol slot is replaced. Timeout applies to each execution wait.
+ Waiting may be canceled without stopping the physical method; stop_protocol does that.
+ """
+ method = compile_protocol(
+ protocol,
+ variant=self.variant,
+ volume=volume,
+ lid_temperature=lid_temperature,
+ plate_type=plate_type,
+ post_heating=post_heating,
+ apply_overshoot=apply_overshoot,
+ pid_set=pid_set,
+ )
+ self._validate_timeout(timeout)
+ generation = self._stop_generation
+ async with self._require_workflow_lock():
+ self._check_stopped(generation)
+ await self._check_name(method.name, overwrite or protocol.name is None)
+ await self._upload(
+ ODTCMethodSet(methods=[method]), dynamic_pre_method_duration=dynamic_pre_method_duration
+ )
+ await self._run(method, timeout, dynamic_pre_method_duration, generation)
+
+ async def run_stored_protocol(self, name: str, *, timeout: float | None = None) -> None:
+ """Prepare and execute an existing method using its stored thermal settings."""
+ self._validate_timeout(timeout)
+ generation = self._stop_generation
+ async with self._require_workflow_lock():
+ self._check_stopped(generation)
+ method = await self.request_protocol(name)
+ if method is None:
+ raise ValueError(f"No stored protocol named {name!r}")
+ if method.variant != self.variant:
+ raise ValueError("Stored method variant does not match the ODTC")
+ await self._run(method, timeout, True, generation)
+
+ async def stop_protocol(self) -> None:
+ """Stop a running method or an idle post-method hold and await acknowledgement."""
+ self._require_connection()
+ self._stop_generation += 1
+ self._prepared = None
+ await self._sila.send_command("StopMethod")
+ execution = self._execution
+ if execution is not None and not execution.future.done():
+ execution.future.set_exception(ODTCProtocolStopped("ODTC protocol was stopped"))
+ logger.info("ODTC %s temperature control stopped", self.name)
+
+ async def stop(self) -> None:
+ """Stop temperature control and close transports, even if stopping the device fails."""
+ try:
+ if self._connected:
+ await self.stop_protocol()
+ finally:
+ self._connected = False
+ self._door_open = None
+ self._prepared = None
+ await self._sila.close()
+ logger.info("ODTC %s disconnected", self.name)
diff --git a/pylabrobot/inheco/odtc/protocol.py b/pylabrobot/inheco/odtc/protocol.py
new file mode 100644
index 00000000000..488518a361c
--- /dev/null
+++ b/pylabrobot/inheco/odtc/protocol.py
@@ -0,0 +1,406 @@
+"""ODTC temperature profiles and compilation to the instrument's method format."""
+
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass, field, replace
+from typing import Literal
+
+logger = logging.getLogger(__name__)
+
+ODTCVariant = Literal[96, 384]
+DEFAULT_PLATE_TYPE = 0
+DEFAULT_POST_HEATING = True
+DEFAULT_APPLY_OVERSHOOT = False
+
+
+@dataclass(frozen=True)
+class ODTCOvershoot:
+ """A temperature excursion relative to a plateau, in degrees C and seconds.
+
+ `temperature` is the positive excursion magnitude, not an absolute setpoint. The
+ controller chooses its direction from the temperature transition.
+ """
+
+ temperature: float
+ duration: float = 0
+ return_rate: float = 2.2
+ approach_rate: float | None = None
+
+
+@dataclass(frozen=True)
+class ODTCStep:
+ """Hold a block temperature for `duration` seconds, with optional ramp and lid settings."""
+
+ temperature: float
+ duration: float
+ ramp_rate: float | None = None
+ lid_temperature: float | None = None
+ overshoot: ODTCOvershoot | None = None
+ pid_number: int = 1
+
+
+@dataclass(frozen=True)
+class ODTCStage:
+ """An ordered sequence of steps or nested stages, repeated a fixed number of times."""
+
+ steps: list[ODTCStep | ODTCStage]
+ repeats: int = 1
+
+
+@dataclass(frozen=True)
+class ODTCProtocol:
+ """A temperature profile; an omitted name selects the driver's working method slot."""
+
+ stages: list[ODTCStage]
+ name: str | None = None
+
+
+@dataclass(frozen=True)
+class ODTCPID:
+ """Device PID parameters; defaults match the firmware manual's example PID set."""
+
+ number: int = 1
+ p_heating: float = 60
+ p_cooling: float = 80
+ i_heating: float = 250
+ i_cooling: float = 100
+ d_heating: float = 10
+ d_cooling: float = 10
+ p_lid: float = 100
+ i_lid: float = 70
+
+
+@dataclass(frozen=True)
+class ODTCMethodStep:
+ """A native step including its backward jump and additional repeat count."""
+
+ number: int
+ temperature: float
+ duration: float
+ ramp_rate: float
+ lid_temperature: float
+ overshoot: ODTCOvershoot
+ goto: int = 0
+ loops: int = 0
+ pid_number: int = 1
+
+
+@dataclass(frozen=True)
+class ODTCMethod:
+ """A compiled or downloaded method with explicit device settings and native steps."""
+
+ name: str
+ variant: ODTCVariant
+ plate_type: int
+ fluid_quantity: int
+ post_heating: bool
+ start_temperature: float
+ start_lid_temperature: float
+ steps: list[ODTCMethodStep]
+ pid_set: list[ODTCPID]
+ creator: str | None = None
+ description: str | None = None
+ date_time: str | None = None
+
+
+@dataclass(frozen=True)
+class ODTCPreMethod:
+ """A stored preparation method that establishes block and lid temperatures."""
+
+ name: str
+ temperature: float
+ lid_temperature: float
+ creator: str | None = None
+ description: str | None = None
+ date_time: str | None = None
+
+
+@dataclass(frozen=True)
+class ODTCMethodSet:
+ """Methods and premethods returned by the instrument or uploaded together."""
+
+ methods: list[ODTCMethod] = field(default_factory=list)
+ premethods: list[ODTCPreMethod] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class ODTCTemperatures:
+ """Temperature sensor values in degrees C, plus the device's timestamp."""
+
+ mount: float
+ mount_monitor: float
+ lid: float
+ lid_monitor: float
+ ambient: float
+ pcb: float
+ heatsink: float
+ heatsink_tec: float
+ timestamp: str | None = None
+
+
+@dataclass(frozen=True)
+class ODTCProgress:
+ """The latest device-reported sample; elapsed time is in seconds, temperatures in C."""
+
+ elapsed: float
+ target_temperature: float
+ current_temperature: float
+ lid_temperature: float
+
+
+def validate_variant(variant: int) -> ODTCVariant:
+ """Accept the two public well-count variants, without wire-code aliases."""
+ if variant == 96:
+ return 96
+ if variant == 384:
+ return 384
+ raise ValueError("variant must be 96 or 384")
+
+
+def validate_temperature(temperature: float, *, lid: bool, variant: ODTCVariant) -> None:
+ """Check the supported block or lid temperature range."""
+ low, high = (30, 110 if variant == 96 else 115) if lid else (4, 99)
+ if not math.isfinite(temperature) or not low <= temperature <= high:
+ raise ValueError(f"{'lid' if lid else 'block'} temperature must be between {low} and {high}")
+
+
+def volume_to_fluid_quantity(volume: float) -> int:
+ """Convert maximum sample volume in microliters to a native compensation range."""
+ if not math.isfinite(volume) or not 0 < volume <= 100:
+ raise ValueError("volume must be greater than zero and at most 100 microliters")
+ return 0 if volume < 30 else 1 if volume < 75 else 2
+
+
+def calculate_overshoot(
+ temperature: float, previous_temperature: float, rate: float, duration: float, fluid_quantity: int
+) -> ODTCOvershoot:
+ """Compute the optional empirical volume compensation; hardware validation is required."""
+ heating = temperature > previous_temperature
+ delta = abs(temperature - previous_temperature)
+ if duration == 0 or rate <= 0.5 or temperature <= 35 or delta <= (5 if heating else 10):
+ return ODTCOvershoot(0, approach_rate=rate)
+ heat = ((5.1144, -6.6037), (19.469, -20.875), (22.829, -13.278))
+ cool = ((4.0941, -6.1247), (4.9773, -8.0866), (9.0513, -17.015))
+ if fluid_quantity not in (0, 1, 2):
+ return ODTCOvershoot(0, approach_rate=rate)
+ a, b = (heat if heating else cool)[fluid_quantity]
+ energy = a * math.log(min(delta, 60)) + b
+ if energy <= 0:
+ return ODTCOvershoot(0, approach_rate=rate)
+ excursion = math.sqrt(2 * energy / (1 / rate + 1 / 2.2))
+ hold = 0.0
+ if heating and temperature + excursion > 102:
+ excursion = 102 - temperature
+ triangle_time = excursion / rate + excursion / 2.2
+ hold = (energy - 0.5 * triangle_time * excursion) / excursion
+ return ODTCOvershoot(round(excursion, 1), round(hold, 1), approach_rate=rate)
+
+
+def _first_step(stage: ODTCStage) -> ODTCStep:
+ """Find the first executable step without inventing a start temperature."""
+ if not stage.steps:
+ raise ValueError("Stages must not be empty")
+ first = stage.steps[0]
+ return _first_step(first) if isinstance(first, ODTCStage) else first
+
+
+def _validate_method_step(
+ step: ODTCMethodStep, variant: ODTCVariant, pid_numbers: set[int]
+) -> None:
+ """Check a native step before uploading it."""
+ validate_temperature(step.temperature, lid=False, variant=variant)
+ validate_temperature(step.lid_temperature, lid=True, variant=variant)
+ if not math.isfinite(step.duration) or step.duration < 0:
+ raise ValueError("Step duration must be finite and nonnegative")
+ maximum_rate = 4.4 if variant == 96 else 5.0
+ if not math.isfinite(step.ramp_rate) or not 0.1 <= step.ramp_rate <= maximum_rate:
+ raise ValueError(f"Ramp rate must be between 0.1 and {maximum_rate}")
+ overshoot = step.overshoot
+ for value in (overshoot.temperature, overshoot.duration):
+ if not math.isfinite(value) or value < 0:
+ raise ValueError("Overshoot magnitude and duration must be finite and nonnegative")
+ for rate in (overshoot.return_rate, overshoot.approach_rate):
+ if rate is not None and (not math.isfinite(rate) or not 0.1 <= rate <= maximum_rate):
+ raise ValueError("Invalid overshoot ramp rate")
+ if step.pid_number not in pid_numbers:
+ raise ValueError(f"Step references undefined PID {step.pid_number}")
+
+
+def _incoming_temperatures(steps: list[ODTCMethodStep], start: float) -> list[set[float]]:
+ """Collect the temperatures that can precede each step, including loop-back edges."""
+ incoming = [{start}] + [{step.temperature} for step in steps[:-1]]
+ for step in steps:
+ if step.goto:
+ incoming[step.goto - 1].add(step.temperature)
+ return incoming
+
+
+def validate_method(method: ODTCMethod) -> None:
+ """Validate a compiled or edited native method before device communication."""
+ validate_variant(method.variant)
+ if method.name == "plr_currentPreMethod":
+ raise ValueError("plr_currentPreMethod is reserved for preparation")
+ if not method.name.strip():
+ raise ValueError("Method name must not be empty")
+ if method.plate_type not in ((0,) if method.variant == 96 else (0, 2)):
+ raise ValueError("Unsupported plate_type for this variant")
+ if method.fluid_quantity not in (-1, 0, 1, 2):
+ raise ValueError("Invalid fluid quantity code")
+ validate_temperature(method.start_temperature, lid=False, variant=method.variant)
+ validate_temperature(method.start_lid_temperature, lid=True, variant=method.variant)
+ if not 1 <= len(method.steps) <= 100:
+ raise ValueError("ODTC methods require between 1 and 100 native steps")
+ pid_numbers = {pid.number for pid in method.pid_set}
+ if len(pid_numbers) != len(method.pid_set) or not 1 <= len(pid_numbers) <= 9:
+ raise ValueError("A method needs 1 to 9 uniquely numbered PID sets")
+ for pid in method.pid_set:
+ if not 1 <= pid.number <= 9:
+ raise ValueError("PID number must be between 1 and 9")
+ values = (
+ pid.p_heating,
+ pid.p_cooling,
+ pid.i_heating,
+ pid.i_cooling,
+ pid.d_heating,
+ pid.d_cooling,
+ pid.p_lid,
+ pid.i_lid,
+ )
+ if any(not math.isfinite(value) or value < 0 for value in values):
+ raise ValueError("PID coefficients must be finite and nonnegative")
+ for number, step in enumerate(method.steps, 1):
+ if any(
+ not isinstance(value, int) or isinstance(value, bool)
+ for value in (step.number, step.goto, step.loops, step.pid_number)
+ ):
+ raise ValueError("Step numbers, loop counts, and PID numbers must be integers")
+ if step.number != number:
+ raise ValueError("Method steps must be numbered consecutively from 1")
+ if step.goto < 0 or step.goto > number or step.loops < 0:
+ raise ValueError("Invalid backward loop")
+ if bool(step.goto) != bool(step.loops):
+ raise ValueError("A loop needs both a target and an additional repeat count")
+ _validate_method_step(step, method.variant, pid_numbers)
+ for step, incoming in zip(
+ method.steps, _incoming_temperatures(method.steps, method.start_temperature)
+ ):
+ if any(previous > step.temperature for previous in incoming) and step.ramp_rate > 2.2:
+ raise ValueError("Cooling transitions, including loop entries, cannot exceed 2.2 degrees C/s")
+
+
+def compile_protocol(
+ protocol: ODTCProtocol,
+ *,
+ variant: ODTCVariant,
+ volume: float,
+ lid_temperature: float,
+ plate_type: int = DEFAULT_PLATE_TYPE,
+ post_heating: bool = DEFAULT_POST_HEATING,
+ apply_overshoot: bool = DEFAULT_APPLY_OVERSHOOT,
+ pid_set: list[ODTCPID] | None = None,
+ name: str | None = None,
+) -> ODTCMethod:
+ """Compile a profile without I/O, retaining loops and explicit per-step settings.
+
+ Automatic overshoot is opt-in because the empirical coefficients have not been
+ validated for this implementation. Explicit excursions are always retained.
+ """
+ validate_variant(variant)
+ quantity = volume_to_fluid_quantity(volume)
+ if not protocol.stages:
+ raise ValueError("Protocol must contain at least one stage")
+ if protocol.name in ("plr_currentProtocol", "plr_currentPreMethod"):
+ raise ValueError("Names beginning plr_current are reserved for working methods")
+ first = _first_step(protocol.stages[0])
+ start_lid = first.lid_temperature if first.lid_temperature is not None else lid_temperature
+ if apply_overshoot:
+ logger.warning(
+ "Automatic ODTC overshoot requires validation on the selected variant and labware"
+ )
+ result: list[ODTCMethodStep] = []
+ source_steps: list[ODTCStep] = []
+
+ def append_stage(stage: ODTCStage) -> None:
+ """Compile a stage, expanding coincident loop endpoints when necessary."""
+ if not isinstance(stage.repeats, int) or isinstance(stage.repeats, bool) or stage.repeats < 1:
+ raise ValueError("Stage repeats must be a positive integer")
+ if not stage.steps:
+ raise ValueError("Stages must not be empty")
+ start = len(result)
+ for item in stage.steps:
+ if isinstance(item, ODTCStage):
+ append_stage(item)
+ continue
+ validate_temperature(item.temperature, lid=False, variant=variant)
+ rate = item.ramp_rate if item.ramp_rate is not None else (4.4 if variant == 96 else 5.0)
+ overshoot = item.overshoot or ODTCOvershoot(0)
+ result.append(
+ ODTCMethodStep(
+ number=len(result) + 1,
+ temperature=item.temperature,
+ duration=item.duration,
+ ramp_rate=rate,
+ lid_temperature=lid_temperature if item.lid_temperature is None else item.lid_temperature,
+ overshoot=overshoot,
+ pid_number=item.pid_number,
+ )
+ )
+ source_steps.append(item)
+ if len(result) > 100:
+ raise ValueError("Compiled method exceeds the ODTC's 100-step limit")
+ if stage.repeats > 1:
+ if result[-1].goto:
+ block = result[start:]
+ source_block = source_steps[start:]
+ if len(result) + len(block) * (stage.repeats - 1) > 100:
+ raise ValueError("Nested loops sharing an endpoint exceed the ODTC's 100-step limit")
+ for _ in range(stage.repeats - 1):
+ offset = len(result) - start
+ result.extend(
+ replace(step, number=step.number + offset, goto=step.goto + offset if step.goto else 0)
+ for step in block
+ )
+ source_steps.extend(source_block)
+ else:
+ result[-1] = replace(result[-1], goto=start + 1, loops=stage.repeats - 1)
+
+ for stage in protocol.stages:
+ append_stage(stage)
+ for index, (step, source, incoming) in enumerate(
+ zip(result, source_steps, _incoming_temperatures(result, first.temperature))
+ ):
+ maximum_rate = (
+ 2.2
+ if any(previous > step.temperature for previous in incoming)
+ else (4.4 if variant == 96 else 5.0)
+ )
+ rate = maximum_rate if source.ramp_rate is None else source.ramp_rate
+ if rate > maximum_rate:
+ raise ValueError("Explicit ramp rate exceeds a loop-entry transition limit")
+ overshoot = source.overshoot
+ if overshoot is None:
+ if apply_overshoot and len(incoming) > 1:
+ raise ValueError(
+ "Automatic overshoot is ambiguous at this loop entry; supply explicit overshoot"
+ )
+ overshoot = (
+ calculate_overshoot(step.temperature, next(iter(incoming)), rate, step.duration, quantity)
+ if apply_overshoot
+ else ODTCOvershoot(0, approach_rate=rate)
+ )
+ result[index] = replace(step, ramp_rate=rate, overshoot=overshoot)
+ method = ODTCMethod(
+ name=name or protocol.name or "plr_currentProtocol",
+ variant=variant,
+ plate_type=plate_type,
+ fluid_quantity=quantity,
+ post_heating=post_heating,
+ start_temperature=first.temperature,
+ start_lid_temperature=start_lid,
+ steps=result,
+ pid_set=list(pid_set) if pid_set is not None else [ODTCPID()],
+ )
+ validate_method(method)
+ return method
diff --git a/pylabrobot/inheco/odtc/tests/__init__.py b/pylabrobot/inheco/odtc/tests/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_data_event.xml b/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_data_event.xml
new file mode 100644
index 00000000000..389d53e1efa
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_data_event.xml
@@ -0,0 +1 @@
+<containerData version="1.0"><experimentStep name="Method" sequence="1" type="Measurement"><dataSeries nameId="Elapsed time" unit="ms" repeat="0"><seriesValueSet><integerValue mapId="">190</integerValue><integerValue mapId="">1286</integerValue><integerValue mapId="">2314</integerValue><integerValue mapId="">3335</integerValue><integerValue mapId="">4355</integerValue></seriesValueSet></dataSeries><dataSeries nameId="Target temperature" unit="1/100°C" repeat="0"><seriesValueSet><integerValue mapId="">2142</integerValue><integerValue mapId="">2000</integerValue><integerValue mapId="">1780</integerValue><integerValue mapId="">1580</integerValue><integerValue mapId="">1566</integerValue></seriesValueSet></dataSeries><dataSeries nameId="Current temperature" unit="1/100°C" repeat="0"><seriesValueSet><integerValue mapId="">2139</integerValue><integerValue mapId="">2064</integerValue><integerValue mapId="">1931</integerValue><integerValue mapId="">1791</integerValue><integerValue mapId="">1656</integerValue></seriesValueSet></dataSeries><dataSeries nameId="LID temperature" unit="1/100°C" repeat="0"><seriesValueSet><integerValue mapId="">2368</integerValue><integerValue mapId="">2374</integerValue><integerValue mapId="">2397</integerValue><integerValue mapId="">2423</integerValue><integerValue mapId="">2463</integerValue></seriesValueSet></dataSeries></experimentStep></containerData>
diff --git a/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_method_set.xml b/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_method_set.xml
new file mode 100644
index 00000000000..6df29ac4a0c
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/fixtures/firmware_manual_method_set.xml
@@ -0,0 +1,98 @@
+
+
+20
+110
+
+
+960000
+0
+0
+false
+20
+110
+
+1
+4.4
+95
+60
+4.4
+5
+0
+2.2
+0
+0
+1
+110
+
+
+2
+4.4
+95
+10
+4.4
+5
+0
+2.2
+0
+0
+1
+110
+
+
+3
+2.2
+65
+5
+2.2
+5
+0
+2.2
+0
+0
+1
+110
+
+
+4
+4.4
+72
+5
+4.4
+5
+0
+2.2
+2
+5
+1
+110
+
+
+5
+4.4
+95
+60
+4.4
+5
+0
+2.2
+0
+0
+1
+110
+
+
+
+60
+80
+250
+100
+10
+10
+100
+70
+
+
+
+
diff --git a/pylabrobot/inheco/odtc/tests/fixtures/flat_loop_xml.xml b/pylabrobot/inheco/odtc/tests/fixtures/flat_loop_xml.xml
new file mode 100644
index 00000000000..720c47ec52c
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/fixtures/flat_loop_xml.xml
@@ -0,0 +1,13 @@
+
+
+ false
+
+ 96000000
+ false
+ 25
+ 110
+ 14.495100.1000.1001110
+ 22.255100.1000.1121110
+ 6080250100101010070
+
+
diff --git a/pylabrobot/inheco/odtc/tests/fixtures/nested_loop_xml.xml b/pylabrobot/inheco/odtc/tests/fixtures/nested_loop_xml.xml
new file mode 100644
index 00000000000..6c949773d61
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/fixtures/nested_loop_xml.xml
@@ -0,0 +1,16 @@
+
+
+ false
+
+ 96000000
+ false
+ 25
+ 110
+ 14.495100.1000.1001110
+ 22.255100.1000.1001110
+ 34.472100.1000.1001110
+ 44.495100.1000.1241110
+ 52.250200.1000.11291110
+ 6080250100101010070
+
+
diff --git a/pylabrobot/inheco/odtc/tests/fixtures/overshoot_xml.xml b/pylabrobot/inheco/odtc/tests/fixtures/overshoot_xml.xml
new file mode 100644
index 00000000000..be9b80fb2cf
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/fixtures/overshoot_xml.xml
@@ -0,0 +1,12 @@
+
+
+ false
+
+ 96000001
+ true
+ 25
+ 110
+ 14.495304.45.202.2001110
+ 6080250100101010070
+
+
diff --git a/pylabrobot/inheco/odtc/tests/odtc_tests.py b/pylabrobot/inheco/odtc/tests/odtc_tests.py
new file mode 100644
index 00000000000..b6c98a3167b
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/odtc_tests.py
@@ -0,0 +1,365 @@
+"""Drive the ODTC through SOAP exchanges and asynchronous device callbacks."""
+
+import asyncio
+import unittest
+import xml.etree.ElementTree as ET
+from collections.abc import Awaitable, Callable
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+from pylabrobot.inheco.odtc import (
+ ODTC,
+ ODTCMethodSet,
+ ODTCProtocol,
+ ODTCProtocolStopped,
+ ODTCStage,
+ ODTCStep,
+ compile_protocol,
+ method_set_to_xml,
+ parse_method_set,
+)
+from pylabrobot.inheco.soap import soap_body_payload, soap_encode
+from pylabrobot.inheco.transport.sila import SiLAError, required_text
+from pylabrobot.io.http import HTTP, HTTPResponse
+from pylabrobot.io.http_server import HTTPServer
+
+FIXTURES = Path(__file__).with_name("fixtures")
+
+
+def profile(name=None):
+ return ODTCProtocol(
+ [ODTCStage([ODTCStep(95, 30), ODTCStep(55, 20), ODTCStep(72, 60)], 3)], name=name
+ )
+
+
+class MockODTC:
+ """A test-only device speaking the manual's SOAP and MethodSet formats."""
+
+ def __init__(self):
+ self.callback: Callable[[bytes], Awaitable[bytes]] | None = None
+ self.receiver_url = None
+ self.commands = []
+ self.executed = []
+ self.methods = {}
+ self.premethods = {}
+ self.parameters = {}
+ self.state = "Idle"
+ self.hold_name = None
+ self.running_id = None
+ self.started = asyncio.Event()
+ self.reject = {}
+ self.emit_progress = False
+
+ async def complete(self, request_id=None, code=3, data=""):
+ request_id = self.running_id if request_id is None else request_id
+ if request_id is None:
+ raise AssertionError("No simulated command to complete")
+ assert self.callback is not None
+ await self.callback(
+ soap_encode(
+ "ResponseEvent",
+ {
+ "requestId": request_id,
+ "returnValue": {"returnCode": code, "message": "simulated device response"},
+ "responseData": data,
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+ )
+ self.state = "Idle"
+
+ def parameter_data(self, name, value):
+ root = ET.Element("ResponseData")
+ parameters = ET.SubElement(root, "ParameterSet")
+ parameter = ET.SubElement(parameters, "Parameter", name=name)
+ ET.SubElement(parameter, "String").text = value
+ return ET.tostring(root, encoding="unicode")
+
+ async def handle(self, body):
+ request = soap_body_payload(body.decode())
+ name = request.tag.rsplit("}", 1)[-1]
+ request_id = int(required_text(request, "requestId"))
+ self.commands.append(name)
+ code = self.reject.get(name, 1)
+ extra = {}
+ if name in self.reject:
+ pass
+ elif name == "GetStatus":
+ extra = {"state": self.state}
+ elif name == "Reset":
+ self.receiver_url = required_text(request, "eventReceiverURI")
+ self.state = "Standby"
+ elif name == "Initialize":
+ self.state = "Idle"
+ elif name == "SetParameters":
+ parameters = ET.fromstring(required_text(request, "paramsXML"))
+ for parameter in parameters:
+ self.parameters[parameter.attrib["name"]] = parameter[0].text
+ methods = parse_method_set(self.parameters["MethodsXML"])
+ self.methods.update((m.name, m) for m in methods.methods)
+ self.premethods.update((m.name, m) for m in methods.premethods)
+ elif name == "GetParameters":
+ code = 2
+ xml = method_set_to_xml(
+ ODTCMethodSet(list(self.methods.values()), list(self.premethods.values()))
+ )
+ await self.complete(request_id, data=self.parameter_data("MethodsXML", xml))
+ elif name == "ReadActualTemperature":
+ code = 2
+ xml = (
+ "95009501"
+ "10500105012500"
+ "330026002400"
+ )
+ await self.complete(request_id, data=self.parameter_data("SensorValues", xml))
+ elif name in ("OpenDoor", "CloseDoor"):
+ code = 2
+ await self.complete(request_id)
+ elif name == "ExecuteMethod":
+ code = 2
+ method_name = required_text(request, "methodName")
+ self.executed.append(method_name)
+ self.running_id = request_id
+ self.state = "Busy"
+ if self.emit_progress:
+ assert self.callback is not None
+ await self.callback(
+ soap_encode(
+ "DataEvent",
+ {
+ "requestId": request_id,
+ "dataValue": (FIXTURES / "firmware_manual_data_event.xml").read_text(),
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+ )
+ if method_name == self.hold_name:
+ self.started.set()
+ else:
+ await self.complete(request_id)
+ elif name == "StopMethod":
+ self.state = "Idle"
+ else:
+ raise AssertionError(f"Unhandled test command {name}")
+ return soap_encode(
+ f"{name}Response",
+ {
+ f"{name}Result": {
+ "returnCode": code,
+ "message": "simulated device response",
+ "duration": "PT10S",
+ },
+ **extra,
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+
+ async def post(self, method, path, body, **kwargs):
+ return HTTPResponse(200, await self.handle(body), {})
+
+
+class ODTCTests(unittest.IsolatedAsyncioTestCase):
+ async def asyncSetUp(self):
+ self.server_patch = patch("pylabrobot.inheco.transport.sila.HTTPServer")
+ server = self.server_patch.start().return_value
+ server.setup = AsyncMock()
+ server.stop = AsyncMock()
+ server.bound_port = 8000
+ self.odtc = ODTC("odtc", "127.0.0.1", client_ip="127.0.0.1")
+ self.fake = MockODTC()
+ self.fake.callback = self.odtc._sila._handle_event
+ self.post_patch = patch.object(self.odtc._sila.io, "request_raw", side_effect=self.fake.post)
+ self.post_patch.start()
+ await self.odtc.setup()
+
+ async def asyncTearDown(self):
+ self.fake.reject.clear()
+ await self.odtc.stop()
+ self.post_patch.stop()
+ self.server_patch.stop()
+
+ async def test_setup_lifecycle_and_repeated_setup(self):
+ self.assertEqual(
+ self.fake.commands, ["GetStatus", "Reset", "GetStatus", "Initialize", "GetStatus"]
+ )
+ await self.odtc.setup()
+ self.assertEqual(len(self.fake.commands), 5)
+ self.assertIsNone(self.odtc.door_open)
+
+ async def test_run_prepares_matching_lid_and_retains_progress(self):
+ self.fake.emit_progress = True
+ await self.odtc.run_protocol(profile(), volume=25, lid_temperature=105)
+ self.assertEqual(self.fake.executed, ["plr_currentPreMethod", "plr_currentProtocol"])
+ pre = self.fake.premethods["plr_currentPreMethod"]
+ self.assertEqual((pre.temperature, pre.lid_temperature), (95, 105))
+ self.assertEqual(self.fake.parameters["MatchLidTemperatures"], "true")
+ progress = await self.odtc.request_progress()
+ assert progress is not None
+ self.assertEqual(progress.elapsed, 4.355)
+ self.assertEqual(self.odtc._prepared, (72, 105))
+
+ async def test_read_while_running_then_complete(self):
+ self.fake.hold_name = "plr_currentProtocol"
+ task = asyncio.create_task(self.odtc.run_protocol(profile(), volume=25, lid_temperature=105))
+ await asyncio.wait_for(self.fake.started.wait(), 1)
+ self.assertEqual((await self.odtc.request_temperatures()).mount, 95)
+ self.assertFalse(task.done())
+ await self.fake.complete()
+ await task
+
+ async def test_stop_during_premethod_never_starts_main_method(self):
+ self.fake.hold_name = "plr_currentPreMethod"
+ task = asyncio.create_task(self.odtc.run_protocol(profile(), volume=25, lid_temperature=105))
+ await asyncio.wait_for(self.fake.started.wait(), 1)
+ await self.odtc.stop_protocol()
+ with self.assertRaises(ODTCProtocolStopped):
+ await task
+ self.assertEqual(self.fake.executed, ["plr_currentPreMethod"])
+ self.assertIsNone(self.odtc._prepared)
+
+ async def test_late_waiter_sees_completed_device_error(self):
+ self.fake.hold_name = "plr_currentProtocol"
+ task = asyncio.create_task(self.odtc.run_protocol(profile(), volume=25, lid_temperature=105))
+ await asyncio.wait_for(self.fake.started.wait(), 1)
+ await self.fake.complete(code=2002)
+ with self.assertRaises(SiLAError):
+ await task
+ with self.assertRaises(SiLAError):
+ await self.odtc.wait_for_completion()
+
+ async def test_cancel_wait_leaves_device_command_running(self):
+ self.fake.hold_name = "plr_currentProtocol"
+ task = asyncio.create_task(self.odtc.run_protocol(profile(), volume=25, lid_temperature=105))
+ await asyncio.wait_for(self.fake.started.wait(), 1)
+ task.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await task
+ assert self.odtc._execution is not None
+ self.assertFalse(self.odtc._execution.future.done())
+ await self.fake.complete()
+ await self.odtc.wait_for_completion()
+ self.assertIsNone(self.odtc._prepared)
+
+ async def test_timeout_does_not_report_completion(self):
+ self.fake.hold_name = "plr_currentProtocol"
+ with self.assertRaises(asyncio.TimeoutError):
+ await self.odtc.run_protocol(profile(), volume=25, lid_temperature=105, timeout=0.001)
+ assert self.odtc._execution is not None
+ self.assertFalse(self.odtc._execution.future.done())
+ await self.fake.complete()
+ await self.odtc.wait_for_completion()
+
+ async def test_stored_protocol_preserves_settings_and_requires_overwrite(self):
+ method = compile_protocol(profile("stored"), variant=96, volume=75, lid_temperature=100)
+ await self.odtc.upload_protocol(method)
+ self.assertEqual(await self.odtc.request_protocol("stored"), method)
+ with self.assertRaisesRegex(ValueError, "overwrite"):
+ await self.odtc.upload_protocol(method)
+ await self.odtc.run_stored_protocol("stored")
+ self.assertEqual(self.fake.premethods["plr_currentPreMethod"].lid_temperature, 100)
+ self.assertEqual(self.fake.methods["stored"].fluid_quantity, 2)
+
+ async def test_repeated_door_commands_and_failed_move(self):
+ await self.odtc.open_door()
+ await self.odtc.open_door()
+ self.assertEqual(self.fake.commands.count("OpenDoor"), 1)
+ self.fake.reject["CloseDoor"] = 2002
+ with self.assertRaises(SiLAError):
+ await self.odtc.close_door()
+ self.assertIsNone(self.odtc.door_open)
+
+ async def test_reconnect_invalidates_tracked_state(self):
+ await self.odtc.open_door()
+ await self.odtc.stop()
+ await self.odtc.setup()
+ self.assertIsNone(self.odtc.door_open)
+ self.assertIsNone(self.odtc._prepared)
+
+ async def test_setup_does_not_reset_busy_instrument(self):
+ await self.odtc.stop()
+ self.fake.state = "Busy"
+ count = self.fake.commands.count("Reset")
+ with self.assertRaisesRegex(RuntimeError, "will not reset"):
+ await self.odtc.setup()
+ self.assertEqual(self.fake.commands.count("Reset"), count)
+ self.assertIsNone(self.odtc._sila._server)
+
+ async def test_stop_closes_transport_even_if_device_refuses(self):
+ self.fake.reject["StopMethod"] = 2000
+ with self.assertRaises(SiLAError):
+ await self.odtc.stop()
+ self.assertIsNone(self.odtc._sila._server)
+ self.assertFalse(self.odtc._connected)
+
+ async def test_device_status_fault_reaches_pending_execution(self):
+ self.fake.hold_name = "plr_currentProtocol"
+ task = asyncio.create_task(self.odtc.run_protocol(profile(), volume=25, lid_temperature=105))
+ await asyncio.wait_for(self.fake.started.wait(), 1)
+ await self.odtc._sila._handle_event(
+ soap_encode(
+ "StatusEvent",
+ {
+ "eventDescription": "InError",
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+ )
+ with self.assertRaises(SiLAError):
+ await task
+ self.assertIsNone(self.odtc.door_open)
+
+ async def test_invalid_timeout_is_rejected_before_upload_or_execution(self):
+ commands = list(self.fake.commands)
+ for timeout in (0, -1, float("nan"), float("inf")):
+ with self.subTest(timeout=timeout), self.assertRaises(ValueError):
+ await self.odtc.run_protocol(profile(), volume=25, lid_temperature=105, timeout=timeout)
+ self.assertEqual(self.fake.commands, commands)
+
+ async def test_failed_initialization_cleans_up_and_can_retry(self):
+ await self.odtc.stop()
+ self.fake.reject["Initialize"] = 2000
+ with self.assertRaises(SiLAError):
+ await self.odtc.setup()
+ self.assertIsNone(self.odtc._sila._server)
+ self.fake.reject.clear()
+ await self.odtc.setup()
+ self.assertTrue(self.odtc._connected)
+
+
+class ODTCLoopbackTests(unittest.IsolatedAsyncioTestCase):
+ async def test_full_stack_with_real_http_and_callback_sockets(self):
+ fake = MockODTC()
+ server = HTTPServer(fake.handle, "127.0.0.1")
+ await server.setup()
+ odtc = ODTC(
+ "loopback", "127.0.0.1", variant=384, client_ip="127.0.0.1", odtc_port=server.bound_port
+ )
+ callback_client = None
+
+ async def callback(body):
+ nonlocal callback_client
+ if callback_client is None:
+ assert fake.receiver_url is not None
+ callback_client = HTTP("simulated ODTC callback", fake.receiver_url)
+ await callback_client.setup()
+ result = await callback_client.request_raw("POST", "/", body)
+ self.assertEqual(result.status, 200)
+ return result.body
+
+ fake.callback = callback
+ fake.emit_progress = True
+ try:
+ await odtc.setup()
+ await odtc.open_door()
+ await odtc.close_door()
+ await odtc.run_protocol(profile(), volume=25, lid_temperature=115, plate_type=2)
+ self.assertEqual((await odtc.request_temperatures()).lid, 105)
+ self.assertEqual(fake.methods["plr_currentProtocol"].variant, 384)
+ progress = await odtc.request_progress()
+ assert progress is not None
+ self.assertEqual(progress.current_temperature, 16.56)
+ finally:
+ await odtc.stop()
+ if callback_client is not None:
+ await callback_client.stop()
+ await server.stop()
diff --git a/pylabrobot/inheco/odtc/tests/protocol_tests.py b/pylabrobot/inheco/odtc/tests/protocol_tests.py
new file mode 100644
index 00000000000..80ed21f0e2a
--- /dev/null
+++ b/pylabrobot/inheco/odtc/tests/protocol_tests.py
@@ -0,0 +1,217 @@
+"""Native XML fixtures and compiler regressions.
+
+firmware_manual fixtures use examples from TD_SILA_FWCommandSet v1.06, sections 6.1
+and 9.1.5: https://github.com/user-attachments/files/26010799/TD_SILA_FWCommandSet.25210.pdf
+The other fixtures exercise flat/nested loops and independent overshoot ramp values.
+"""
+
+import unittest
+import xml.etree.ElementTree as ET
+from dataclasses import replace
+from pathlib import Path
+from typing import Any
+
+from pylabrobot.inheco.odtc import (
+ ODTCPID,
+ ODTCMethodSet,
+ ODTCOvershoot,
+ ODTCProtocol,
+ ODTCStage,
+ ODTCStep,
+ compile_protocol,
+ method_set_to_xml,
+ parse_method_set,
+)
+from pylabrobot.inheco.odtc.protocol import (
+ ODTCVariant,
+ calculate_overshoot,
+ volume_to_fluid_quantity,
+)
+from pylabrobot.inheco.odtc.xml import parse_progress, parse_sensor_values
+
+FIXTURES = Path(__file__).with_name("fixtures")
+
+
+class ProtocolTests(unittest.TestCase):
+ def test_native_roundtrip_retains_every_field(self):
+ for name in ("flat_loop_xml", "nested_loop_xml", "overshoot_xml", "firmware_manual_method_set"):
+ with self.subTest(name=name):
+ original = parse_method_set((FIXTURES / f"{name}.xml").read_text())
+ self.assertEqual(parse_method_set(method_set_to_xml(original)), original)
+
+ def test_manual_method_and_loop(self):
+ methods = parse_method_set((FIXTURES / "firmware_manual_method_set.xml").read_text())
+ self.assertEqual(len(methods.methods[0].steps), 5)
+ self.assertEqual(methods.methods[0].steps[3].goto, 2)
+ self.assertEqual(methods.methods[0].steps[3].loops, 5)
+ self.assertEqual(methods.premethods[0].temperature, 20)
+
+ def test_compile_cycles_and_variant(self):
+ variants: tuple[tuple[ODTCVariant, float], ...] = ((96, 4.4), (384, 5.0))
+ for variant, rate in variants:
+ with self.subTest(variant=variant):
+ profile = ODTCProtocol(
+ [ODTCStage([ODTCStep(95, 30), ODTCStep(55, 20), ODTCStep(72, 60)], 30)]
+ )
+ method = compile_protocol(profile, variant=variant, volume=29.5, lid_temperature=105)
+ self.assertEqual(method.start_temperature, 95)
+ self.assertEqual(method.start_lid_temperature, 105)
+ self.assertEqual(method.fluid_quantity, 0)
+ self.assertEqual(method.steps[2].ramp_rate, rate)
+ self.assertEqual(method.steps[2].goto, 1)
+ self.assertEqual(method.steps[2].loops, 29)
+ self.assertEqual(method.steps[1].ramp_rate, 2.2)
+ self.assertEqual(
+ parse_method_set(method_set_to_xml(ODTCMethodSet([method]))).methods[0], method
+ )
+
+ def test_nested_loop_order(self):
+ profile = ODTCProtocol(
+ [
+ ODTCStage(
+ [ODTCStep(95, 10), ODTCStage([ODTCStep(55, 10), ODTCStep(72, 10)], 5), ODTCStep(50, 20)],
+ 30,
+ )
+ ]
+ )
+ method = compile_protocol(profile, variant=96, volume=50, lid_temperature=110)
+ self.assertEqual(
+ [(s.temperature, s.goto, s.loops) for s in method.steps],
+ [(95, 0, 0), (55, 0, 0), (72, 2, 4), (50, 1, 29)],
+ )
+
+ def test_coincident_loop_endpoints_preserve_repetitions(self):
+ profile = ODTCProtocol([ODTCStage([ODTCStage([ODTCStep(95, 1), ODTCStep(55, 1)], 3)], 2)])
+ method = compile_protocol(profile, variant=96, volume=50, lid_temperature=110)
+ self.assertEqual(
+ [(s.number, s.goto, s.loops) for s in method.steps],
+ [(1, 0, 0), (2, 1, 2), (3, 0, 0), (4, 3, 2)],
+ )
+
+ def test_explicit_overshoot_lid_and_pid_survive_compilation(self):
+ overshoot = ODTCOvershoot(3.5, duration=2, approach_rate=1, return_rate=0.5)
+ profile = ODTCProtocol(
+ [ODTCStage([ODTCStep(95, 30, lid_temperature=100, overshoot=overshoot, pid_number=2)])]
+ )
+ method = compile_protocol(
+ profile, variant=96, volume=25, lid_temperature=105, pid_set=[ODTCPID(number=2)]
+ )
+ self.assertEqual(method.steps[0].overshoot, overshoot)
+ self.assertEqual(method.start_lid_temperature, 100)
+ self.assertEqual(
+ parse_method_set(method_set_to_xml(ODTCMethodSet([method]))).methods[0], method
+ )
+
+ def test_invalid_profiles_fail_without_io(self):
+ for step in (
+ ODTCStep(float("nan"), 5),
+ ODTCStep(100, 5),
+ ODTCStep(95, -1),
+ ODTCStep(95, float("inf")),
+ ODTCStep(95, 5, ramp_rate=100),
+ ODTCStep(95, 5, pid_number=9),
+ ):
+ with self.subTest(step=step), self.assertRaises(ValueError):
+ compile_protocol(
+ ODTCProtocol([ODTCStage([step])]), variant=96, volume=25, lid_temperature=110
+ )
+ invalid_repeats: list[Any] = [0, -1, 1.5, True]
+ for repeats in invalid_repeats:
+ with self.subTest(repeats=repeats), self.assertRaises(ValueError):
+ compile_protocol(
+ ODTCProtocol([ODTCStage([ODTCStep(95, 5)], repeats)]),
+ variant=96,
+ volume=25,
+ lid_temperature=110,
+ )
+
+ def test_volume_boundaries(self):
+ for volume, code in ((1, 0), (29.9, 0), (30, 1), (74.9, 1), (75, 2), (100, 2)):
+ self.assertEqual(volume_to_fluid_quantity(volume), code)
+ for volume in (0, -1, 101, float("nan"), float("inf")):
+ with self.assertRaises(ValueError):
+ volume_to_fluid_quantity(volume)
+
+ def test_automatic_overshoot_is_opt_in(self):
+ profile = ODTCProtocol([ODTCStage([ODTCStep(50, 10), ODTCStep(95, 30)])])
+ method = compile_protocol(profile, variant=96, volume=25, lid_temperature=110)
+ self.assertEqual(method.steps[1].overshoot.temperature, 0)
+ with self.assertLogs("pylabrobot.inheco.odtc.protocol", level="WARNING"):
+ method = compile_protocol(
+ profile, variant=96, volume=25, lid_temperature=110, apply_overshoot=True
+ )
+ self.assertGreater(method.steps[1].overshoot.temperature, 0)
+ self.assertLessEqual(99 + calculate_overshoot(99, 40, 4.4, 30, 2).temperature, 102)
+
+ def test_no_implicit_delete_and_unique_names(self):
+ original = parse_method_set((FIXTURES / "flat_loop_xml.xml").read_text())
+ self.assertNotIn("DeleteAllMethods", method_set_to_xml(original))
+ with self.assertRaises(ValueError):
+ method_set_to_xml(ODTCMethodSet([original.methods[0], original.methods[0]]))
+
+ def test_missing_xml_fields_are_errors(self):
+ xml = (FIXTURES / "flat_loop_xml.xml").read_text().replace("10", "")
+ with self.assertRaisesRegex(ValueError, "PlateauTime"):
+ parse_method_set(xml)
+
+ def test_invalid_native_loop_is_rejected(self):
+ method = parse_method_set((FIXTURES / "flat_loop_xml.xml").read_text()).methods[0]
+ method = replace(method, steps=[replace(method.steps[0], goto=5, loops=1), method.steps[1]])
+ with self.assertRaisesRegex(ValueError, "loop"):
+ method_set_to_xml(ODTCMethodSet([method]))
+
+ def test_manual_progress_units(self):
+ sample = parse_progress((FIXTURES / "firmware_manual_data_event.xml").read_text())
+ self.assertEqual(
+ (
+ sample.elapsed,
+ sample.target_temperature,
+ sample.current_temperature,
+ sample.lid_temperature,
+ ),
+ (4.355, 15.66, 16.56, 24.63),
+ )
+
+ def test_malformed_progress_is_not_reported_as_zero(self):
+ xml = (FIXTURES / "firmware_manual_data_event.xml").read_text()
+ with self.assertRaisesRegex(ValueError, "units"):
+ parse_progress(xml.replace("1/100°C", "C"))
+ root = ET.fromstring(xml)
+ any_data = root.find("AnyData")
+ assert any_data is not None and any_data.text is not None
+ data = ET.fromstring(any_data.text)
+ step = data.find("experimentStep")
+ assert step is not None
+ step.remove(step[-1])
+ any_data.text = ET.tostring(data, encoding="unicode")
+ with self.assertRaisesRegex(ValueError, "missing series"):
+ parse_progress(ET.tostring(root, encoding="unicode"))
+
+ def test_sensor_values_from_manual(self):
+ xml = (
+ '2463'
+ "264225752627"
+ "2450330825962487"
+ )
+ values = parse_sensor_values(xml)
+ self.assertEqual(values.mount, 24.63)
+ self.assertEqual(values.lid, 25.75)
+ with self.assertRaises(ValueError):
+ parse_sensor_values(xml.replace("2463", ""))
+
+ def test_loop_entry_rate_accounts_for_return_transition(self):
+ profile = ODTCProtocol([ODTCStage([ODTCStep(50, 5), ODTCStep(95, 5)], repeats=2)])
+ method = compile_protocol(profile, variant=96, volume=25, lid_temperature=105)
+ self.assertEqual(method.steps[0].ramp_rate, 2.2)
+ with self.assertRaisesRegex(ValueError, "ambiguous"):
+ compile_protocol(profile, variant=96, volume=25, lid_temperature=105, apply_overshoot=True)
+
+ def test_working_premethod_name_cannot_overwrite_main_method(self):
+ with self.assertRaisesRegex(ValueError, "reserved"):
+ compile_protocol(
+ ODTCProtocol([ODTCStage([ODTCStep(95, 10)])]),
+ variant=96,
+ volume=25,
+ lid_temperature=105,
+ name="plr_currentPreMethod",
+ )
diff --git a/pylabrobot/inheco/odtc/xml.py b/pylabrobot/inheco/odtc/xml.py
new file mode 100644
index 00000000000..9a97087117b
--- /dev/null
+++ b/pylabrobot/inheco/odtc/xml.py
@@ -0,0 +1,258 @@
+"""Explicit ODTC MethodSet, SensorValues, and DataEvent XML codecs."""
+
+import math
+import xml.etree.ElementTree as ET
+
+from pylabrobot.inheco.odtc.protocol import (
+ ODTCPID,
+ ODTCMethod,
+ ODTCMethodSet,
+ ODTCMethodStep,
+ ODTCOvershoot,
+ ODTCPreMethod,
+ ODTCProgress,
+ ODTCTemperatures,
+ validate_method,
+)
+from pylabrobot.inheco.soap import required_text
+
+
+def _number(root: ET.Element, name: str) -> float:
+ """Read a required finite numeric element."""
+ number = float(required_text(root, name))
+ if not math.isfinite(number):
+ raise ValueError(f"{name} must be finite")
+ return number
+
+
+def _boolean(root: ET.Element, name: str) -> bool:
+ """Read the XML boolean forms used by the device."""
+ value = required_text(root, name)
+ if value not in ("true", "false", "1", "0"):
+ raise ValueError(f"Invalid {name}: {value!r}")
+ return value in ("true", "1")
+
+
+def _fields(root: ET.Element, values: dict[str, str | float | int | bool]) -> None:
+ """Write an explicit ordered mapping of device fields."""
+ for name, value in values.items():
+ text = str(value).lower() if isinstance(value, bool) else str(value)
+ if isinstance(value, float):
+ text = format(value, ".15g")
+ ET.SubElement(root, name).text = text
+
+
+def _metadata(root: ET.Element, method: ODTCMethod | ODTCPreMethod) -> None:
+ """Write only metadata that was supplied by the user or instrument."""
+ root.set("methodName", method.name)
+ if method.creator is not None:
+ root.set("creator", method.creator)
+ if method.description is not None:
+ root.set("description", method.description)
+ if method.date_time is not None:
+ root.set("dateTime", method.date_time)
+
+
+def _write_pid(parent: ET.Element, pid: ODTCPID) -> None:
+ """Encode one native PID set without reflection."""
+ element = ET.SubElement(parent, "PID", number=str(pid.number))
+ _fields(
+ element,
+ {
+ "PHeating": pid.p_heating,
+ "PCooling": pid.p_cooling,
+ "IHeating": pid.i_heating,
+ "ICooling": pid.i_cooling,
+ "DHeating": pid.d_heating,
+ "DCooling": pid.d_cooling,
+ "PLid": pid.p_lid,
+ "ILid": pid.i_lid,
+ },
+ )
+
+
+def method_set_to_xml(method_set: ODTCMethodSet) -> str:
+ """Serialize methods for an additive SetParameters upload; never delete stored methods."""
+ root = ET.Element("MethodSet")
+ names = [method.name for method in method_set.methods] + [p.name for p in method_set.premethods]
+ if len(set(names)) != len(names):
+ raise ValueError("Method and premethod names must be unique")
+ for premethod in method_set.premethods:
+ if not premethod.name.strip():
+ raise ValueError("Premethod name must not be empty")
+ element = ET.SubElement(root, "PreMethod")
+ _metadata(element, premethod)
+ _fields(
+ element,
+ {"TargetBlockTemperature": premethod.temperature, "TargetLidTemp": premethod.lid_temperature},
+ )
+ for method in method_set.methods:
+ validate_method(method)
+ element = ET.SubElement(root, "Method")
+ _metadata(element, method)
+ _fields(
+ element,
+ {
+ "Variant": {96: 960000, 384: 384000}[method.variant],
+ "PlateType": method.plate_type,
+ "FluidQuantity": method.fluid_quantity,
+ "PostHeating": method.post_heating,
+ "StartBlockTemperature": method.start_temperature,
+ "StartLidTemperature": method.start_lid_temperature,
+ },
+ )
+ for step in method.steps:
+ overshoot = step.overshoot
+ _fields(
+ ET.SubElement(element, "Step"),
+ {
+ "Number": step.number,
+ "Slope": step.ramp_rate,
+ "PlateauTemperature": step.temperature,
+ "PlateauTime": step.duration,
+ "OverShootSlope1": step.ramp_rate
+ if overshoot.approach_rate is None
+ else overshoot.approach_rate,
+ "OverShootTemperature": overshoot.temperature,
+ "OverShootTime": overshoot.duration,
+ "OverShootSlope2": overshoot.return_rate,
+ "GotoNumber": step.goto,
+ "LoopNumber": step.loops,
+ "PIDNumber": step.pid_number,
+ "LidTemp": step.lid_temperature,
+ },
+ )
+ pids = ET.SubElement(element, "PIDSet")
+ for pid in method.pid_set:
+ _write_pid(pids, pid)
+ return ET.tostring(root, encoding="unicode")
+
+
+def _read_pid(element: ET.Element) -> ODTCPID:
+ """Decode every PID coefficient explicitly."""
+ return ODTCPID(
+ number=int(element.attrib["number"]),
+ p_heating=_number(element, "PHeating"),
+ p_cooling=_number(element, "PCooling"),
+ i_heating=_number(element, "IHeating"),
+ i_cooling=_number(element, "ICooling"),
+ d_heating=_number(element, "DHeating"),
+ d_cooling=_number(element, "DCooling"),
+ p_lid=_number(element, "PLid"),
+ i_lid=_number(element, "ILid"),
+ )
+
+
+def _read_step(element: ET.Element) -> ODTCMethodStep:
+ """Preserve all native ramp, PID, and loop values on a downloaded step."""
+ return ODTCMethodStep(
+ number=int(required_text(element, "Number")),
+ temperature=_number(element, "PlateauTemperature"),
+ duration=_number(element, "PlateauTime"),
+ ramp_rate=_number(element, "Slope"),
+ lid_temperature=_number(element, "LidTemp"),
+ overshoot=ODTCOvershoot(
+ temperature=_number(element, "OverShootTemperature"),
+ duration=_number(element, "OverShootTime"),
+ return_rate=_number(element, "OverShootSlope2"),
+ approach_rate=_number(element, "OverShootSlope1"),
+ ),
+ goto=int(required_text(element, "GotoNumber")),
+ loops=int(required_text(element, "LoopNumber")),
+ pid_number=int(required_text(element, "PIDNumber")),
+ )
+
+
+def parse_method_set(xml: str) -> ODTCMethodSet:
+ """Read native methods and preparation methods without reconstructing guessed stages."""
+ root = ET.fromstring(xml)
+ if root.tag.rsplit("}", 1)[-1] != "MethodSet":
+ raise ValueError("Expected MethodSet XML")
+ methods = []
+ for element in root.findall("{*}Method"):
+ code = int(required_text(element, "Variant"))
+ if code not in (960000, 384000):
+ raise ValueError(f"Unsupported ODTC variant code {code}")
+ method = ODTCMethod(
+ name=element.attrib["methodName"],
+ variant=96 if code == 960000 else 384,
+ plate_type=int(required_text(element, "PlateType")),
+ fluid_quantity=int(required_text(element, "FluidQuantity")),
+ post_heating=_boolean(element, "PostHeating"),
+ start_temperature=_number(element, "StartBlockTemperature"),
+ start_lid_temperature=_number(element, "StartLidTemperature"),
+ steps=[_read_step(step) for step in element.findall("{*}Step")],
+ pid_set=[_read_pid(pid) for pid in element.findall("{*}PIDSet/{*}PID")],
+ creator=element.get("creator"),
+ description=element.get("description"),
+ date_time=element.get("dateTime"),
+ )
+ validate_method(method)
+ methods.append(method)
+ premethods = [
+ ODTCPreMethod(
+ name=element.attrib["methodName"],
+ temperature=_number(element, "TargetBlockTemperature"),
+ lid_temperature=_number(element, "TargetLidTemp"),
+ creator=element.get("creator"),
+ description=element.get("description"),
+ date_time=element.get("dateTime"),
+ )
+ for element in root.findall("{*}PreMethod")
+ ]
+ return ODTCMethodSet(methods, premethods)
+
+
+def parse_sensor_values(xml: str) -> ODTCTemperatures:
+ """Convert all eight integer sensor readings from hundredths of a degree."""
+ root = ET.fromstring(xml)
+ if root.tag.rsplit("}", 1)[-1] != "SensorValues":
+ raise ValueError("Expected SensorValues XML")
+ return ODTCTemperatures(
+ mount=_number(root, "Mount") / 100,
+ mount_monitor=_number(root, "Mount_Monitor") / 100,
+ lid=_number(root, "Lid") / 100,
+ lid_monitor=_number(root, "Lid_Monitor") / 100,
+ ambient=_number(root, "Ambient") / 100,
+ pcb=_number(root, "PCB") / 100,
+ heatsink=_number(root, "Heatsink") / 100,
+ heatsink_tec=_number(root, "Heatsink_TEC") / 100,
+ timestamp=root.get("timestamp"),
+ )
+
+
+def parse_progress(xml: str) -> ODTCProgress:
+ """Read the four documented DataCapture series, rejecting missing or mismatched samples."""
+ outer = ET.fromstring(xml)
+ data = required_text(outer, "AnyData")
+ root = ET.fromstring(data)
+ expected = {
+ "Elapsed time": "ms",
+ "Target temperature": "1/100°C",
+ "Current temperature": "1/100°C",
+ "LID temperature": "1/100°C",
+ }
+ samples = {}
+ sample_count = None
+ for series in root.findall(".//{*}dataSeries"):
+ name = series.attrib["nameId"]
+ if name not in expected:
+ continue
+ if name in samples or series.attrib["unit"] != expected[name]:
+ raise ValueError(f"Duplicate series or unexpected units for {name}")
+ values = series.findall("{*}seriesValueSet/{*}integerValue")
+ if not values or (sample_count is not None and len(values) != sample_count):
+ raise ValueError("DataEvent series must contain matching nonempty samples")
+ sample_count = len(values)
+ text = values[-1].text
+ if text is None:
+ raise ValueError(f"Missing value for {name}")
+ samples[name] = int(text)
+ if samples.keys() != expected.keys():
+ raise ValueError(f"DataEvent is missing series: {expected.keys() - samples.keys()}")
+ return ODTCProgress(
+ samples["Elapsed time"] / 1000,
+ samples["Target temperature"] / 100,
+ samples["Current temperature"] / 100,
+ samples["LID temperature"] / 100,
+ )
diff --git a/pylabrobot/inheco/scila/soap.py b/pylabrobot/inheco/scila/soap.py
index 4517fade635..cd7d4253e1a 100644
--- a/pylabrobot/inheco/scila/soap.py
+++ b/pylabrobot/inheco/scila/soap.py
@@ -1,266 +1,5 @@
-from __future__ import annotations
+"""Compatibility imports for the Inheco SOAP codec."""
-import datetime as _dt
-import re as _re
-import xml.etree.ElementTree as ET
-from typing import Any, Dict, List, Mapping, Optional, Union
+from pylabrobot.inheco.soap import XSI, soap_body_payload, soap_decode, soap_encode
-SOAP_ENV = "http://schemas.xmlsoap.org/soap/envelope/"
-XSI = "http://www.w3.org/2001/XMLSchema-instance"
-
-ET.register_namespace("s", SOAP_ENV)
-ET.register_namespace("xsi", XSI)
-
-
-# --------- scalar parsing/formatting ---------
-
-_INT_RE = _re.compile(r"^-?\d+$")
-_FLOAT_RE = _re.compile(r"^-?\d+\.\d+$")
-# ISO-8601 duration subset: PnDTnHnMnS (supports fractional seconds)
-_DUR_RE = _re.compile(
- r"^P"
- r"(?:(?P\d+)D)?"
- r"(?:T"
- r"(?:(?P\d+)H)?"
- r"(?:(?P\d+)M)?"
- r"(?:(?P\d+(?:\.\d+)?)S)?"
- r")?$"
-)
-
-# Basic ISO datetime with optional fractional seconds and optional Z
-_DT_RE = _re.compile(
- r"^(?P\d{4})-(?P\d{2})-(?P\d{2})"
- r"T(?P\d{2}):(?P\d{2}):(?P\d{2})"
- r"(?:\.(?P\d+))?"
- r"(?PZ)?$"
-)
-
-
-def _parse_duration_iso8601(s: str) -> Optional[_dt.timedelta]:
- m = _DUR_RE.match(s)
- if not m:
- return None
- days = int(m.group("days") or 0)
- hours = int(m.group("hours") or 0)
- minutes = int(m.group("minutes") or 0)
- sec_str = m.group("seconds") or "0"
- seconds = float(sec_str) if sec_str else 0.0
- return _dt.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
-
-
-def _format_duration_iso8601(td: _dt.timedelta) -> str:
- total_seconds = td.total_seconds()
- if total_seconds < 0:
- raise ValueError("negative timedeltas not supported for ISO-8601 formatting here")
- days = int(total_seconds // 86400)
- rem = total_seconds - days * 86400
- hours = int(rem // 3600)
- rem -= hours * 3600
- minutes = int(rem // 60)
- seconds = rem - minutes * 60
-
- out = "P"
- if days:
- out += f"{days}D"
- if hours or minutes or seconds or not days:
- out += "T"
- if hours:
- out += f"{hours}H"
- if minutes:
- out += f"{minutes}M"
- # keep seconds even if 0 when nothing else was emitted in T
- if seconds or (not hours and not minutes):
- # avoid trailing .0 when it’s integral
- if abs(seconds - round(seconds)) < 1e-12:
- out += f"{int(round(seconds))}S"
- else:
- out += f"{seconds}S"
- return out
-
-
-Scalar = Union[str, int, float, bool, _dt.timedelta, _dt.datetime]
-
-
-def _parse_scalar(text: Optional[str]) -> Scalar:
- if text is None:
- return ""
- s = text.strip()
- if s == "":
- return ""
-
- if s in ("true", "false"):
- return s == "true"
-
- dur = _parse_duration_iso8601(s)
- if dur is not None and s.startswith("P"):
- return dur
-
- dm = _DT_RE.match(s)
- if dm:
- y, mo, d = int(dm["y"]), int(dm["m"]), int(dm["d"])
- h, mi, se = int(dm["h"]), int(dm["mi"]), int(dm["s"])
- frac = dm["frac"]
- micro = int((frac or "0").ljust(6, "0")[:6])
- dt = _dt.datetime(y, mo, d, h, mi, se, microsecond=micro)
- if dm["z"] == "Z":
- dt = dt.replace(tzinfo=_dt.timezone.utc)
- return dt
-
- if _INT_RE.match(s):
- try:
- return int(s)
- except ValueError:
- pass
-
- if _FLOAT_RE.match(s):
- try:
- return float(s)
- except ValueError:
- pass
-
- return s
-
-
-def _format_scalar(value: Any) -> str:
- if isinstance(value, bool):
- return "true" if value else "false"
- if isinstance(value, int):
- return str(value)
- if isinstance(value, float):
- return repr(value)
- if isinstance(value, _dt.timedelta):
- return _format_duration_iso8601(value)
- if isinstance(value, _dt.datetime):
- # if timezone-aware UTC, emit Z
- if value.tzinfo is not None and value.utcoffset() == _dt.timedelta(0):
- v = value.astimezone(_dt.timezone.utc).replace(tzinfo=None)
- return v.isoformat(timespec="seconds") + "Z"
- return value.isoformat()
- return str(value)
-
-
-def _localname(tag: str) -> str:
- # "{uri}Name" -> "Name"
- return tag.split("}", 1)[1] if tag.startswith("{") else tag
-
-
-def _is_xsi_nil(el: ET.Element) -> bool:
- nil_attr = el.attrib.get(f"{{{XSI}}}nil")
- return (nil_attr or "").lower() == "true"
-
-
-# --------- generic XML -> Python structure ---------
-
-
-Object = Union[Dict[str, Union["Object", List["Object"]]], Scalar]
-
-
-def _xml_to_obj(el: ET.Element) -> Object:
- if _is_xsi_nil(el):
- return {}
-
- children = list(el)
- if not children:
- return _parse_scalar(el.text)
-
- # group by localname
- grouped: Dict[str, list[ET.Element]] = {}
- for ch in children:
- grouped.setdefault(_localname(ch.tag), []).append(ch)
-
- out: Object = {}
- for name, items in grouped.items():
- if len(items) == 1:
- out[name] = _xml_to_obj(items[0]) # type: ignore
- else:
- out[name] = [_xml_to_obj(i) for i in items] # type: ignore
- return out
-
-
-# --------- SOAP decode helpers ---------
-
-
-def soap_body_payload(xml: str) -> ET.Element:
- """
- Returns the first element inside SOAP Body (e.g., GetStatusResponse).
- """
- root = ET.fromstring(xml)
- body = root.find(f".//{{{SOAP_ENV}}}Body")
- if body is None:
- raise ValueError("SOAP Body not found")
- for child in list(body):
- return child
- raise ValueError("SOAP Body is empty")
-
-
-def soap_decode(xml: str, *, take: str | None = None) -> Any:
- """
- Decode a SOAP response into Python structures.
- - If take is provided, finds the first element with that localname anywhere
- under the SOAP Body payload and returns its decoded value.
- - If take is None, returns the decoded payload element.
- """
- payload = soap_body_payload(xml)
-
- if take is None:
- return {_localname(payload.tag): _xml_to_obj(payload)}
-
- for el in payload.iter():
- if _localname(el.tag) == take:
- return _xml_to_obj(el)
-
- raise KeyError(f"Element {take!r} not found under SOAP Body payload")
-
-
-# --------- SOAP encode helpers ---------
-
-
-def _append_value(parent: ET.Element, ns: str, name: str, value: Any) -> None:
- el = ET.SubElement(parent, f"{{{ns}}}{name}")
-
- if value is None:
- el.set(f"{{{XSI}}}nil", "true")
- return
-
- if isinstance(value, Mapping):
- for k, v in value.items():
- _append_value(el, ns, str(k), v)
- return
-
- if isinstance(value, (list, tuple)):
- # repeated child elements with the same name (common in SOAP doc/literal)
- # here we encode as - ...
by default
- for item in value:
- _append_value(el, ns, "item", item)
- return
-
- el.text = _format_scalar(value)
-
-
-def soap_encode(
- method: str,
- params: Mapping[str, Any] | None = None,
- *,
- method_ns: str,
- extra_method_xmlns: Mapping[str, str] | None = None,
-) -> str:
- """
- Builds a doc/literal style SOAP 1.1 envelope:
- ...
-
- extra_method_xmlns lets you add xmlns:prefix="uri" attributes on the method element.
- """
- env = ET.Element(f"{{{SOAP_ENV}}}Envelope")
- body = ET.SubElement(env, f"{{{SOAP_ENV}}}Body")
-
- # Method element uses the *default* namespace (no prefix) like your example.
- method_el = ET.SubElement(body, f"{{{method_ns}}}{method}")
-
- if extra_method_xmlns:
- for prefix, uri in extra_method_xmlns.items():
- method_el.set(f"xmlns:{prefix}", uri)
-
- for k, v in (params or {}).items():
- _append_value(method_el, method_ns, str(k), v)
-
- return ET.tostring(env, encoding="unicode", xml_declaration=False)
+__all__ = ["XSI", "soap_body_payload", "soap_decode", "soap_encode"]
diff --git a/pylabrobot/inheco/soap.py b/pylabrobot/inheco/soap.py
new file mode 100644
index 00000000000..8fec4c2b06f
--- /dev/null
+++ b/pylabrobot/inheco/soap.py
@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import datetime as _dt
+import re as _re
+import xml.etree.ElementTree as ET
+from typing import Any, Dict, List, Mapping, Optional, Union
+
+SOAP_ENV = "http://schemas.xmlsoap.org/soap/envelope/"
+XSI = "http://www.w3.org/2001/XMLSchema-instance"
+
+ET.register_namespace("s", SOAP_ENV)
+ET.register_namespace("xsi", XSI)
+
+
+# --------- scalar parsing/formatting ---------
+
+_INT_RE = _re.compile(r"^-?\d+$")
+_FLOAT_RE = _re.compile(r"^-?\d+\.\d+$")
+# ISO-8601 duration subset: PnDTnHnMnS (supports fractional seconds)
+_DUR_RE = _re.compile(
+ r"^P"
+ r"(?:(?P\d+)D)?"
+ r"(?:T"
+ r"(?:(?P\d+)H)?"
+ r"(?:(?P\d+)M)?"
+ r"(?:(?P\d+(?:\.\d+)?)S)?"
+ r")?$"
+)
+
+# Basic ISO datetime with optional fractional seconds and optional Z
+_DT_RE = _re.compile(
+ r"^(?P\d{4})-(?P\d{2})-(?P\d{2})"
+ r"T(?P\d{2}):(?P\d{2}):(?P\d{2})"
+ r"(?:\.(?P\d+))?"
+ r"(?PZ)?$"
+)
+
+
+def _parse_duration_iso8601(s: str) -> Optional[_dt.timedelta]:
+ m = _DUR_RE.match(s)
+ if not m:
+ return None
+ days = int(m.group("days") or 0)
+ hours = int(m.group("hours") or 0)
+ minutes = int(m.group("minutes") or 0)
+ sec_str = m.group("seconds") or "0"
+ seconds = float(sec_str) if sec_str else 0.0
+ return _dt.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
+
+
+def _format_duration_iso8601(td: _dt.timedelta) -> str:
+ total_seconds = td.total_seconds()
+ if total_seconds < 0:
+ raise ValueError("negative timedeltas not supported for ISO-8601 formatting here")
+ days = int(total_seconds // 86400)
+ rem = total_seconds - days * 86400
+ hours = int(rem // 3600)
+ rem -= hours * 3600
+ minutes = int(rem // 60)
+ seconds = rem - minutes * 60
+
+ out = "P"
+ if days:
+ out += f"{days}D"
+ if hours or minutes or seconds or not days:
+ out += "T"
+ if hours:
+ out += f"{hours}H"
+ if minutes:
+ out += f"{minutes}M"
+ # keep seconds even if 0 when nothing else was emitted in T
+ if seconds or (not hours and not minutes):
+ # avoid trailing .0 when it’s integral
+ if abs(seconds - round(seconds)) < 1e-12:
+ out += f"{int(round(seconds))}S"
+ else:
+ out += f"{seconds}S"
+ return out
+
+
+Scalar = Union[str, int, float, bool, _dt.timedelta, _dt.datetime]
+
+
+def _parse_scalar(text: Optional[str]) -> Scalar:
+ if text is None:
+ return ""
+ s = text.strip()
+ if s == "":
+ return ""
+
+ if s in ("true", "false"):
+ return s == "true"
+
+ dur = _parse_duration_iso8601(s)
+ if dur is not None and s.startswith("P"):
+ return dur
+
+ dm = _DT_RE.match(s)
+ if dm:
+ y, mo, d = int(dm["y"]), int(dm["m"]), int(dm["d"])
+ h, mi, se = int(dm["h"]), int(dm["mi"]), int(dm["s"])
+ frac = dm["frac"]
+ micro = int((frac or "0").ljust(6, "0")[:6])
+ dt = _dt.datetime(y, mo, d, h, mi, se, microsecond=micro)
+ if dm["z"] == "Z":
+ dt = dt.replace(tzinfo=_dt.timezone.utc)
+ return dt
+
+ if _INT_RE.match(s):
+ try:
+ return int(s)
+ except ValueError:
+ pass
+
+ if _FLOAT_RE.match(s):
+ try:
+ return float(s)
+ except ValueError:
+ pass
+
+ return s
+
+
+def _format_scalar(value: Any) -> str:
+ if isinstance(value, bool):
+ return "true" if value else "false"
+ if isinstance(value, int):
+ return str(value)
+ if isinstance(value, float):
+ return repr(value)
+ if isinstance(value, _dt.timedelta):
+ return _format_duration_iso8601(value)
+ if isinstance(value, _dt.datetime):
+ # if timezone-aware UTC, emit Z
+ if value.tzinfo is not None and value.utcoffset() == _dt.timedelta(0):
+ v = value.astimezone(_dt.timezone.utc).replace(tzinfo=None)
+ return v.isoformat(timespec="seconds") + "Z"
+ return value.isoformat()
+ return str(value)
+
+
+def _localname(tag: str) -> str:
+ # "{uri}Name" -> "Name"
+ return tag.split("}", 1)[1] if tag.startswith("{") else tag
+
+
+def _is_xsi_nil(el: ET.Element) -> bool:
+ nil_attr = el.attrib.get(f"{{{XSI}}}nil")
+ return (nil_attr or "").lower() == "true"
+
+
+# --------- generic XML -> Python structure ---------
+
+
+Object = Union[Dict[str, Union["Object", List["Object"]]], Scalar]
+
+
+def _xml_to_obj(el: ET.Element) -> Object:
+ if _is_xsi_nil(el):
+ return {}
+
+ children = list(el)
+ if not children:
+ return _parse_scalar(el.text)
+
+ # group by localname
+ grouped: Dict[str, list[ET.Element]] = {}
+ for ch in children:
+ grouped.setdefault(_localname(ch.tag), []).append(ch)
+
+ out: Object = {}
+ for name, items in grouped.items():
+ if len(items) == 1:
+ out[name] = _xml_to_obj(items[0]) # type: ignore
+ else:
+ out[name] = [_xml_to_obj(i) for i in items] # type: ignore
+ return out
+
+
+# --------- SOAP decode helpers ---------
+
+
+def soap_body_payload(xml: str) -> ET.Element:
+ """
+ Returns the first element inside SOAP Body (e.g., GetStatusResponse).
+ """
+ root = ET.fromstring(xml)
+ body = root.find(f".//{{{SOAP_ENV}}}Body")
+ if body is None:
+ raise ValueError("SOAP Body not found")
+ for child in list(body):
+ return child
+ raise ValueError("SOAP Body is empty")
+
+
+def soap_decode(xml: str, *, take: str | None = None) -> Any:
+ """
+ Decode a SOAP response into Python structures.
+ - If take is provided, finds the first element with that localname anywhere
+ under the SOAP Body payload and returns its decoded value.
+ - If take is None, returns the decoded payload element.
+ """
+ payload = soap_body_payload(xml)
+
+ if take is None:
+ return {_localname(payload.tag): _xml_to_obj(payload)}
+
+ for el in payload.iter():
+ if _localname(el.tag) == take:
+ return _xml_to_obj(el)
+
+ raise KeyError(f"Element {take!r} not found under SOAP Body payload")
+
+
+# --------- SOAP encode helpers ---------
+
+
+def _append_value(parent: ET.Element, ns: str, name: str, value: Any) -> None:
+ el = ET.SubElement(parent, f"{{{ns}}}{name}")
+
+ if value is None:
+ el.set(f"{{{XSI}}}nil", "true")
+ return
+
+ if isinstance(value, Mapping):
+ for k, v in value.items():
+ _append_value(el, ns, str(k), v)
+ return
+
+ if isinstance(value, (list, tuple)):
+ # repeated child elements with the same name (common in SOAP doc/literal)
+ # here we encode as - ...
by default
+ for item in value:
+ _append_value(el, ns, "item", item)
+ return
+
+ el.text = _format_scalar(value)
+
+
+def soap_encode(
+ method: str,
+ params: Mapping[str, Any] | None = None,
+ *,
+ method_ns: str,
+ extra_method_xmlns: Mapping[str, str] | None = None,
+) -> str:
+ """
+ Builds a doc/literal style SOAP 1.1 envelope:
+ ...
+
+ extra_method_xmlns lets you add xmlns:prefix="uri" attributes on the method element.
+ """
+ env = ET.Element(f"{{{SOAP_ENV}}}Envelope")
+ body = ET.SubElement(env, f"{{{SOAP_ENV}}}Body")
+
+ # Method element uses the *default* namespace (no prefix) like your example.
+ method_el = ET.SubElement(body, f"{{{method_ns}}}{method}")
+
+ if extra_method_xmlns:
+ for prefix, uri in extra_method_xmlns.items():
+ method_el.set(f"xmlns:{prefix}", uri)
+
+ for k, v in (params or {}).items():
+ _append_value(method_el, method_ns, str(k), v)
+
+ return ET.tostring(env, encoding="unicode", xml_declaration=False)
+
+
+def required_text(root: ET.Element, name: str, *, allow_empty: bool = False) -> str:
+ """Read a required direct child, allowing an XML namespace."""
+ child = root.find(f"{{*}}{name}")
+ if child is None or (child.text is None and not allow_empty):
+ raise ValueError(f"{root.tag} is missing {name}")
+ return child.text or ""
diff --git a/pylabrobot/inheco/transport/sila.py b/pylabrobot/inheco/transport/sila.py
index f36cffd0fc6..57c5274576b 100644
--- a/pylabrobot/inheco/transport/sila.py
+++ b/pylabrobot/inheco/transport/sila.py
@@ -1,333 +1,306 @@
-"""Inheco flavor of sila"""
+"""Inheco SiLA 1 SOAP sessions with correlated asynchronous callbacks."""
from __future__ import annotations
import asyncio
-import http.server
import logging
+import math
import random
-import socket
-import socketserver
-import urllib.parse
-import urllib.request
import xml.etree.ElementTree as ET
+from collections.abc import Callable
from dataclasses import dataclass
-from typing import Any, Optional, Tuple
-
-from pylabrobot.inheco.scila.soap import XSI, soap_decode, soap_encode
-
-SOAP_RESPONSE_ResponseEventResponse = """
-
-
-
- 1
- Success
- PT0.0006262S
- 0
-
-
-
-"""
-
-
-SOAP_RESPONSE_StatusEventResponse = """
-
-
-
- 1
- Success
- PT0.0005967S
- 0
-
-
-
-"""
-
-
-def _get_local_ip(machine_ip: str) -> str:
- s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- try:
- # Doesn't actually connect, just determines the route
- s.connect((machine_ip, 1))
- local_ip: str = s.getsockname()[0] # type: ignore
- if local_ip is None or local_ip.startswith("127."):
- raise RuntimeError("Could not determine local IP address.")
- finally:
- s.close()
- return local_ip
+from typing import Any
+
+from pylabrobot.inheco.soap import (
+ _parse_duration_iso8601,
+ required_text,
+ soap_body_payload,
+ soap_decode,
+ soap_encode,
+)
+from pylabrobot.io.http import HTTP
+from pylabrobot.io.http_server import HTTPServer, local_address_for
+
+logger = logging.getLogger(__name__)
+
+
+def response_data(root: ET.Element) -> ET.Element | None:
+ """Decode an escaped asynchronous ResponseData document."""
+ child = root.find("{*}responseData")
+ if child is None or not child.text or not child.text.strip():
+ return None
+ return ET.fromstring(child.text)
class SiLAError(RuntimeError):
- def __init__(self, code: int, message: str, command: str, details: Optional[dict] = None):
+ """A device refusal, retaining its command and original return code."""
+
+ def __init__(self, code: int, message: str, command: str) -> None:
self.code = code
self.message = message
self.command = command
- self.details = details or {}
+ super().__init__(f"{command}: {message} (SiLA code {code})")
+
+
+@dataclass
+class SiLACommand:
+ """An accepted command whose completion remains available after its callback."""
+
+ name: str
+ request_id: int
+ future: asyncio.Future[Any]
+ duration: float | None = None
+
+ async def wait(self, timeout: float | None = None) -> Any:
+ """Wait without canceling the instrument command when the caller stops waiting."""
+ if timeout is None:
+ fallback = 10800.0 if self.name == "ExecuteMethod" else 60.0
+ timeout = max(60.0, self.duration * 1.5) if self.duration else fallback
+ if not math.isfinite(timeout) or timeout <= 0:
+ raise ValueError("timeout must be positive")
+ return await asyncio.wait_for(asyncio.shield(self.future), timeout)
+
+
+def _observe_completion(future: asyncio.Future[Any]) -> None:
+ """Observe failures even if a waiter was canceled; future.result still raises them."""
+ if not future.cancelled():
+ future.exception()
class InhecoSiLAInterface:
- @dataclass(frozen=True)
- class _HTTPRequest:
- method: str
- path: str
- query: str
- headers: dict[str, str]
- body: bytes
-
- @dataclass(frozen=True)
- class _SiLACommand:
- name: str
- request_id: int
- fut: asyncio.Future[Any]
+ """One HTTP connection and callback endpoint, shared by Inheco devices.
+
+ Only the short HTTP exchanges are serialized by HTTP. Multiple asynchronous commands
+ may remain pending; the instrument decides which combinations it permits.
+ """
def __init__(
self,
machine_ip: str,
- client_ip: Optional[str] = None,
- logger: Optional[logging.Logger] = None,
+ client_ip: str | None = None,
+ *,
+ machine_port: int = 8080,
+ callback_port: int = 0,
+ on_data: Callable[[int, str], None] | None = None,
+ on_status: Callable[[ET.Element], None] | None = None,
) -> None:
- self._client_ip = client_ip or _get_local_ip(machine_ip)
self._machine_ip = machine_ip
- self._logger = logger or logging.getLogger(__name__)
-
- # single "in-flight token"
- self._making_request = asyncio.Lock()
+ self.machine_port = machine_port
+ self._client_ip = client_ip
+ self._callback_port = callback_port
+ self.io = HTTP("Inheco SiLA 1", f"http://{machine_ip}:{machine_port}")
+ self._server: HTTPServer | None = None
+ self._pending: dict[int, SiLACommand] = {}
+ self._next_id = random.randint(1, 2**30)
+ self._on_data = on_data
+ self._on_status = on_status
+ self._lock_id: str | None = None
- # pending command information
- self._pending: Optional[InhecoSiLAInterface._SiLACommand] = None
-
- # server plumbing
- self._loop: Optional[asyncio.AbstractEventLoop] = None
- self._httpd: Optional[socketserver.TCPServer] = None
- self._server_task: Optional[asyncio.Task[None]] = None
- self._closed = False
+ @property
+ def machine_ip(self) -> str:
+ """Return the instrument address."""
+ return self._machine_ip
@property
def client_ip(self) -> str:
+ """Return the selected callback interface after setup."""
+ if self._client_ip is None:
+ raise RuntimeError("SiLA session is not set up")
return self._client_ip
- @property
- def machine_ip(self) -> str:
- return self._machine_ip
-
@property
def bound_port(self) -> int:
- if self._httpd is None:
- raise RuntimeError("Server not started yet")
- return self._httpd.server_address[1]
-
- async def start(self) -> None:
- if self._httpd is not None:
- return
- if self._closed:
- raise RuntimeError("Bridge is closed")
-
- self._loop = asyncio.get_running_loop()
- outer = self
-
- class _Handler(http.server.BaseHTTPRequestHandler):
- server_version = "OneInFlightHTTPBridge/0.1"
-
- def log_message(self, fmt: str, *args) -> None:
- return # silence
-
- def _read_body(self) -> bytes:
- length = int(self.headers.get("Content-Length", "0") or "0")
- return self.rfile.read(length) if length else b""
-
- def _do(self) -> None:
- assert outer._loop is not None
-
- parsed = urllib.parse.urlsplit(self.path)
- req = InhecoSiLAInterface._HTTPRequest(
- method=self.command,
- path=parsed.path,
- query=parsed.query,
- headers={k.lower(): v for k, v in self.headers.items()},
- body=self._read_body(),
- )
-
- fut = asyncio.run_coroutine_threadsafe(outer._on_http(req), outer._loop)
- try:
- resp_body = fut.result()
- status = 200
- except Exception as e:
- resp_body = f"Internal Server Error: {type(e).__name__}: {e}\n".encode()
- status = 500
-
- self.send_response(status)
- self.send_header("Content-Type", "text/xml; charset=utf-8")
- self.send_header("Content-Length", str(len(resp_body)))
- self.end_headers()
- self.wfile.write(resp_body)
+ """Return the callback port after setup."""
+ if self._server is None:
+ raise RuntimeError("SiLA session is not set up")
+ return self._server.bound_port
- def do_GET(self) -> None:
- self._do()
-
- def do_POST(self) -> None:
- self._do()
-
- def do_PUT(self) -> None:
- self._do()
-
- def do_DELETE(self) -> None:
- self._do()
-
- # port 0: choose random free port
- self._httpd = http.server.ThreadingHTTPServer((self._client_ip, 0), _Handler)
-
- async def run_server() -> None:
- assert self._httpd is not None
- await asyncio.to_thread(self._httpd.serve_forever)
-
- self._server_task = asyncio.create_task(run_server(), name="http-server")
-
- async def close(self) -> None:
- self._closed = True
- if self._httpd is None:
+ @property
+ def event_receiver_uri(self) -> str:
+ """Return the address that the instrument must be able to reach."""
+ return f"http://{self.client_ip}:{self.bound_port}/"
+
+ def pending_command_name(self, request_id: int) -> str | None:
+ """Identify a callback's active request without retaining an unbounded event history."""
+ command = self._pending.get(request_id)
+ return command.name if command is not None else None
+
+ def fail_pending(self, error: Exception) -> None:
+ """Fail commands when the device reports a session-wide fault."""
+ for command in self._pending.values():
+ if not command.future.done():
+ command.future.set_exception(error)
+ self._pending.clear()
+
+ def _finish(self, request_id: int, event: ET.Element, *, error_event: bool = False) -> None:
+ """Complete one registered command from its response or error event."""
+ command = self._pending.get(request_id)
+ if command is None or command.future.done():
+ logger.debug("Ignoring late SiLA event for request %d", request_id)
return
-
- self._httpd.shutdown()
- self._httpd.server_close()
-
- if self._server_task is not None:
- await self._server_task
-
- self._httpd = None
- self._server_task = None
-
- async def _on_http(self, req: _HTTPRequest) -> bytes:
- """
- Called on the asyncio loop for every incoming HTTP request.
- If there's a pending command, try to match and resolve it.
- """
-
- cmd = self._pending
-
- if cmd is not None and not cmd.fut.done():
- response_event = soap_decode(req.body.decode("utf-8"))
- if "ResponseEvent" in response_event:
- request_id = response_event["ResponseEvent"].get("requestId")
- if request_id != cmd.request_id:
- self._logger.warning("Request ID does not match pending command.")
- else:
- return_value = response_event["ResponseEvent"].get("returnValue", {})
- return_code = return_value.get("returnCode")
- if return_code != 3: # error
- err_msg = return_value.get("message", "Unknown error").replace("\n", " ")
- cmd.fut.set_exception(
- RuntimeError(f"Command {cmd.name} failed with code {return_code}: '{err_msg}'")
- )
- else:
- response_data = response_event["ResponseEvent"].get("responseData", "")
- root = ET.fromstring(response_data)
- cmd.fut.set_result(root)
+ value = event.find("{*}returnValue")
+ if value is None:
+ raise ValueError("SiLA event is missing returnValue")
+ code = int(required_text(value, "returnCode"))
+ message = required_text(value, "message", allow_empty=True)
+ if error_event or code not in (1, 3, 12):
+ command.future.set_exception(SiLAError(code, message, command.name))
else:
- self._logger.warning("No pending command to match response to.")
-
- if "ResponseEvent" in req.body.decode("utf-8"):
- return SOAP_RESPONSE_ResponseEventResponse.encode("utf-8")
- if "StatusEvent" in req.body.decode("utf-8"):
- return SOAP_RESPONSE_StatusEventResponse.encode("utf-8")
- self._logger.warning("Unknown event type received.")
- return SOAP_RESPONSE_ResponseEventResponse.encode("utf-8")
-
- def _get_return_code_and_message(self, command_name: str, response: Any) -> Tuple[int, str]:
- resp_level = response.get(f"{command_name}Response", {}) # first level
- result_level = resp_level.get(f"{command_name}Result", {}) # second level
- return_code = result_level.get("returnCode")
- if return_code is None:
- raise ValueError(f"returnCode not found in response for {command_name}")
- return return_code, result_level.get("message", "")
-
- async def setup(self) -> None:
- await self.start()
-
- def _make_request_id(self):
- return random.randint(1, 2**31 - 1)
-
- async def send_command(
- self,
- command: str,
- **kwargs,
- ) -> Any:
- if self._closed:
- raise RuntimeError("Bridge is closed")
-
- request_id = self._make_request_id()
- cmd_xml = soap_encode(
- command,
- {"requestId": request_id, **kwargs},
- method_ns="http://sila.coop",
- extra_method_xmlns={"i": XSI},
- )
-
- # make POST request to machine
- url = f"http://{self._machine_ip}:8080/"
- req = urllib.request.Request(
- url=url,
- data=cmd_xml.encode("utf-8"),
- method="POST",
- headers={
- "Content-Type": "text/xml; charset=utf-8",
- "Content-Length": str(len(cmd_xml)),
- "SOAPAction": f"http://sila.coop/{command}",
- "Expect": "100-continue",
- "Accept-Encoding": "gzip, deflate",
- },
- )
-
- if self._making_request.locked():
- raise RuntimeError("can't send multiple commands at the same time")
-
- async with self._making_request:
- try:
-
- def _do_request() -> bytes:
- with urllib.request.urlopen(req) as resp:
- return resp.read() # type: ignore
-
- body = await asyncio.to_thread(_do_request)
- return_code, message = self._get_return_code_and_message(
- command, soap_decode(body.decode("utf-8"))
- )
- if return_code == 1: # success
- return soap_decode(body.decode("utf-8"))
- elif return_code == 2: # concurrent command
- fut: asyncio.Future[Any] = asyncio.get_running_loop().create_future()
- self._pending = InhecoSiLAInterface._SiLACommand(
- name=command, request_id=request_id, fut=fut
- )
- return await fut # wait for response to be handled in _on_http
+ result = response_data(event)
+ if code == 12:
+ logger.warning("%s completed with warning: %s", command.name, message)
+ command.future.set_result(result)
+ self._pending.pop(request_id, None)
+
+ async def _handle_event(self, body: bytes) -> bytes:
+ """Dispatch a SOAP callback and acknowledge the exact event type."""
+ event = soap_body_payload(body.decode("utf-8"))
+ name = event.tag.rsplit("}", 1)[-1]
+ if name not in {"ResponseEvent", "StatusEvent", "DataEvent", "ErrorEvent"}:
+ raise ValueError(f"Unsupported SiLA event {name}")
+ request_id: int | None = None
+ try:
+ if name == "StatusEvent":
+ if self._on_status is not None:
+ self._on_status(event)
+ else:
+ request_id = int(required_text(event, "requestId"))
+ if name == "DataEvent":
+ if self._on_data is not None:
+ self._on_data(request_id, required_text(event, "dataValue"))
else:
- raise RuntimeError(f"command {command} failed: {return_code} {message}")
- finally:
- self._pending = None
+ self._finish(request_id, event, error_event=name == "ErrorEvent")
+ except (ValueError, ET.ParseError) as error:
+ logger.exception("Malformed %s", name)
+ command = self._pending.pop(request_id, None) if request_id is not None else None
+ if command is not None and not command.future.done():
+ command.future.set_exception(error)
+ raise
+ return soap_encode(
+ f"{name}Response",
+ {
+ f"{name}Result": {
+ "returnCode": 1,
+ "message": "Success",
+ "duration": "PT0S",
+ "deviceClass": 0,
+ }
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+ async def setup(self) -> None:
+ """Open HTTP and callback transports without resetting the instrument."""
+ if self._server is not None:
+ return
+ if self._client_ip is None:
+ self._client_ip = await local_address_for(self.machine_ip, self.machine_port)
+ await self.io.setup()
+ server = HTTPServer(self._handle_event, self.client_ip, self._callback_port)
+ try:
+ await server.setup()
+ except BaseException:
+ await server.stop()
+ await self.io.stop()
+ raise
+ self._server = server
-def _parse_scalar(text: Optional[str], tag: str) -> object:
+ async def close(self) -> None:
+ """Close transports and notify pending waiters of the lost connection."""
+ self.fail_pending(ConnectionError("SiLA connection closed"))
+ try:
+ if self._server is not None:
+ await self._server.stop()
+ finally:
+ self._server = None
+ self._lock_id = None
+ await self.io.stop()
+
+ async def start_command(self, command: str, **kwargs: Any) -> SiLACommand:
+ """Register before sending so an early callback cannot be lost."""
+ if self._server is None:
+ raise RuntimeError("SiLA session is not set up")
+ self._next_id = self._next_id % (2**31 - 1) + 1
+ request_id = self._next_id
+ pending = SiLACommand(command, request_id, asyncio.get_running_loop().create_future())
+ pending.future.add_done_callback(_observe_completion)
+ pending.future.add_done_callback(lambda _: self._pending.pop(request_id, None))
+ self._pending[request_id] = pending
+ if self._lock_id is not None:
+ kwargs.setdefault("lockId", self._lock_id)
+ body = soap_encode(command, {"requestId": request_id, **kwargs}, method_ns="http://sila.coop")
+ try:
+ response = await self.io.request_raw(
+ "POST",
+ "/",
+ body.encode(),
+ headers={
+ "Content-Type": "text/xml; charset=utf-8",
+ "SOAPAction": f"http://sila.coop/{command}",
+ },
+ allow_redirects=False,
+ )
+ if response.status != 200:
+ raise ConnectionError(f"{command}: HTTP {response.status}: {response.text()}")
+ root = soap_body_payload(response.text())
+ if root.tag.rsplit("}", 1)[-1] != f"{command}Response":
+ raise ValueError(f"Unexpected response to {command}: {root.tag}")
+ result = root.find(f"{{*}}{command}Result")
+ if result is None:
+ raise ValueError(f"Missing {command}Result")
+ code = int(required_text(result, "returnCode"))
+ message = required_text(result, "message", allow_empty=True)
+ duration = result.find("{*}duration")
+ if duration is not None and duration.text:
+ parsed = _parse_duration_iso8601(duration.text)
+ if parsed is None:
+ raise ValueError(f"Invalid SiLA duration: {duration.text}")
+ pending.duration = parsed.total_seconds()
+ if code == 2:
+ return pending
+ if code not in (1, 3, 12):
+ raise SiLAError(code, message, command)
+ if code == 12:
+ logger.warning("%s completed with warning: %s", command, message)
+ if not pending.future.done():
+ pending.future.set_result(soap_decode(response.text()))
+ self._pending.pop(request_id, None)
+ return pending
+ except BaseException:
+ self._pending.pop(request_id, None)
+ if not pending.future.done():
+ pending.future.cancel()
+ raise
+
+ async def send_command(self, command: str, *, timeout: float | None = None, **kwargs: Any) -> Any:
+ """Send a command and await its synchronous response or completion callback."""
+ pending = await self.start_command(command, **kwargs)
+ return await pending.wait(timeout)
+
+
+def _parse_scalar(text: str | None, tag: str) -> object:
+ """Convert a typed SiLA Parameter value."""
if text is None:
- return None
- s = text.strip()
- t = tag.lower()
- if t in ("float64", "float32", "double", "single"):
- return float(s)
- if t in ("int32", "int64", "uint32", "uint64", "integer"):
- return int(s)
- if t in ("boolean", "bool"):
- return s.lower() == "true"
- return s
-
-
-def _get_param(root: ET.Element, name: str):
- p = root.find(f".//Parameter[@name='{name}']")
- if p is None or len(p) == 0:
- raise RuntimeError(f"Response missing parameter '{name}'")
- child = next(iter(p))
- return _parse_scalar(child.text, child.tag)
+ raise ValueError(f"Missing {tag} parameter value")
+ tag = tag.rsplit("}", 1)[-1].lower()
+ if tag in ("float64", "float32", "double", "single"):
+ return float(text)
+ if tag in ("int32", "int64", "uint32", "uint64", "integer"):
+ return int(text)
+ if tag in ("boolean", "bool"):
+ if text.lower() not in ("true", "false"):
+ raise ValueError(f"Invalid boolean {text!r}")
+ return text.lower() == "true"
+ return text
def get_params(root: ET.Element, names: list[str]) -> dict[str, object]:
- return {n: _get_param(root, n) for n in names}
+ """Read named parameters from a SiLA ResponseData document."""
+ parameters = {p.attrib["name"]: p for p in root.findall(".//{*}Parameter")}
+ result = {}
+ for name in names:
+ parameter = parameters[name]
+ if len(parameter) != 1:
+ raise ValueError(f"Parameter {name} must contain one typed value")
+ result[name] = _parse_scalar(parameter[0].text, parameter[0].tag)
+ return result
diff --git a/pylabrobot/inheco/transport/sila_tests.py b/pylabrobot/inheco/transport/sila_tests.py
new file mode 100644
index 00000000000..f83cc621723
--- /dev/null
+++ b/pylabrobot/inheco/transport/sila_tests.py
@@ -0,0 +1,142 @@
+"""SiLA acceptance, callback, fault, and timeout behavior at the HTTP boundary."""
+
+import asyncio
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from pylabrobot.inheco.soap import soap_body_payload, soap_encode
+from pylabrobot.inheco.transport.sila import InhecoSiLAInterface, SiLAError, required_text
+from pylabrobot.io.http import HTTPResponse
+
+
+def response(command, code=2, message="Success", duration="PT10S"):
+ return HTTPResponse(
+ 200,
+ soap_encode(
+ f"{command}Response",
+ {f"{command}Result": {"returnCode": code, "message": message, "duration": duration}},
+ method_ns="http://sila.coop",
+ ).encode(),
+ {},
+ )
+
+
+def event(request_id, code=3, name="ResponseEvent", data=""):
+ return soap_encode(
+ name,
+ {
+ "requestId": request_id,
+ "returnValue": {"returnCode": code, "message": "device message"},
+ "responseData": data,
+ },
+ method_ns="http://sila.coop",
+ ).encode()
+
+
+class SiLATests(unittest.IsolatedAsyncioTestCase):
+ async def asyncSetUp(self):
+ self.server_patch = patch("pylabrobot.inheco.transport.sila.HTTPServer")
+ server = self.server_patch.start().return_value
+ server.setup = AsyncMock()
+ server.stop = AsyncMock()
+ server.bound_port = 8000
+ self.session = InhecoSiLAInterface("127.0.0.1", client_ip="127.0.0.1")
+ await self.session.setup()
+ self.post_patch = patch.object(self.session.io, "request_raw", new_callable=AsyncMock)
+ self.post = self.post_patch.start()
+
+ async def asyncTearDown(self):
+ await self.session.close()
+ self.post_patch.stop()
+ self.server_patch.stop()
+
+ async def test_callback_before_acceptance_is_retained(self):
+ async def post(method, path, body, **kwargs):
+ request = soap_body_payload(body.decode())
+ request_id = int(required_text(request, "requestId"))
+ await self.session._handle_event(event(request_id))
+ return response("OpenDoor")
+
+ self.post.side_effect = post
+ pending = await self.session.start_command("OpenDoor")
+ self.assertEqual((await pending.wait()).tag, "ResponseData")
+ self.assertEqual(pending.duration, 10)
+
+ async def test_error_is_still_raised_by_later_waiter(self):
+ self.post.return_value = response("ExecuteMethod")
+ pending = await self.session.start_command("ExecuteMethod", methodName="PCR")
+ await self.session._handle_event(event(pending.request_id, 2002))
+ await asyncio.sleep(0)
+ for _ in range(2):
+ with self.assertRaises(SiLAError) as error:
+ await pending.wait()
+ self.assertEqual(error.exception.code, 2002)
+
+ async def test_timeout_and_cancellation_do_not_cancel_command(self):
+ self.post.return_value = response("ExecuteMethod")
+ pending = await self.session.start_command("ExecuteMethod")
+ with self.assertRaises(asyncio.TimeoutError):
+ await pending.wait(0.001)
+ self.assertFalse(pending.future.done())
+ waiter = asyncio.create_task(pending.wait())
+ await asyncio.sleep(0)
+ waiter.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await waiter
+ self.assertFalse(pending.future.done())
+ await self.session._handle_event(event(pending.request_id))
+ await pending.wait()
+
+ async def test_concurrent_reads_do_not_wait_for_method_completion(self):
+ self.post.return_value = response("ExecuteMethod")
+ pending = await self.session.start_command("ExecuteMethod")
+ self.post.return_value = response("GetStatus", 1)
+ await self.session.send_command("GetStatus")
+ self.assertFalse(pending.future.done())
+ await self.session.close()
+ with self.assertRaises(ConnectionError):
+ await pending.wait()
+
+ async def test_busy_rejection_retains_firmware_error(self):
+ self.post.return_value = response("OpenDoor", 4, "Busy with ExecuteMethod")
+ with self.assertRaisesRegex(SiLAError, "Busy with ExecuteMethod"):
+ await self.session.send_command("OpenDoor")
+ self.assertFalse(self.session._pending)
+
+ async def test_finished_with_warning_logs_and_succeeds(self):
+ self.post.return_value = response("OpenDoor")
+ pending = await self.session.start_command("OpenDoor")
+ with self.assertLogs("pylabrobot.inheco.transport.sila", level="WARNING"):
+ await self.session._handle_event(event(pending.request_id, 12))
+ await pending.wait()
+
+ async def test_response_code_one_and_empty_payload(self):
+ self.post.return_value = response("OpenDoor")
+ pending = await self.session.start_command("OpenDoor")
+ await self.session._handle_event(event(pending.request_id, 1, data=""))
+ self.assertIsNone(await pending.wait())
+
+ async def test_error_event_fails_corresponding_command(self):
+ self.post.return_value = response("ExecuteMethod")
+ pending = await self.session.start_command("ExecuteMethod")
+ await self.session._handle_event(event(pending.request_id, 2001, name="ErrorEvent"))
+ with self.assertRaises(SiLAError):
+ await pending.wait()
+
+ async def test_malformed_callback_fails_waiter(self):
+ self.post.return_value = response("ReadActualTemperature")
+ pending = await self.session.start_command("ReadActualTemperature")
+ with self.assertRaises(Exception):
+ await self.session._handle_event(event(pending.request_id, data=" str:
+ """Find the IPv4 interface used to reach a device, without sending a packet."""
+
+ def resolve() -> str:
+ """Ask the routing table for the outgoing interface."""
+ with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as connection:
+ connection.connect((host, port))
+ return str(connection.getsockname()[0])
+
+ return await asyncio.to_thread(resolve)
+
+
+class HTTPServer:
+ """A POST callback endpoint whose handler receives and returns raw bytes.
+
+ The server accepts Content-Length framed requests and closes each connection after its
+ response. Requests, including Expect: 100-continue, are handled on the owning event loop.
+ Incoming exchanges are recorded alongside outgoing HTTP traffic during capture.
+ """
+
+ def __init__(
+ self,
+ handler: Callable[[bytes], Awaitable[bytes]],
+ host: str,
+ port: int = 0,
+ *,
+ max_body_size: int = 8 * 1024 * 1024,
+ timeout: float = 30,
+ ) -> None:
+ if max_body_size <= 0 or timeout <= 0:
+ raise ValueError("max_body_size and timeout must be positive")
+ self.handler = handler
+ self.host = host
+ self.port = port
+ self.max_body_size = max_body_size
+ self.timeout = timeout
+ self._server: asyncio.Server | None = None
+ self._clients: set[asyncio.Task] = set()
+
+ @property
+ def bound_port(self) -> int:
+ """Return the listening port after setup."""
+ if self._server is None:
+ raise RuntimeError("HTTP callback server is not set up")
+ return int(self._server.sockets[0].getsockname()[1])
+
+ @property
+ def url(self) -> str:
+ """Return the callback URL advertised to the device."""
+ return f"http://{self.host}:{self.bound_port}/"
+
+ async def _respond(self, writer: asyncio.StreamWriter, status: str, body: bytes) -> None:
+ """Send one response and request connection closure."""
+ writer.write(
+ f"HTTP/1.1 {status}\r\nContent-Type: text/xml; charset=utf-8\r\n"
+ f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n".encode()
+ + body
+ )
+ await writer.drain()
+
+ async def _exchange(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+ """Read a framed request, invoke the handler, and record its response."""
+ header = (await reader.readuntil(b"\r\n\r\n")).decode("iso-8859-1")
+ lines = header.split("\r\n")
+ method, path, _ = lines[0].split(" ", 2)
+ headers = {}
+ for line in lines[1:]:
+ if line:
+ key, value = line.split(":", 1)
+ key = key.lower().strip()
+ if key in headers:
+ raise ValueError("Duplicate HTTP header")
+ headers[key] = value.strip()
+ if method != "POST":
+ await self._respond(writer, "405 Method Not Allowed", b"POST required")
+ return
+ if "transfer-encoding" in headers or "content-length" not in headers:
+ await self._respond(writer, "411 Length Required", b"Content-Length required")
+ return
+ length = int(headers["content-length"])
+ if not 0 <= length <= self.max_body_size:
+ await self._respond(writer, "413 Content Too Large", b"Callback too large")
+ return
+ if headers.get("expect", "").lower() == "100-continue":
+ writer.write(b"HTTP/1.1 100 Continue\r\n\r\n")
+ await writer.drain()
+ body = await reader.readexactly(length)
+ response = await self.handler(body)
+ if capturer.capture_active:
+ capturer.record(
+ HTTPRawCommand(
+ module="http_server",
+ device_id=self.url,
+ action=method,
+ path=path,
+ request_base64=base64.b64encode(body).decode("ascii"),
+ status=200,
+ response_base64=base64.b64encode(response).decode("ascii"),
+ )
+ )
+ await self._respond(writer, "200 OK", response)
+
+ async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+ """Bound the lifetime of a connection and always release its socket."""
+ task = asyncio.current_task()
+ assert task is not None
+ self._clients.add(task)
+ try:
+ await asyncio.wait_for(self._exchange(reader, writer), self.timeout)
+ except (ValueError, asyncio.IncompleteReadError, asyncio.LimitOverrunError):
+ await self._respond(writer, "400 Bad Request", b"Malformed callback")
+ except (ConnectionError, asyncio.TimeoutError):
+ logger.debug("HTTP callback disconnected or timed out")
+ except Exception:
+ logger.exception("HTTP callback handler failed")
+ await self._respond(writer, "500 Internal Server Error", b"Callback handler failed")
+ finally:
+ writer.close()
+ try:
+ await writer.wait_closed()
+ except ConnectionError:
+ pass
+ self._clients.discard(task)
+
+ async def stop(self) -> None:
+ """Close the listener and cancel unfinished callbacks."""
+ if self._server is not None:
+ self._server.close()
+ await self._server.wait_closed()
+ self._server = None
+ clients = list(self._clients)
+ for task in clients:
+ task.cancel()
+ await asyncio.gather(*clients, return_exceptions=True)
+
+ async def setup(self) -> None:
+ """Start listening; repeated setup leaves the existing endpoint intact."""
+ if self._server is None:
+ self._server = await asyncio.start_server(self._handle, self.host, self.port)
+ if not self._server.sockets:
+ await self.stop()
+ raise OSError("HTTP callback server could not bind a listening socket")
diff --git a/pylabrobot/io/http_server_tests.py b/pylabrobot/io/http_server_tests.py
new file mode 100644
index 00000000000..e263fbb51f6
--- /dev/null
+++ b/pylabrobot/io/http_server_tests.py
@@ -0,0 +1,81 @@
+"""Exercise callback framing and lifecycle against a loopback HTTP listener."""
+
+import asyncio
+import unittest
+from unittest.mock import AsyncMock
+
+from pylabrobot.io.http import HTTP
+from pylabrobot.io.http_server import HTTPServer
+
+
+class HTTPServerTests(unittest.IsolatedAsyncioTestCase):
+ async def asyncSetUp(self):
+ self.handler = AsyncMock(return_value=b"")
+ self.server = HTTPServer(self.handler, "127.0.0.1", max_body_size=1024)
+ await self.server.setup()
+ self.client = HTTP("callback test", self.server.url)
+ await self.client.setup()
+
+ async def asyncTearDown(self):
+ await self.client.stop()
+ await self.server.stop()
+
+ async def test_raw_callback_roundtrip_and_idempotent_setup(self):
+ port = self.server.bound_port
+ await self.server.setup()
+ self.assertEqual(self.server.bound_port, port)
+ response = await self.client.request_raw("POST", "/", b"")
+ self.assertEqual((response.status, response.body), (200, b""))
+ self.handler.assert_awaited_once_with(b"")
+
+ async def test_body_limit_prevents_dispatch(self):
+ response = await self.client.request_raw("POST", "/", b"x" * 1025)
+ self.assertEqual(response.status, 413)
+ self.handler.assert_not_awaited()
+
+ async def test_expect_continue_and_partial_body(self):
+ reader, writer = await asyncio.open_connection("127.0.0.1", self.server.bound_port)
+ writer.write(b"POST / HTTP/1.1\r\nContent-Length: 8\r\nExpect: 100-continue\r\n\r\n")
+ await writer.drain()
+ self.assertEqual(await reader.readuntil(b"\r\n\r\n"), b"HTTP/1.1 100 Continue\r\n\r\n")
+ writer.write(b"")
+ await writer.drain()
+ self.assertIn(b"", await reader.read())
+ writer.close()
+ await writer.wait_closed()
+
+ async def test_stop_cancels_incomplete_request_and_allows_restart(self):
+ reader, writer = await asyncio.open_connection("127.0.0.1", self.server.bound_port)
+ writer.write(b"POST / HTTP/1.1\r\nContent-Length: 100\r\n\r\nx")
+ await writer.drain()
+ await asyncio.sleep(0)
+ await self.server.stop()
+ self.assertEqual(await asyncio.wait_for(reader.read(), 1), b"")
+ self.assertFalse(self.server._clients)
+ writer.close()
+ await writer.wait_closed()
+ await self.server.setup()
+ self.assertGreater(self.server.bound_port, 0)
+
+ async def test_capture_records_callback_request_and_response(self):
+ import base64
+ import json
+ import tempfile
+ from pathlib import Path
+
+ from pylabrobot.io.capture import start_capture, stop_capture
+
+ with tempfile.TemporaryDirectory() as directory:
+ path = Path(directory) / "callback.json"
+ start_capture(path)
+ try:
+ await self.client.request_raw("POST", "/", b"")
+ finally:
+ stop_capture()
+ commands = json.loads(path.read_text())["commands"]
+ callback = next(command for command in commands if command["module"] == "http_server")
+ self.assertEqual(base64.b64decode(callback["request_base64"]), b"")
+ self.assertEqual(base64.b64decode(callback["response_base64"]), b"")