diff --git a/docs/api/pylabrobot.formulatrix.rst b/docs/api/pylabrobot.formulatrix.rst index b0c31326126..51f608179db 100644 --- a/docs/api/pylabrobot.formulatrix.rst +++ b/docs/api/pylabrobot.formulatrix.rst @@ -16,9 +16,13 @@ Mantis driver.master.MantisDriver driver.simulator.MantisSimulationDriver driver.simulator.MantisSimulationPlateClamp + driver.simulator.MantisSimulationPump driver.simulator.MantisSimulationRfid + driver.simulator.MantisSimulationLedBar driver.features.plate_clamp.PlateClamp + driver.features.pump.Pump driver.features.rfid.Rfid + driver.features.led_bar.LedBar driver.configuration.DeviceConfiguration driver.errors.MantisError driver.errors.FrameError diff --git a/docs/user_guide/formulatrix/mantis/hello-world.ipynb b/docs/user_guide/formulatrix/mantis/hello-world.ipynb index 91cae7f86c5..4667bf4896d 100644 --- a/docs/user_guide/formulatrix/mantis/hello-world.ipynb +++ b/docs/user_guide/formulatrix/mantis/hello-world.ipynb @@ -7,7 +7,7 @@ "source": [ "# Formulatrix Mantis\n", "\n", - "The Mantis is a liquid dispenser from Formulatrix that dispenses through exchangeable chips. PyLabRobot connects to it, identifies it, reads its firmware versions, name and motor limits, locks and unlocks the plate clamp, and reads which chip sits in each input slot. Nothing else in this driver moves the device, switches the pump on or opens a valve." + "The Mantis is a liquid dispenser from Formulatrix that dispenses through exchangeable chips. PyLabRobot connects to it, identifies it, reads its firmware versions, name, motors and pressure sensors, locks and unlocks the plate clamp, reads which chip sits in each input slot, and shows colours on a v4's LED bar. Nothing else in this driver moves the device, switches the pump on or opens a valve." ] }, { @@ -39,12 +39,13 @@ "| | |\n", "|---|---|\n", "| Generations | v3 and v4, told apart at setup |\n", - "| Reads | firmware versions, device name, motor limits |\n", + "| Reads | firmware versions, device name, motor limits and state, the arm's absolute encoders (v4), line pressures and the pressure board's state |\n", "| Interface | USB, an FTDI bridge (VID 0x0403 / PID 0x6010): interface B for commands, interface A for the RFID reader |\n", "| Plate clamp | lock at one of the unit's levels, waiting for clamping done; unlock |\n", "| RFID | the chip in each input slot: serial, chip type, date of manufacture |\n", + "| LED bar (v4) | one colour, one per LED, or an animation; shows setup, off at stop |\n", "| Unit configuration | the clamp's lock levels and each slot's RFID antenna, from the unit's vendor folder or a saved file; else a recording of its generation |\n", - "| Not supported yet | motion and the motor reads, pneumatics and the pressure reads, chips, dispensing |" + "| Not supported yet | motion, switching the pump or a pressure controller, chips, dispensing |" ] }, { @@ -80,7 +81,7 @@ "\n", "Some of a Mantis's values are set per unit and cannot be read off the device; they sit in the `.config` files of its vendor install, in `Data/Device/Configs`. Give setup that folder once, then save the configuration: the saved file carries the unit's values from then on, and the folder is no longer needed. So far these are the plate clamp's lock levels and each input slot's RFID antenna.\n", "\n", - "Without them, a recording of a unit of the same generation supplies them, with a warning: the Mantis runs, less accurately. Where the folder and a saved file both hold a value, they must agree, or setup refuses." + "Without them, a recording of a unit of the same generation supplies them, with a warning: the Mantis runs, less accurately. Where the folder and a saved file both hold a value, they must agree, or setup refuses. A file saved by an earlier version still loads: a value it lacks comes from the folder, else the recording, with one warning naming it." ] }, { @@ -102,7 +103,7 @@ "source": [ "## Setup\n", "\n", - "`setup()` opens both links, reads the firmware versions and the motor limits, takes the unit's values and sets up the RFID reader. It does not move anything." + "`setup()` opens both links, reads the firmware versions, the motor limits and the pressure sensors' calibration, takes the unit's values and sets up the RFID reader. It does not move anything. On a v4 the LED bar shows the setup, green for a second at its end, then the idle animation." ] }, { @@ -143,7 +144,7 @@ "source": [ "### Simulation\n", "\n", - "`Mantis(simulation=True)` builds a Mantis with nothing behind it; nothing reaches a wire. It takes its configuration, lock levels and RFID antennas from a Mantis v4's recording (`RECORDING_MANTIS_V4`, or a file given as `declared_configuration_json`), its reads answer from it, its plate clamp's state changes at once, and every declared RFID slot reads empty. The device name is not simulated." + "`Mantis(simulation=True)` builds a Mantis with nothing behind it; nothing reaches a wire. It takes its configuration, lock levels and RFID antennas from a Mantis v4's recording (`RECORDING_MANTIS_V4`, or a file given as `declared_configuration_json`), its reads answer from it, its plate clamp's state changes at once, every declared RFID slot reads empty, and its LED bar takes every command. Its motors read idle at 0; its pump reads off, every line at 0, with the recorded sensor calibration. The device name is not simulated." ] }, { @@ -259,6 +260,190 @@ "mantis.driver.configuration.motor_limits" ] }, + { + "cell_type": "markdown", + "id": "motor-state", + "metadata": {}, + "source": [ + "### Motor state\n", + "\n", + "A motor's status word; `MOTOR_STATUS_ERROR_BITS` names the bits that report a fault:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-status", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_motor_status(motor=0)" + ] + }, + { + "cell_type": "markdown", + "id": "motor-positions", + "metadata": {}, + "source": [ + "Its demand position, where the controller drives it. It is not measured: it reads back the last position declared to the motor until the motor moves. Its servo encoder position counts only while the device is powered, from about 0 at power-on wherever the motor sits." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-position", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_motor_position(motor=0)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-encoder-position", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_encoder_position(motor=0)" + ] + }, + { + "cell_type": "markdown", + "id": "motor-settings", + "metadata": {}, + "source": [ + "Its following-error check (enabled, the largest following error), its three currents and its configuration, as the device sends them:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-following-error", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_following_error_configuration(motor=0)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-currents", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_motor_currents(motor=0)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "motor-configuration", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_motor_configuration(motor=0)" + ] + }, + { + "cell_type": "markdown", + "id": "absolute-encoders", + "metadata": {}, + "source": [ + "### Absolute encoders (v4)\n", + "\n", + "Board 11 reads the absolute encoder of each arm motor, 0 and 1, kept across power-off. The reading is raw, not yet an angle. Each motor's first read after connecting is not valid; the driver reads again until one is." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "absolute-encoder-position", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.driver.request_absolute_encoder_position(motor=0)" + ] + }, + { + "cell_type": "markdown", + "id": "pressure", + "metadata": {}, + "source": [ + "## Pressure\n", + "\n", + "The pump exists only when the pressure board (board 10) answered at setup. Its three pressure controllers, `\"bottle\"`, `\"chip\"` and `\"vacuum\"`, each regulate one line from a feedback sensor. Nothing here switches the pump or a controller on.\n", + "\n", + "Each sensor's calibration, (scale, offset), as setup read it; it is never written nor saved. The offsets are re-zeroed on the device between sessions, so setup reads them every time:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "pressure-calibration", + "metadata": {}, + "outputs": [], + "source": [ + "mantis.driver.configuration.feedback_sensor_parameters" + ] + }, + { + "cell_type": "markdown", + "id": "pressure-measure", + "metadata": {}, + "source": [ + "A line's pressure, in psi; inHg on the vacuum:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "pressure-measure-chip", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.pump.measure_pressure(controller=\"chip\")" + ] + }, + { + "cell_type": "markdown", + "id": "pressure-state", + "metadata": {}, + "source": [ + "Whether the pump runs, whether a controller regulates, and its status: 0 off, 1 settled, 2 unsettled." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "pump-on", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.pump.request_pump_on()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "controller-enabled", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.pump.request_controller_enabled(controller=\"chip\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "controller-status", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.pump.request_controller_status(controller=\"chip\")" + ] + }, { "cell_type": "markdown", "id": "mantis-clamp-text", @@ -375,6 +560,152 @@ "await mantis.rfid.request_tags()" ] }, + { + "cell_type": "markdown", + "id": "led-bar", + "metadata": {}, + "source": [ + "## LED bar\n", + "\n", + "The v4's LED bar shows what the device does: blue while setup runs, green for a second when it is done, then the idle animation; off at stop. An action that fails leaves it red. It exists only on a v4 whose LC3 board (board 12) answered at setup; a v3 has none. Its LED count is the unit's (`NumberOfLed` in `Device.config`), as declared:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-count", + "metadata": {}, + "outputs": [], + "source": [ + "mantis.led_bar.number_of_leds" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-color-text", + "metadata": {}, + "source": [ + "Show one colour on every LED: a name (`\"blue\"`, `\"green\"`, ...), a web hex or an RGB565 value. Any animation stops first:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-color", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.set_color(color=\"#FF0080\")" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-per-led-text", + "metadata": {}, + "source": [ + "Or one colour per LED, as many as it has:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-per-led", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.set_color(color=[\"green\"] * 10 + [\"off\"] * 10)" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-action-color-text", + "metadata": {}, + "source": [ + "Each action's colour is an attribute of the bar (`setup_color`, `setup_done_color`, `error_color`, `dispense_progress_color`, ...), checked when it is set; each bar has its own:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-action-color", + "metadata": {}, + "outputs": [], + "source": [ + "mantis.led_bar.setup_color = \"#FF0080\"" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-progress-text", + "metadata": {}, + "source": [ + "Show progress: a share of the LEDs from the left in `dispense_progress_color`, the rest off. Each frame holds the link about 62 ms, so show at most one per LED in an action:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-progress", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.show_progress(done=3, total=8)" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-start-animation-text", + "metadata": {}, + "source": [ + "Play the idle animation, `\"wave\"`, repeating:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-start-animation", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.start_animation(effect=\"wave\")" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-stop-animation-text", + "metadata": {}, + "source": [ + "Stop it; the bar holds what it shows:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-stop-animation", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.stop_animation()" + ] + }, + { + "cell_type": "markdown", + "id": "led-bar-off-text", + "metadata": {}, + "source": [ + "Turn every LED off. `Mantis(show_actions=False)` shows no action's colour, only the idle animation from setup on." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "led-bar-off", + "metadata": {}, + "outputs": [], + "source": [ + "await mantis.led_bar.turn_off()" + ] + }, { "cell_type": "markdown", "id": "mantis-20", diff --git a/pylabrobot/formulatrix/mantis/__init__.py b/pylabrobot/formulatrix/mantis/__init__.py index de93b8450f3..090428c3ebb 100644 --- a/pylabrobot/formulatrix/mantis/__init__.py +++ b/pylabrobot/formulatrix/mantis/__init__.py @@ -6,12 +6,15 @@ CONFIRMED_FIRMWARE_VERSIONS, MOTOR_ERROR_CODE_DESCRIPTIONS, MOTOR_STATUS_ERROR_BITS, + PRESSURE_CONTROLLER_STATUS_DESCRIPTIONS, + PRESSURE_CONTROLLERS, SENSOR_ALARM_CODE_DESCRIPTIONS, ChecksumError, CommandRefusedError, DeviceConfiguration, FrameError, InvalidQueueItemError, + LedBar, MantisConfigurationError, MantisConnectionLostError, MantisCoordinateError, @@ -23,11 +26,15 @@ MantisPressureError, MantisPrimeError, MantisSimulationDriver, + MantisSimulationLedBar, MantisSimulationPlateClamp, + MantisSimulationPump, MantisSimulationRfid, MantisTimeoutError, PlateClamp, PlateClampError, + PressureController, + Pump, Rfid, ) from .driver.configuration import RECORDING_MANTIS_V3 diff --git a/pylabrobot/formulatrix/mantis/design.md b/pylabrobot/formulatrix/mantis/design.md index 6cec7758966..619ed12bb49 100644 --- a/pylabrobot/formulatrix/mantis/design.md +++ b/pylabrobot/formulatrix/mantis/design.md @@ -10,10 +10,11 @@ last section lists what is open and where the code does not follow a rule yet. device.py MantisDevice, a Resource, and the Mantis() factory. setup() / stop(), the driver, the parts. driver/master.py MantisDriver. Owns the link, the reader, reply matching, events, - discovery and the saved configuration. + discovery and the saved configuration; reads the motors; shows + setup and stop on the LED bar. driver/simulator.py MantisSimulationDriver, MantisSimulationPlateClamp, - MantisSimulationRfid. Answer from the declared configuration; nothing - reaches a wire. + MantisSimulationPump, MantisSimulationRfid, MantisSimulationLedBar. + Answer from the declared configuration; nothing reaches a wire. driver/framing.py The frame: header, check word, splitting, payload strings, boards. driver/errors.py MantisError and one error per kind; what the device's codes mean. driver/configuration.py DeviceConfiguration: what the device answered about itself. @@ -21,7 +22,8 @@ driver/oem_config.py The unit's own values, read from its vendor .config files, into a declared configuration. driver/confirmed_firmware_versions.py The firmware sets confirmed on a device, per generation. -driver/features/ One class per thing the device is fitted with: PlateClamp, Rfid. +driver/features/ One class per thing the device is fitted with: PlateClamp, Pump, + Rfid, LedBar. driver/recordings/ One real Mantis of each generation: its configuration and unit values. resource_model/ The parts the device is modelled as, their files and manifest. liquid_classes/ MantisLiquidClass and mantis_mapping, by viscosity band and chip type. @@ -44,26 +46,37 @@ configuration (`_request_configuration`), checks it against a declared one, then is the exception: `discover` builds it when it answers its ping on interface A, after every opening of the link. Each feature is an `Optional` attribute of the driver, `None` until `_discover_features` builds -it: `plate_clamp` is built only when board 12 answered discovery, so a v3 without it has none. A -difference between v3 and v4 is expressed there, by which feature is built, never by a -branch on the generation inside a command. +it: `plate_clamp` is built only when board 12 answered discovery, so a v3 without it has none; +`pump` only when board 10 answered. `led_bar` is gated by the generation: built only on a v4 whose +board 12 answered, never on a v3, whose install sets no LED count. A difference between v3 and v4 +is expressed there, by which feature is built, never by a branch on the generation inside a +command. **M4. A configuration holds facts only, and is data.** (STAR P3, P13.) `DeviceConfiguration` holds -what the device answered: generation, firmware versions, motor limits. `save_configuration` writes -it; `declared_configuration_json` reads one back and setup refuses if the device disagrees on a -fitted field (`FITTED_FIELDS`). Each feature gets its own configuration when it is added, holding -its own facts; the device configuration does not grow per feature. Motor limits are compared as a -build fact: every v4 is taken to share them until a second device shows otherwise. +what the device answered: generation, firmware versions, motor limits, and the pressure board's +sensor calibration (`feedback_sensor_parameters`), read at setup and never written. +`save_configuration` writes it; `declared_configuration_json` reads one back and setup refuses if +the device disagrees on a fitted field (`FITTED_FIELDS`). The calibration is neither saved nor +declared: the bottle and chip sensors' offsets are re-zeroed on the device between sessions, so +they are read at every setup. A declared file that holds one, saved by an earlier version, loads; +it is ignored. A simulated pump answers the generation's recording's. +Each feature gets its own configuration when it is added, holding its own facts; the device +configuration does not grow per feature, the calibration excepted, as discovery reads it with the +firmware versions. Motor limits are compared as a build fact: every v4 is taken to share them until +a second device shows otherwise. **M4a. A unit's own values come from its folder or its declared file, else its generation's recording.** What is set per unit and cannot be read off the device sits in the unit's vendor `Data/Device/Configs` folder (`oem_config_directory`). Each value is a block (`UNIT_BLOCKS`): where the folder and the declared file both hold it they must agree, or setup refuses listing the -differences; one source suffices; with neither, the generation's recording supplies it with a -warning, less accurate on this unit. `save_configuration` saves the unit's blocks, never a -recording's, so after one setup with the folder the saved file alone carries them. A block is -loaded only where its source applies: `plate_clamp.levels` for a clamp driven as a DC motor, -`rfid.slot_to_antenna` where `RFID.config` exists. +differences; one source suffices; with neither, the generation's recording supplies it, less +accurate on this unit. A file saved before a block or a non-fitted device field existed still loads: +each value it lacks comes from the folder, else the recording, in simulation too, with one warning +naming each and where it came from. A later block or field is optional the same way. +`save_configuration` saves the unit's blocks, never a recording's, so after one setup with the +folder the saved file alone carries them. A block is loaded only where its source applies: +`plate_clamp.levels` for a clamp driven as a DC motor, `rfid.slot_to_antenna` where `RFID.config` +exists, `led_bar.number_of_leds` where `Device.config` sets `NumberOfLed`. Which vendor values are the unit's own, from a v3 and a v4 install side by side: @@ -74,7 +87,7 @@ Which vendor values are the unit's own, from a v3 and a v4 install side by side: | absolute-encoder correction (v4) | `System/Encoder Correction/*.csv` | the unit's | | plate holder origin | `Device/Adapters/SBS Adapter.ad.txt` | the unit's | | chip changer and sequence safe Z (`ChipChanger.MovementZ`, `*.seq.txt`) | `ChipChanger.config`, sequences | unit or generation, unsettled | -| clamp levels, RFID antennas | `Device.config`, `RFID.config` | one unit seen; read as the unit's | +| clamp levels, RFID antennas, LED count (`NumberOfLed`, v4) | `Device.config`, `RFID.config` | one unit seen; read as the unit's | | deck reference points and heights, arm geometry, drop verification | `StageTuning.config`, `RobotArm.config`, `DropVerification.config` | the same on both units | | chip catalogue and liquid classes | `System/`, `Device/Others/` | the software's; liquid classes ship as code (M4c) | @@ -98,7 +111,8 @@ starts. A read that times out is resent with the same packet number, up to `atte that writes or moves is sent once (`attempts=1`), never resent. A failed read ends the reader: the command in flight and every event waiter fail with it, and every later command, including one already waiting for the lock, raises until setup runs again. `stop` fails the command in flight and -every event waiter. +every event waiter. After a failed read, `stop` (and `setup`, which runs it first) sends nothing: +it closes the dead link with one warning. **M6. The wire speaks the device's units; reads return them.** (STAR P4.) Every read returns what the device sent, decoded but not converted, until the feature that owns it knows the conversion. @@ -106,8 +120,8 @@ A feature then converts in one place, on its configuration, and speaks PyLabRobo suffix on names in default units. **M7. A range is checked before anything is sent.** (STAR P5.) A read accepts only the indices -seen on a device (`MOTORS`, the boards); `_check_index` raises -`ValueError` before a frame is built. +seen on a device (`MOTORS`, `ABSOLUTE_ENCODER_MOTORS`, the boards, `PRESSURE_CONTROLLERS`); +`_check_index`, and the pump's `_get_controller_index`, raise `ValueError` before a frame is built. **M8. A command that changes the device has one raw call and one public method.** (STAR P7, P8, P9.) The raw call is `_unchecked_fw_*` and sends once; the public method checks its arguments, then @@ -120,8 +134,8 @@ confirms what the device reports. `lock` checks the level against the declared l |---|---|---| | the device, doing work | `move_` | | | the device, a process left running | `start_`, `stop_` | | -| the device's sensors | `measure_`, `sense_` | | -| the device's memory, reading | `request_` | `request_motor_limits`, `request_name` | +| the device's sensors | `measure_`, `sense_` | `pump.measure_pressure` | +| the device's memory, reading | `request_` | `request_motor_limits`, `request_motor_status` | | the device's memory, writing | `set_` | | | the resource model, reading | `get_` | | | the resource model, writing | `update_` | | @@ -129,11 +143,11 @@ confirms what the device reports. `lock` checks the level against the declared l Commands, boards and fields are named for what they do; no vendor class, method or opcode names. **M10. What is acquired is released on every path.** A feature that turns on or holds anything -(pump, pressure controllers, clamp, a chip) turns it off or releases it in a `finally`, each step -in its own `try`, and the first fault is the one raised. `stop` fails what waits on the link, then -releases each built feature, then closes the link. A `lock` that does not report clamping -done releases the clamp before raising, and `stop` releases a clamp that does not read released. -Failure behaviour is designed before coverage. +(pump, pressure controllers, clamp, a chip) turns it off or releases it in a `finally`, each step in +its own `try`, and the first fault is the one raised. `stop` fails what waits on the link, releases +each built feature, turns the LED bar off, then closes the link. A `lock` that does not report +clamping done releases the clamp before raising, and `stop` releases a clamp that does not read +released. Failure behaviour is designed before coverage. **M11. Firmware is confirmed, never gated.** A version set outside `CONFIRMED_FIRMWARE_VERSIONS` warns once at setup; nothing refuses. A set goes in once it has been seen on a device. @@ -154,11 +168,14 @@ serials or dates; a constant is referenced by name, not repeated. **M15. The simulator answers from the declared configuration; nothing reaches a wire.** (CLARIOstar simulator.) `MantisSimulationDriver` builds no link and `_send_command` raises. Setup takes the -configuration from the file's `device` block, and the reads answer from it; a board it leaves -`None` does not answer, and the device name is not simulated. Each feature has a simulation -subclass that overrides its `_unchecked_fw_*` calls and reads with state above the wire: -`MantisSimulationPlateClamp` reads 0 until a command sets clamping done or unclamping done; -`MantisSimulationRfid` reads every slot empty. +configuration from the file's `device` block (a field it lacks filled as in M4a), and the reads +answer from it; a board it leaves `None` does not answer, and the device name is not simulated. Each +feature has a simulation subclass that overrides its `_unchecked_fw_*` calls and reads with state +above the wire: `MantisSimulationPlateClamp` reads 0 until a command sets clamping done or +unclamping done; `MantisSimulationRfid` reads every slot empty; `MantisSimulationPump` reads off, +every line at 0, with the recorded calibration; `MantisSimulationLedBar` takes every command and +waits no hold. The motors read idle at 0, as after power-on; a motor read the configuration does not +hold is not simulated. **M16. Locking is by subsystem.** (STAR P11.) One lock serializes the link today. When features add commands that the device runs together, each subsystem gets its own lock, taken in declaration @@ -172,6 +189,20 @@ master's links. Interface B carries the command frames; interface A, the RFID re builds the parts and holds the driver; a feature reaches the parts through the device, and moves keep them in step. A driver with no resource tree still drives. +**M19. The LED bar shows what the device does, never at the cost of it.** An action runs inside +`_show_action`: the bar shows its colour, the enclosing action's when it ends, the idle animation +when none is left, and `error_color` after one that failed. Each colour is an attribute of the bar +(`setup_color`, ...), checked when it is set and read when it is shown; each bar has its own. +`show_progress` lights a share of the LEDs from the left in `dispense_progress_color` and sends +nothing when that many are already lit: each frame holds the link about 62 ms on a v4, so an action +shows at most `number_of_leds` frames. Setup shows blue, then green for `SETUP_DONE_HOLD`, then +idle; `stop` turns the bar off and shows no colour first: one sent there was replaced within +milliseconds, too briefly to see. An LED bar fault only warns: it never fails or +masks the action. `show_actions=False` shows no action, only the idle animation from setup on. A +colour change costs about 0.1 s of link time (two commands), so an action that is quick on its own +shows none: the plate clamp's lock and unlock, where a colour took a release from 0.21 s to 0.34 s +on the recorded v4. + ## Open, and where this is not consistent yet 1. **`CommandRefusedError` is untested.** No reply seen has a non-zero status. @@ -181,7 +212,8 @@ keep them in step. A driver with no resource tree still drives. (10 ms) and the reply timeout (5 s). A frame whose reserved word is not zero is logged, not refused. 4. **The arm and Z column are not modelled yet.** They come with the arm feature, placed from a - pose read from the device; which motor drives which link is not measured. + pose read from the device; which motor drives which link is not measured. Until then the motor + reads sit on the driver and return the device's units. 5. **Whether motor limits differ between devices of one generation** is unknown; one v4 has been read. 6. **The plate clamp and the RFID reader have run on one v4 only,** on macOS and on Windows. No v3 @@ -206,3 +238,23 @@ keep them in step. A driver with no resource tree still drives. `RfidMultiplexerDelay` is unconfirmed. 15. **The liquid classes' units are unconfirmed.** The delays may be ms or the valve sequence's time steps, the pressures psi and the vacuum inHg; the dispensing PR settles them. +16. **Some motor read fields are named by their order only.** The three currents, and the + configuration's flag and three values, have no meaning confirmed on the wire; the + position read returns its first field, the demand position, and drops the five after it. +17. **Each arm motor's first absolute encoder read after connecting is not valid,** on both + motors of the v4. It is read again, up to `ABSOLUTE_ENCODER_READS` (5, ours); the recorded v4 + startup reads it twice. Why it is not valid is not known. +18. **The vacuum sensor's offset reads about -38.8,** against about -13.6 on the bottle and chip + sensors. Nothing uses the calibration yet. The target pressure read (command 10) is left out: + no reply to it has been recorded. +19. **A v3's pressure board is unproven.** Setup reads the calibration from any board 10 that + answers its version; one that does not answer that read leaves it empty, with a warning. +20. **The LED bar has run on one v4 only.** The `fatal_error` and `unconnected` animations have + never been sent to a device, nor yellow shown; only writes of all 20 LEDs are recorded, so + `set_color` takes one colour per LED, no fewer. +21. **The LED bar's replies are not checked.** A colour write and an animation start answer 1 + when taken; a colour write while an animation plays answers 0 and is not shown, as in the + recorded v4, so every colour write stops the animation first. +22. **The dispense progress has no caller yet.** `_show_fill` (`show_progress`), `_follow_fill` and + `_show_dispense_done` wait for the dispense; `_follow_fill` reads `_sequence_progress`, which + the sequencer sets. diff --git a/pylabrobot/formulatrix/mantis/device.py b/pylabrobot/formulatrix/mantis/device.py index 6af1ef1ff11..6aee57c31d9 100644 --- a/pylabrobot/formulatrix/mantis/device.py +++ b/pylabrobot/formulatrix/mantis/device.py @@ -8,7 +8,9 @@ from pylabrobot.resources.resource_holder import ResourceHolder from .driver.configuration import RECORDING_MANTIS_V4 +from .driver.features.led_bar import LedBar from .driver.features.plate_clamp import PlateClamp +from .driver.features.pump import Pump from .driver.features.rfid import Rfid from .driver.master import MantisDriver from .driver.simulator import MantisSimulationDriver @@ -41,6 +43,7 @@ def __init__( declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, ftdi_library: Optional[str] = None, + show_actions: bool = True, simulation: bool = False, ): """ @@ -59,6 +62,8 @@ def __init__( declared file both hold a unit value, they must agree. ftdi_library: "libftdi" or "d2xx"; None picks D2XX on Windows when FTDI's D2XX library loads, else libftdi. + show_actions: whether the v4's LED bar shows each action's colour, and the idle animation + between actions. simulation: build a simulated Mantis, with no link. Raises: @@ -83,6 +88,7 @@ def __init__( declared_configuration_json=declared_configuration_json or RECORDING_MANTIS_V4, name=name, oem_config_directory=oem_config_directory, + show_actions=show_actions, ) else: self.driver = MantisDriver( @@ -93,6 +99,7 @@ def __init__( declared_configuration_json=declared_configuration_json, oem_config_directory=oem_config_directory, ftdi_library=ftdi_library, + show_actions=show_actions, ) self.mesh = { "path": os.path.join(os.path.dirname(__file__), "resource_model", MANIFEST["base"]), @@ -128,11 +135,21 @@ def plate_clamp(self) -> Optional[PlateClamp]: """The plate clamp, once setup found the LC3 board; None before or without it.""" return self.driver.plate_clamp + @property + def pump(self) -> Optional[Pump]: + """The pump, once setup found the pressure board; None before or without it.""" + return self.driver.pump + @property def rfid(self) -> Optional[Rfid]: """The RFID reader, once setup found it on interface A; None before or without it.""" return self.driver.rfid + @property + def led_bar(self) -> Optional[LedBar]: + """The LED bar, once setup found board 12 on a v4; None before, without it, or on a v3.""" + return self.driver.led_bar + async def setup(self) -> None: """Connect, read the device's configuration and check it. Nothing moves.""" await self.driver.setup() @@ -154,6 +171,7 @@ def Mantis( declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, ftdi_library: Optional[str] = None, + show_actions: bool = True, simulation: bool = False, ) -> MantisDevice: """A Formulatrix Mantis, v3 or v4. @@ -172,6 +190,8 @@ def Mantis( declared file both hold a unit value, they must agree. ftdi_library: "libftdi" or "d2xx"; None picks D2XX on Windows when FTDI's D2XX library loads, else libftdi. + show_actions: whether the v4's LED bar shows each action's colour, and the idle animation + between actions. simulation: build a simulated Mantis, with no link. Raises: @@ -189,5 +209,6 @@ def Mantis( declared_configuration_json=declared_configuration_json, oem_config_directory=oem_config_directory, ftdi_library=ftdi_library, + show_actions=show_actions, simulation=simulation, ) diff --git a/pylabrobot/formulatrix/mantis/driver/__init__.py b/pylabrobot/formulatrix/mantis/driver/__init__.py index 41241350e27..2c3e5f3bc68 100644 --- a/pylabrobot/formulatrix/mantis/driver/__init__.py +++ b/pylabrobot/formulatrix/mantis/driver/__init__.py @@ -23,7 +23,20 @@ MantisTimeoutError, PlateClampError, ) +from .features.led_bar import LedBar from .features.plate_clamp import PlateClamp +from .features.pump import ( + PRESSURE_CONTROLLER_STATUS_DESCRIPTIONS, + PRESSURE_CONTROLLERS, + PressureController, + Pump, +) from .features.rfid import Rfid from .master import MantisDriver -from .simulator import MantisSimulationDriver, MantisSimulationPlateClamp, MantisSimulationRfid +from .simulator import ( + MantisSimulationDriver, + MantisSimulationLedBar, + MantisSimulationPlateClamp, + MantisSimulationPump, + MantisSimulationRfid, +) diff --git a/pylabrobot/formulatrix/mantis/driver/configuration.py b/pylabrobot/formulatrix/mantis/driver/configuration.py index 00a36ccd576..bf68f8c759a 100644 --- a/pylabrobot/formulatrix/mantis/driver/configuration.py +++ b/pylabrobot/formulatrix/mantis/driver/configuration.py @@ -1,10 +1,12 @@ """Mantis configuration: what the device reported about itself at setup.""" +import copy import dataclasses import json import os -from typing import Any, Dict, Literal, Optional, Tuple +from typing import Any, Dict, List, Literal, Optional, Tuple +from .features.pump import PressureController from .framing import ABSOLUTE_POSITION_SENSOR_BOARD @@ -17,18 +19,24 @@ class DeviceConfiguration: firmware_versions: each board's firmware version, by board; None for a board that did not answer. motor_limits: each motor's (lower, upper) limit, by motor, in the device's units. + feedback_sensor_parameters: each pressure controller's feedback sensor (scale, offset), by + controller name (``PRESSURE_CONTROLLERS``), not by its index as the motors are; as the + pressure board holds them, empty without one. Read at every setup, never saved. """ generation: Optional[Literal["v3", "v4"]] = None firmware_versions: Dict[int, Optional[str]] = dataclasses.field(default_factory=dict) motor_limits: Dict[int, Tuple[float, float]] = dataclasses.field(default_factory=dict) + feedback_sensor_parameters: Dict[PressureController, Tuple[float, float]] = dataclasses.field( + default_factory=dict + ) # One real unit of each generation: its device block and its unit blocks. RECORDING_MANTIS_V3 = os.path.join(os.path.dirname(__file__), "recordings", "mantis_v3.json") RECORDING_MANTIS_V4 = os.path.join(os.path.dirname(__file__), "recordings", "mantis_v4.json") # The blocks a unit's own values fill: from its vendor folder, its declared file or a recording. -UNIT_BLOCKS = ("plate_clamp", "rfid") +UNIT_BLOCKS = ("plate_clamp", "rfid", "led_bar") # What is fitted: compared between a declared configuration and the device at setup. Firmware is # not, so a configuration taken off one device still describes another of its build. @@ -48,12 +56,15 @@ def compute_generation(*, firmware_versions: Dict[int, Optional[str]]) -> Litera def configuration_to_json(configuration: DeviceConfiguration) -> Dict[str, Any]: - """The fields as JSON values, as saved.""" - return dict(json.loads(json.dumps(dataclasses.asdict(configuration)))) + """The fields as JSON values, as saved: the sensor calibration is not.""" + fields = dict(json.loads(json.dumps(dataclasses.asdict(configuration)))) + del fields["feedback_sensor_parameters"] + return fields def configuration_from_json(fields: Dict[str, Any]) -> DeviceConfiguration: - """A configuration from saved fields, with JSON's lists and string keys turned back.""" + """A configuration from saved fields, with JSON's lists and string keys turned back; a sensor + calibration a file holds is not taken.""" return DeviceConfiguration( generation=fields.get("generation"), firmware_versions={ @@ -66,6 +77,43 @@ def configuration_from_json(fields: Dict[str, Any]) -> DeviceConfiguration: ) +def _fill_declared_configuration( + *, declared: Dict[str, Any], sources: List[Tuple[str, Dict[str, Any]]] +) -> Tuple[Dict[str, Any], Dict[str, str]]: + """``declared`` with each unit block and non-fitted device field it lacks from the first source + holding it; a device field only within a declared ``device`` block. + + Args: + declared: a declared configuration, as read. + sources: (where it comes from, a configuration), in the order they are preferred. + + Returns: + the filled configuration, and where each value it lacked came from, by key ("led_bar", + "device.firmware_versions"). + """ + filled = copy.deepcopy(declared) + taken: Dict[str, str] = {} + keys = [(block,) for block in UNIT_BLOCKS] + [ + ("device", field.name) + for field in dataclasses.fields(DeviceConfiguration) + if field.name not in FITTED_FIELDS + ("feedback_sensor_parameters",) + ] + for *parents, key in keys: + target = filled + for parent in parents: + target = target.get(parent, {}) + if key in target or (parents and not target): + continue + for where, source in sources: + for parent in parents: + source = source.get(parent, {}) + if key in source: + target[key] = copy.deepcopy(source[key]) + taken[".".join([*parents, key])] = where + break + return filled, taken + + def read_configuration(path: str) -> Dict[str, Any]: """Read a configuration written by ``MantisDriver.save_configuration``.""" with open(path, encoding="utf-8") as f: diff --git a/pylabrobot/formulatrix/mantis/driver/features/led_bar.py b/pylabrobot/formulatrix/mantis/driver/features/led_bar.py new file mode 100644 index 00000000000..cfad0e337a9 --- /dev/null +++ b/pylabrobot/formulatrix/mantis/driver/features/led_bar.py @@ -0,0 +1,354 @@ +"""The LED bar: the v4's status light on the LC3 board (board 12), one RGB565 colour per LED. + +It shows a static colour on each LED, or plays an animation the firmware holds. While an action +runs, the bar shows its colour (the bar's ``*_color`` attributes); between actions, the idle +animation. +""" + +import asyncio +import logging +import struct +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, AsyncIterator, Dict, List, Optional, Sequence, Tuple, Union + +from ..framing import LC3_BOARD + +if TYPE_CHECKING: + from ..master import MantisDriver + +logger = logging.getLogger(__name__) + +# The firmware's animations: name -> (first frame, last frame), played repeating. "wave" is the +# idle animation. +EFFECTS: Dict[str, Tuple[int, int]] = { + "wave": (0, 86), + "fatal_error": (800, 802), + "unconnected": (950, 968), +} +# The RGB565 value of each named colour. +LED_COLOR_CODES: Dict[str, int] = { + "off": 0x0000, + "blue": 0x001F, + "green": 0x07E0, + "yellow": 0xFFE0, + "white": 0xFFFF, + "red": 0xF800, + "magenta": 0xF81F, +} +# A colour: an RGB565 value, a name of ``LED_COLOR_CODES`` or a 24-bit web hex ("#FF0080"). +Color = Union[int, str] +# How long the bar shows ``setup_done_color`` before the idle animation, in s. +SETUP_DONE_HOLD = 1.0 +# The end of a dispense blinks the full progress bar this many times, this long off and on, in s. +DISPENSE_DONE_BLINKS = 3 +DISPENSE_DONE_BLINK = 0.35 +# How often the progress bar looks at the sequencer's progress, in s. +FILL_INTERVAL = 0.2 +# Sequencer item ids run 1 to SEQUENCE_ID_COUNT and wrap. +SEQUENCE_ID_COUNT = 32767 + + +def _compute_rgb565(*, color: Color) -> int: + """One colour's RGB565 value: the top 5, 6 and 5 bits of a web hex's R, G and B. + + Args: + color: an RGB565 value, a name of ``LED_COLOR_CODES`` or a web hex ("#FF0080" or "FF0080"). + + Raises: + ValueError: if ``color`` is none of these. + """ + if isinstance(color, int): + if not 0 <= color <= 0xFFFF: + raise ValueError(f"an RGB565 value is 0 to 0xFFFF, got {color:#x}") + return color + if color in LED_COLOR_CODES: + return LED_COLOR_CODES[color] + digits = color[1:] if color.startswith("#") else color + if len(digits) != 6 or any(digit not in "0123456789abcdefABCDEF" for digit in digits): + raise ValueError( + f"{color!r} is neither a colour of {sorted(LED_COLOR_CODES)} nor a web hex like '#FF0080'" + ) + value = int(digits, 16) + red, green, blue = (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF + return ((red & 0xF8) << 8) | ((green & 0xFC) << 3) | (blue >> 3) + + +class LedBar: + """The v4's LED bar: ``number_of_leds`` LEDs, each showing one RGB565 colour, or an animation. + + Each ``*_color`` attribute is the colour shown for one action, checked when it is set; each bar + has its own. + """ + + # Where the ``*_color`` properties start. + _setup_color: Color = "blue" + _setup_done_color: Color = "green" + _error_color: Color = "red" + _dispense_progress_color: Color = "green" + + def __init__(self, *, driver: "MantisDriver", number_of_leds: int, show_actions: bool): + """ + Args: + driver: the driver it sends through. + number_of_leds: the unit's LED count, as declared. + show_actions: whether actions show their colour, and the idle animation between them. + """ + self._driver = driver + self.number_of_leds = number_of_leds + self.show_actions = show_actions + # The actions running, outermost first: (action, whether actions inside it change the bar). + self._running_actions: List[Tuple[str, bool]] = [] + # The LEDs ``show_progress`` lit last, while the bar still shows them. + self._progress_shown: Optional[int] = None + + @property + def setup_color(self) -> Color: + """The colour shown while setup runs.""" + return self._setup_color + + @setup_color.setter + def setup_color(self, value: Color) -> None: + _compute_rgb565(color=value) + self._setup_color = value + + @property + def setup_done_color(self) -> Color: + """The colour shown for ``SETUP_DONE_HOLD`` after a setup.""" + return self._setup_done_color + + @setup_done_color.setter + def setup_done_color(self, value: Color) -> None: + _compute_rgb565(color=value) + self._setup_done_color = value + + @property + def error_color(self) -> Color: + """The colour shown after an action that failed, until the next action starts.""" + return self._error_color + + @error_color.setter + def error_color(self, value: Color) -> None: + _compute_rgb565(color=value) + self._error_color = value + + @property + def dispense_progress_color(self) -> Color: + """The colour of the LEDs ``show_progress`` lights.""" + return self._dispense_progress_color + + @dispense_progress_color.setter + def dispense_progress_color(self, value: Color) -> None: + _compute_rgb565(color=value) + self._dispense_progress_color = value + + async def _unchecked_fw_stop_animator(self) -> None: + """StopAnimator (board 12, command 16) on strip 0, sent once.""" + await self._driver._send_command( + board=LC3_BOARD, command=16, payload=struct.pack(" None: + """StartAnimator (board 12, command 17) on strip 0, repeating, sent once. + + Args: + lowest_index: the first frame. + highest_index: the last frame. + """ + await self._driver._send_command( + board=LC3_BOARD, + command=17, + payload=struct.pack(" None: + """SetRgbLed (board 12, command 13) on strip 0, sent once; not shown while an animation plays. + + Args: + colors: one RGB565 value per LED. + """ + await self._driver._send_command( + board=LC3_BOARD, + command=13, + payload=struct.pack(f" None: + """Play one of ``EFFECTS``, repeating, until a colour is set or the animation is stopped. + + Args: + effect: a key of ``EFFECTS``. + + Raises: + ValueError: if ``effect`` is not in ``EFFECTS``; nothing is sent. + """ + if effect not in EFFECTS: + raise ValueError(f"effect must be one of {sorted(EFFECTS)}, got {effect!r}") + lowest_index, highest_index = EFFECTS[effect] + self._progress_shown = None + await self._unchecked_fw_start_animator(lowest_index=lowest_index, highest_index=highest_index) + + async def stop_animation(self) -> None: + """Stop the animation; the bar holds what it shows.""" + await self._unchecked_fw_stop_animator() + + async def set_color(self, *, color: Union[Color, Sequence[Color]]) -> None: + """Show one colour on every LED, or one colour per LED; any animation is stopped first. + + Args: + color: one ``Color`` for every LED, or ``number_of_leds`` of them, one per LED. + + Raises: + ValueError: if a colour is not a ``Color``, or a sequence does not hold ``number_of_leds`` + colours; nothing is sent. + """ + if isinstance(color, (int, str)): + colors = [_compute_rgb565(color=color)] * self.number_of_leds + else: + colors = [_compute_rgb565(color=one) for one in color] + if len(colors) != self.number_of_leds: + raise ValueError(f"one colour per LED: {self.number_of_leds} colours, got {len(colors)}") + # A colour sent while an animation plays is not shown. + self._progress_shown = None + await self.stop_animation() + await self._unchecked_fw_set_rgb_led(colors=colors) + + async def show_progress(self, *, done: int, total: int) -> None: + """Light ``round(number_of_leds * done / total)`` LEDs from the left in + ``dispense_progress_color``, at least one once ``done`` > 0, the rest off; nothing is sent when + that many are already lit. + + Each frame holds the link about 62 ms (measured on a v4): show at most ``number_of_leds`` + frames per action. + + Args: + done: the steps done, 0 to ``total``. + total: the steps, at least 1. + + Raises: + ValueError: if ``total`` is below 1, or ``done`` below 0 or above ``total``; nothing is sent. + """ + if total < 1: + raise ValueError(f"total must be at least 1, got {total}") + if not 0 <= done <= total: + raise ValueError(f"done must be 0 to {total}, got {done}") + lit = round(self.number_of_leds * done / total) + if done > 0: + lit = max(lit, 1) # any progress shows: a first step of 400 still lights one LED + if lit == self._progress_shown: + return + await self.set_color( + color=[self.dispense_progress_color] * lit + ["off"] * (self.number_of_leds - lit) + ) + self._progress_shown = lit + + async def turn_off(self) -> None: + """Turn every LED off; no action shows its colour until the next one starts.""" + try: + await self.set_color(color="off") + finally: + self._running_actions.clear() + + async def _sleep(self, *, seconds: float) -> None: + """Hold the bar as it is. + + Args: + seconds: how long, in s. + """ + await asyncio.sleep(seconds) + + async def _show(self, *, key: str) -> None: + """Show an action's colour, or the idle animation for "idle"; a fault only warns. + + Args: + key: "setup", "setup_done", "error" or "idle". + """ + colors: Dict[str, Color] = { + "setup": self.setup_color, + "setup_done": self.setup_done_color, + "error": self.error_color, + } + try: + if key == "idle": + await self.start_animation(effect="wave") + else: + await self.set_color(color=colors[key]) + except Exception: + logger.warning("[%s] the LED bar did not show %r", self._driver.name, key, exc_info=True) + + async def _show_setup_done(self) -> None: + """Show ``setup_done_color`` for ``SETUP_DONE_HOLD``; leaving setup then shows idle.""" + await self._show(key="setup_done") + await self._sleep(seconds=SETUP_DONE_HOLD) + + async def _show_fill(self, *, done: int, total: int) -> None: + """``show_progress``, where a fault only warns. + + Args: + done: the steps done. + total: the steps. + """ + try: + await self.show_progress(done=done, total=total) + except Exception: + logger.warning("[%s] the LED bar did not show the progress", self._driver.name, exc_info=True) + + async def _follow_fill(self, *, well_ends: List[int], total: int) -> None: + """Fill the bar as the sequencer reports each well's last item played, until cancelled. + + Args: + well_ends: each well's last queued item so far; the dispense appends to it. + total: the wells of the dispense. + """ + shown = 0 + while True: + latest = self._driver._sequence_progress + done = 0 + if latest is not None: + # An item has played when the latest reported is at or after it, the ids wrapping. + half = SEQUENCE_ID_COUNT // 2 + done = sum((latest - end) % SEQUENCE_ID_COUNT < half for end in well_ends) + if done != shown: + await self._show_fill(done=done, total=total) + shown = done + await asyncio.sleep(FILL_INTERVAL) + + async def _show_dispense_done(self) -> None: + """The full progress bar, then ``DISPENSE_DONE_BLINKS`` blinks off and on.""" + await self._show_fill(done=1, total=1) + for _ in range(DISPENSE_DONE_BLINKS): + await self._show_fill(done=0, total=1) + await self._sleep(seconds=DISPENSE_DONE_BLINK) + await self._show_fill(done=1, total=1) + await self._sleep(seconds=DISPENSE_DONE_BLINK) + + @asynccontextmanager + async def _show_action( + self, *, action: str, show_inner_actions: bool = True, only_when_idle: bool = False + ) -> AsyncIterator[None]: + """Show ``action``'s colour while the block runs. + + Leaving it shows the enclosing action's colour, or idle when none runs. An action that fails + leaves ``error_color`` until the next action starts; a cancelled one changes nothing. + + Args: + action: a key ``_show`` takes, other than "idle". + show_inner_actions: whether actions inside the block change the bar. + only_when_idle: change the bar only when no other action runs. + """ + running = self._running_actions + if running and (only_when_idle or not running[-1][1]): + yield + return + running.append((action, show_inner_actions)) + await self._show(key=action) + try: + yield + except BaseException as error: + running.pop() + if isinstance(error, Exception) and not running: + await self._show(key="error") + raise + running.pop() + await self._show(key=running[-1][0] if running else "idle") diff --git a/pylabrobot/formulatrix/mantis/driver/features/led_bar_tests.py b/pylabrobot/formulatrix/mantis/driver/features/led_bar_tests.py new file mode 100644 index 00000000000..beaa90e1d38 --- /dev/null +++ b/pylabrobot/formulatrix/mantis/driver/features/led_bar_tests.py @@ -0,0 +1,449 @@ +"""LED bar tests: the bytes sent, against replies recorded from a Mantis v4.""" + +import asyncio +import json +import os +import tempfile +import unittest +from typing import Dict, List, Tuple, Union +from unittest import mock + +from pylabrobot.formulatrix.mantis.device import RECORDING_MANTIS_V4 +from pylabrobot.formulatrix.mantis.driver.errors import ( + MantisConfigurationError, + MantisTimeoutError, +) +from pylabrobot.formulatrix.mantis.driver.features import led_bar +from pylabrobot.formulatrix.mantis.driver.features.led_bar import ( + DISPENSE_DONE_BLINK, + LED_COLOR_CODES, + SETUP_DONE_HOLD, + LedBar, + _compute_rgb565, +) +from pylabrobot.formulatrix.mantis.driver.features.plate_clamp_tests import ( + LED_BAR_COMMANDS, + STATES_AFTER_LOCK_AT_LEVEL_2, + STATES_AFTER_RELEASE_WHILE_LOCKED, + STATES_AFTER_RELEASE_WHILE_RELEASED, + answer_states_in_turn, +) +from pylabrobot.formulatrix.mantis.driver.framing import parse_frame +from pylabrobot.formulatrix.mantis.driver.master import MantisDriver +from pylabrobot.formulatrix.mantis.driver.master_tests import DriverTestCase, FakeLink, make_answers + +OFF, BLUE, GREEN, WHITE, RED, MAGENTA = 0x0000, 0x001F, 0x07E0, 0xFFFF, 0xF800, 0xF81F +# "#00FFFF", "#8000FF" and "#FF4000"; a dim white; "#FF0080". +CYAN, VIOLET, ORANGE, DIM_WHITE, PINK = 0x07FF, 0x801F, 0xFA00, 0x4208, 0xF810 +# The colour writes the recorded v4 answered 1, shown: all 20 LEDs one colour, or 1 to 20 LEDs +# green and the rest off. +RECORDED_COLOR_WRITES: List[List[int]] = ( + [ + [color] * 20 + for color in (OFF, BLUE, GREEN, WHITE, RED, MAGENTA, CYAN, VIOLET, ORANGE, DIM_WHITE) + ] + + [[PINK] * 20] + + [[GREEN] * lit + [OFF] * (20 - lit) for lit in range(1, 20)] +) +SHOWN = "0100" +# The animation stopped, and the idle animation (frames 0 to 86, repeating) started. +STOP_ANIMATION = (12, 16, bytes.fromhex("0000")) +START_IDLE_ANIMATION = (12, 17, bytes.fromhex("0000000056000100")) +STATE_READ = (12, 33, b"") + + +def color_write(*, colors: List[int]) -> Tuple[int, int, bytes]: + """The colour write for ``colors``: strip 0, the count, then each colour, little-endian.""" + payload = "0000" + len(colors).to_bytes(2, "little").hex() + payload += "".join(color.to_bytes(2, "little").hex() for color in colors) + return (12, 13, bytes.fromhex(payload)) + + +def make_led_bar_answers() -> Dict[Tuple[int, int, bytes], List[bytes]]: + """The recorded replies, with every recorded colour write answered shown.""" + answers = make_answers() + for colors in RECORDED_COLOR_WRITES: + answers[color_write(colors=colors)] = [bytes.fromhex(SHOWN)] + return answers + + +def led_bar_frames(*, written: List[bytes]) -> List[Tuple[int, int, bytes]]: + """The LED bar's frames among ``written``.""" + frames = [parse_frame(frame=frame)[1:] for frame in written] + return [frame for frame in frames if frame[:2] in LED_BAR_COMMANDS] + + +def bar_states(*, written: List[bytes]) -> List[Union[int, str, List[int]]]: + """What the bar was told, in order: one colour, a colour per LED, or "idle".""" + states: List[Union[int, str, List[int]]] = [] + for frame in led_bar_frames(written=written): + if frame == START_IDLE_ANIMATION: + states.append("idle") + elif frame[1] == 13: + payload = frame[2] + colors = [int.from_bytes(payload[i : i + 2], "little") for i in range(4, len(payload), 2)] + states.append(colors[0] if len(set(colors)) == 1 else colors) + return states + + +class LedBarTestCase(DriverTestCase): + async def set_up_bar(self, **kwargs) -> MantisDriver: + self.link = FakeLink(answers=make_led_bar_answers()) + driver = await self.set_up_driver(link=self.link, **kwargs) + self.sent_before = len(self.link.written) + return driver + + def sent(self) -> List[Tuple[int, int, bytes]]: + return [parse_frame(frame=frame)[1:] for frame in self.link.written[self.sent_before :]] + + def states(self) -> List[Union[int, str, List[int]]]: + return bar_states(written=self.link.written[self.sent_before :]) + + +class ComputeRgb565Tests(unittest.TestCase): + def test_web_hex_gives_each_named_colour(self): + for web, name in ( + ("#000000", "off"), + ("#0000FF", "blue"), + ("#00FF00", "green"), + ("#FFFF00", "yellow"), + ("#FFFFFF", "white"), + ("#FF0000", "red"), + ("#FF00FF", "magenta"), + ): + with self.subTest(web=web): + self.assertEqual(_compute_rgb565(color=web), LED_COLOR_CODES[name]) + + def test_other_forms(self): + self.assertEqual(_compute_rgb565(color="FF00FF"), MAGENTA) + self.assertEqual(_compute_rgb565(color=0x001F), BLUE) + self.assertEqual(_compute_rgb565(color="#FF7A00"), 0xFBC0) + self.assertEqual(_compute_rgb565(color="#404040"), DIM_WHITE) + + def test_refuses_what_is_no_colour(self): + for color in ("purple", "#ZZZZZZ", "#FF00F", 0x10000, -1): + with self.subTest(color=color): + with self.assertRaises(ValueError): + _compute_rgb565(color=color) + + +class LedBarTests(LedBarTestCase): + async def asyncSetUp(self): + driver = await self.set_up_bar() + assert driver.led_bar is not None + self.bar = driver.led_bar + + async def test_set_color_stops_the_animation_then_writes_every_led(self): + await self.bar.set_color(color="magenta") + self.assertEqual(self.sent(), [STOP_ANIMATION, color_write(colors=[MAGENTA] * 20)]) + + async def test_set_color_takes_a_web_hex_and_an_rgb565_value(self): + for color in ("#FF00FF", "FF00FF", MAGENTA): + with self.subTest(color=color): + self.sent_before = len(self.link.written) + await self.bar.set_color(color=color) + self.assertEqual(self.sent(), [STOP_ANIMATION, color_write(colors=[MAGENTA] * 20)]) + + async def test_set_color_one_per_led(self): + await self.bar.set_color(color=["green"] * 10 + ["#000000"] * 9 + [OFF]) + self.assertEqual(self.sent(), [STOP_ANIMATION, color_write(colors=[GREEN] * 10 + [OFF] * 10)]) + + async def test_set_color_refuses_and_sends_nothing(self): + for color in ("purple", 0x10000, ["green"] * 19, ["green"] * 21): + with self.subTest(color=color): + with self.assertRaises(ValueError): + await self.bar.set_color(color=color) + self.assertEqual(self.sent(), []) + + async def test_turn_off(self): + await self.bar.turn_off() + self.assertEqual(self.sent(), [STOP_ANIMATION, color_write(colors=[OFF] * 20)]) + + async def test_start_and_stop_animation(self): + await self.bar.start_animation() + await self.bar.stop_animation() + self.assertEqual(self.sent(), [START_IDLE_ANIMATION, STOP_ANIMATION]) + + async def test_start_animation_refuses_an_unknown_effect(self): + with self.assertRaises(ValueError): + await self.bar.start_animation(effect="strobe") + self.assertEqual(self.sent(), []) + + async def test_fatal_error_and_unconnected_animations_sent(self): + # Never answered on a device: only the frames sent are pinned. + for effect in ("fatal_error", "unconnected"): + with self.assertRaises(MantisTimeoutError): + await self.bar.start_animation(effect=effect) + self.assertEqual( + self.sent(), + [ + (12, 17, bytes.fromhex("0000200322030100")), + (12, 17, bytes.fromhex("0000b603c8030100")), + ], + ) + + +class LedBarDiscoveryTests(LedBarTestCase): + async def test_built_only_on_a_v4_whose_board_12_answers(self): + for absent_boards, built in (((), True), ((11,), False), ((12,), False)): + with self.subTest(absent_boards=absent_boards): + driver = await self.set_up_driver( + link=FakeLink(answers=make_answers(absent_boards=absent_boards)) + ) + self.assertEqual(driver.led_bar is not None, built) + + async def test_the_led_count_comes_from_the_unit_block(self): + driver = await self.set_up_bar(declared_configuration_json=RECORDING_MANTIS_V4) + assert driver.led_bar is not None + self.assertEqual(driver.led_bar.number_of_leds, 20) + + async def test_a_malformed_led_bar_block_refuses_setup(self): + with open(RECORDING_MANTIS_V4, encoding="utf-8") as f: + recording = json.load(f) + for block in ({"number_of_leds": "twenty"}, {"number_of_leds": 0}, {}): + with self.subTest(block=block), tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "declared.json") + with open(path, "w", encoding="utf-8") as f: + json.dump({**recording, "led_bar": block}, f) + driver = self.make_driver( + link=FakeLink(answers=make_led_bar_answers()), declared_configuration_json=path + ) + # every board answers; a slow test runner must not time one out + driver.default_read_timeout = 5.0 + with self.assertRaisesRegex(MantisConfigurationError, "malformed led_bar block"): + await driver.setup() + + async def test_a_v3_setup_and_stop_send_no_led_bar_frame(self): + link = FakeLink(answers=make_answers(absent_boards=(11,))) + driver = await self.set_up_driver(link=link) + await driver.stop() + self.assertEqual(led_bar_frames(written=link.written), []) + + +class SetupAndStopTests(LedBarTestCase): + async def test_setup_shows_blue_then_green_held_then_idle(self): + link = FakeLink(answers=make_led_bar_answers()) + driver = await self.set_up_driver(link=link) + self.assertEqual(bar_states(written=link.written), [BLUE, GREEN, "idle"]) + self.assertEqual( + led_bar_frames(written=link.written), + [ + STOP_ANIMATION, + color_write(colors=[BLUE] * 20), + STOP_ANIMATION, + color_write(colors=[GREEN] * 20), + START_IDLE_ANIMATION, + ], + ) + self.led_bar_sleep.assert_awaited_once_with(driver.led_bar, seconds=SETUP_DONE_HOLD) + + async def test_setup_without_show_actions_plays_the_idle_animation_only(self): + link = FakeLink(answers=make_led_bar_answers()) + await self.set_up_driver(link=link, show_actions=False) + self.assertEqual(led_bar_frames(written=link.written), [START_IDLE_ANIMATION]) + self.led_bar_sleep.assert_not_awaited() + + async def test_stop_releases_the_clamp_then_turns_off(self): + driver = await self.set_up_bar() + await driver.stop() + self.assertEqual(self.sent(), [STATE_READ, STOP_ANIMATION, color_write(colors=[OFF] * 20)]) + + async def test_stop_without_show_actions_turns_off_only(self): + driver = await self.set_up_bar(show_actions=False) + await driver.stop() + self.assertEqual( + led_bar_frames(written=self.link.written[self.sent_before :]), + [STOP_ANIMATION, color_write(colors=[OFF] * 20)], + ) + + async def test_a_turn_off_that_fails_still_lets_the_next_setup_show(self): + driver = await self.set_up_bar() + del self.link.answers[color_write(colors=[OFF] * 20)] + with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="WARNING"): + await driver.stop() + self.sent_before = len(self.link.written) + await driver.setup() + self.assertEqual(self.states(), [BLUE, GREEN, "idle"]) + + async def test_clamp_lock_and_unlock_show_nothing(self): + driver = await self.set_up_bar(declared_configuration_json=RECORDING_MANTIS_V4) + assert driver.plate_clamp is not None + driver.default_read_timeout = 5.0 + answer_states_in_turn( + link=self.link, + turns=[ + (31, STATES_AFTER_RELEASE_WHILE_RELEASED), + (30, STATES_AFTER_LOCK_AT_LEVEL_2), + (31, STATES_AFTER_RELEASE_WHILE_LOCKED), + ], + ) + await driver.plate_clamp.lock(level=2) + await driver.plate_clamp.unlock() + self.assertEqual(led_bar_frames(written=self.link.written[self.sent_before :]), []) + + +class ActionTests(LedBarTestCase): + async def asyncSetUp(self): + self.driver = await self.set_up_bar() + assert self.driver.led_bar is not None + self.bar = self.driver.led_bar + + async def test_a_nested_action_restores_the_enclosing_colour_then_idle(self): + async with self.driver._show_action(action="setup"): + async with self.driver._show_action(action="setup_done"): + async with self.driver._show_action(action="error"): + pass + self.assertEqual(self.states(), [BLUE, GREEN, RED, GREEN, BLUE, "idle"]) + + async def test_nothing_inside_setup_changes_the_bar(self): + async with self.driver._show_action(action="setup", show_inner_actions=False): + async with self.driver._show_action(action="error"): + pass + self.assertEqual(self.states(), [BLUE, "idle"]) + + async def test_an_action_only_when_idle_shows_only_when_nothing_else_runs(self): + async with self.driver._show_action(action="setup"): + async with self.driver._show_action(action="setup_done", only_when_idle=True): + pass + async with self.driver._show_action(action="setup_done", only_when_idle=True): + pass + self.assertEqual(self.states(), [BLUE, "idle", GREEN, "idle"]) + + async def test_a_failed_action_leaves_red_until_the_next_action(self): + with self.assertRaises(ValueError): + async with self.driver._show_action(action="setup"): + async with self.driver._show_action(action="setup_done"): + raise ValueError("no flow") + self.assertEqual(self.states(), [BLUE, GREEN, RED]) + async with self.driver._show_action(action="setup"): + pass + self.assertEqual(self.states()[3:], [BLUE, "idle"]) + + async def test_a_cancelled_action_writes_nothing_more(self): + with self.assertRaises(asyncio.CancelledError): + async with self.driver._show_action(action="setup"): + raise asyncio.CancelledError() + self.assertEqual(self.states(), [BLUE]) + self.assertEqual(self.bar._running_actions, []) + + async def test_a_led_bar_fault_neither_fails_nor_masks_the_action(self): + del self.link.answers[STOP_ANIMATION] + del self.link.answers[START_IDLE_ANIMATION] + with self.assertLogs(led_bar.__name__, level="WARNING"): + async with self.driver._show_action(action="setup"): + pass + with self.assertRaisesRegex(ValueError, "the action's own"): + async with self.driver._show_action(action="setup"): + raise ValueError("the action's own error") + + async def test_show_actions_off_sends_nothing(self): + self.bar.show_actions = False + async with self.driver._show_action(action="setup"): + pass + self.assertEqual(self.sent(), []) + + async def test_each_action_shows_its_colour_attribute_when_it_is_shown(self): + with self.assertRaises(ValueError): + async with self.driver._show_action(action="setup"): + self.bar.setup_color = "magenta" + self.bar.error_color = "#FF4000" + async with self.driver._show_action(action="setup_done"): + pass + raise ValueError("no flow") + self.assertEqual(self.states(), [BLUE, GREEN, MAGENTA, ORANGE]) + + async def test_a_colour_attribute_refuses_what_is_no_colour_and_keeps_its_own(self): + for color in ("purple", "#ZZZZZZ", "#FF00F", 0x10000, -1): + with self.subTest(color=color): + with self.assertRaises(ValueError): + self.bar.setup_color = color + with self.assertRaises(ValueError): + self.bar.setup_done_color = color + with self.assertRaises(ValueError): + self.bar.error_color = color + with self.assertRaises(ValueError): + self.bar.dispense_progress_color = color + self.assertEqual( + [ + self.bar.setup_color, + self.bar.setup_done_color, + self.bar.error_color, + self.bar.dispense_progress_color, + ], + ["blue", "green", "red", "green"], + ) + self.assertEqual(self.sent(), []) + + async def test_each_bar_has_its_own_colours(self): + other = LedBar(driver=self.driver, number_of_leds=20, show_actions=True) + self.bar.setup_color = "FF00FF" + self.bar.dispense_progress_color = MAGENTA + self.assertEqual([self.bar.setup_color, self.bar.dispense_progress_color], ["FF00FF", MAGENTA]) + self.assertEqual([other.setup_color, other.dispense_progress_color], ["blue", "green"]) + + async def test_show_progress_lights_a_rounded_share_from_the_left(self): + for done, total in ((1, 3), (1, 2), (2, 3), (3, 3), (0, 3)): + await self.bar.show_progress(done=done, total=total) + self.assertEqual( + self.states(), + [ + [GREEN] * 7 + [OFF] * 13, + [GREEN] * 10 + [OFF] * 10, + [GREEN] * 13 + [OFF] * 7, + GREEN, + OFF, + ], + ) + self.assertEqual( + self.sent()[:2], [STOP_ANIMATION, color_write(colors=[GREEN] * 7 + [OFF] * 13)] + ) + + async def test_show_progress_lights_one_led_once_progress_starts(self): + await self.bar.show_progress(done=1, total=400) + self.assertEqual(self.states(), [[GREEN] + [OFF] * 19]) + + async def test_show_progress_sends_nothing_for_the_leds_already_lit(self): + await self.bar.show_progress(done=1, total=3) + await self.bar.show_progress(done=2, total=6) + self.assertEqual(len(self.sent()), 2) + await self.bar.set_color(color="blue") + await self.bar.show_progress(done=2, total=6) + self.assertEqual(self.states(), [[GREEN] * 7 + [OFF] * 13, BLUE, [GREEN] * 7 + [OFF] * 13]) + + async def test_show_progress_refuses_and_sends_nothing(self): + for done, total in ((0, 0), (-1, 3), (4, 3)): + with self.subTest(done=done, total=total): + with self.assertRaises(ValueError): + await self.bar.show_progress(done=done, total=total) + self.assertEqual(self.sent(), []) + + async def test_a_fill_lights_a_share_of_the_leds(self): + await self.bar._show_fill(done=1, total=3) + self.assertEqual(self.states(), [[GREEN] * 7 + [OFF] * 13]) + + async def test_a_fill_fault_only_warns(self): + del self.link.answers[STOP_ANIMATION] + with self.assertLogs(led_bar.__name__, level="WARNING"): + await self.bar._show_fill(done=1, total=3) + + async def test_the_fill_follows_the_played_wells_across_the_id_wrap(self): + with mock.patch.object(led_bar, "FILL_INTERVAL", 0.01): + follow = asyncio.create_task(self.bar._follow_fill(well_ends=[32766, 2], total=2)) + for progress in (32765, 32767, 3): + self.driver._sequence_progress = progress + await asyncio.sleep(0.1) + follow.cancel() + await asyncio.gather(follow, return_exceptions=True) + self.assertEqual(self.states(), [[GREEN] * 10 + [OFF] * 10, GREEN]) + + async def test_the_dispense_done_blinks_the_full_bar(self): + await self.bar._show_dispense_done() + self.assertEqual(self.states(), [GREEN] + [OFF, GREEN] * 3) + self.assertEqual( + self.led_bar_sleep.await_args_list[-6:], + [mock.call(self.bar, seconds=DISPENSE_DONE_BLINK)] * 6, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/formulatrix/mantis/driver/features/plate_clamp_tests.py b/pylabrobot/formulatrix/mantis/driver/features/plate_clamp_tests.py index 580441d7c67..b065a6ee130 100644 --- a/pylabrobot/formulatrix/mantis/driver/features/plate_clamp_tests.py +++ b/pylabrobot/formulatrix/mantis/driver/features/plate_clamp_tests.py @@ -16,6 +16,8 @@ RELEASE = (12, 31, bytes.fromhex("000000000000e83f000000000000000088130000")) LOCK_AT_LEVEL_2 = (12, 30, bytes.fromhex("0000000000000000e17a14ae47e1ca3fe8030000")) STATE_READ = (12, 33, b"") +# The LED bar's commands on board 12: a colour, the animation stopped, the animation started. +LED_BAR_COMMANDS = ((12, 13), (12, 16), (12, 17)) # The states the device read after each command, as recorded. STATES_AFTER_RELEASE_WHILE_RELEASED = ["0600", "0600"] STATES_AFTER_RELEASE_WHILE_LOCKED = ["0500", "0600"] @@ -53,7 +55,9 @@ async def asyncSetUp(self): self.sent_before = len(self.link.written) def sent(self) -> List[Tuple[int, int, bytes]]: - return [parse_frame(frame=frame)[1:] for frame in self.link.written[self.sent_before :]] + """The clamp's frames since setup; the LED bar's are pinned in its own tests.""" + frames = [parse_frame(frame=frame)[1:] for frame in self.link.written[self.sent_before :]] + return [frame for frame in frames if frame[:2] not in LED_BAR_COMMANDS] async def test_unlock(self): answer_states_in_turn(link=self.link, turns=[(31, STATES_AFTER_RELEASE_WHILE_LOCKED)]) diff --git a/pylabrobot/formulatrix/mantis/driver/features/pump.py b/pylabrobot/formulatrix/mantis/driver/features/pump.py new file mode 100644 index 00000000000..9b1e18f8642 --- /dev/null +++ b/pylabrobot/formulatrix/mantis/driver/features/pump.py @@ -0,0 +1,108 @@ +"""The pump: the pressure board (board 10), its air pump and its three pressure controllers. + +Each controller regulates one line, bottle, chip or vacuum, from its feedback sensor. Everything +here reads; nothing is written, the sensor calibration included. +""" + +import struct +from typing import TYPE_CHECKING, Dict, Literal, Tuple + +from ..framing import PRESSURE_BOARD + +if TYPE_CHECKING: + from ..master import MantisDriver + +PressureController = Literal["bottle", "chip", "vacuum"] +# The pressure controllers by name, with their number on the wire. +PRESSURE_CONTROLLERS: Dict[PressureController, int] = {"bottle": 0, "chip": 1, "vacuum": 2} +# A controller's status, as it reads it. +PRESSURE_CONTROLLER_STATUS_DESCRIPTIONS: Dict[int, str] = { + 0: "off", + 1: "settled", + 2: "unsettled", +} + + +class Pump: + """The pump and the pressure controllers, built when the pressure board answers.""" + + def __init__(self, *, driver: "MantisDriver"): + """ + Args: + driver: the driver it sends through. + """ + self._driver = driver + + def _get_controller_index(self, *, controller: PressureController) -> int: + """The controller's number on the wire. + + Raises: + ValueError: if ``controller`` is not in ``PRESSURE_CONTROLLERS``. + """ + if controller not in PRESSURE_CONTROLLERS: + raise ValueError( + f"controller must be one of {tuple(PRESSURE_CONTROLLERS)}, got {controller!r}" + ) + return PRESSURE_CONTROLLERS[controller] + + async def _send_controller_read(self, *, command: int, controller: PressureController) -> bytes: + """Send a read of one controller and return its reply; the name is checked before sending.""" + index = self._get_controller_index(controller=controller) + return await self._driver._send_command( + board=PRESSURE_BOARD, command=command, payload=struct.pack(" float: + """The pressure a controller's feedback sensor reads: psi, inHg on the vacuum. + + Args: + controller: "bottle", "chip" or "vacuum". + """ + reply = await self._send_controller_read(command=30, controller=controller) + (pressure,) = struct.unpack_from(" Tuple[float, float]: + """A controller's feedback sensor calibration, (scale, offset), as the pressure board holds it. + + Args: + controller: "bottle", "chip" or "vacuum". + """ + reply = await self._send_controller_read(command=20, controller=controller) + scale, offset = struct.unpack_from(" int: + """A controller's status; ``PRESSURE_CONTROLLER_STATUS_DESCRIPTIONS`` says what each means. + + Args: + controller: "bottle", "chip" or "vacuum". + """ + reply = await self._send_controller_read(command=16, controller=controller) + (status,) = struct.unpack_from(" bool: + """Whether a controller regulates its line. + + Args: + controller: "bottle", "chip" or "vacuum". + """ + reply = await self._send_controller_read(command=12, controller=controller) + (enabled,) = struct.unpack_from(" bool: + """Whether the air pump runs.""" + reply = await self._driver._send_command(board=PRESSURE_BOARD, command=14, attempts=3) + (on,) = struct.unpack_from(" Dict[PressureController, Tuple[float, float]]: + """Every controller's feedback sensor calibration, by controller, read at setup.""" + return { + controller: await self.request_feedback_sensor_parameters(controller=controller) + for controller in PRESSURE_CONTROLLERS + } diff --git a/pylabrobot/formulatrix/mantis/driver/features/pump_tests.py b/pylabrobot/formulatrix/mantis/driver/features/pump_tests.py new file mode 100644 index 00000000000..de73d619dae --- /dev/null +++ b/pylabrobot/formulatrix/mantis/driver/features/pump_tests.py @@ -0,0 +1,149 @@ +"""Pump tests: the reads sent, against replies recorded from a Mantis v4.""" + +import json +import os +import tempfile +import unittest +from typing import List, Tuple + +from pylabrobot.formulatrix.mantis.device import RECORDING_MANTIS_V4 +from pylabrobot.formulatrix.mantis.driver.features.pump import PressureController +from pylabrobot.formulatrix.mantis.driver.framing import parse_frame +from pylabrobot.formulatrix.mantis.driver.master_tests import ( + RECORDED_CALIBRATION, + RECORDING, + DriverTestCase, + FakeLink, + make_answers, +) + +# The pressure board's reads, as sent. +CALIBRATION_READS = [(10, 20, bytes.fromhex(request)) for request in ("0000", "0100", "0200")] +# With the pump running: the chip controller enabled and the pump on, as recorded. +CHIP_CONTROLLER_ENABLED = (10, 12, bytes.fromhex("0100"), bytes.fromhex("0100")) +PUMP_ON = (10, 14, b"", bytes.fromhex("0100")) +# The chip and vacuum controllers' status while regulating: settled, then unsettled, as recorded. +CHIP_CONTROLLER_SETTLED = (10, 16, bytes.fromhex("0100"), bytes.fromhex("0100")) +VACUUM_CONTROLLER_UNSETTLED = (10, 16, bytes.fromhex("0200"), bytes.fromhex("0200")) + + +class PumpTests(DriverTestCase): + async def asyncSetUp(self): + self.link = FakeLink(answers=make_answers()) + self.driver = await self.set_up_driver(link=self.link) + assert self.driver.pump is not None + self.pump = self.driver.pump + self.sent_before = len(self.link.written) + + def sent(self) -> List[Tuple[int, int, bytes]]: + return [parse_frame(frame=frame)[1:] for frame in self.link.written[self.sent_before :]] + + async def test_setup_reads_the_calibration_and_writes_nothing(self): + self.assertEqual(self.driver.configuration.feedback_sensor_parameters, RECORDED_CALIBRATION) + sent = [parse_frame(frame=frame)[1:] for frame in self.link.written[: self.sent_before]] + self.assertEqual([read for read in sent if read[0] == 10], [(10, 1, b"")] + CALIBRATION_READS) + + async def test_measure_pressure(self): + cases: List[Tuple[PressureController, str, float]] = [ + ("bottle", "0000", 1.1709537506103516), + ("chip", "0100", 1.148512840270996), + ("vacuum", "0200", -0.557586669921875), + ] + for controller, request, expected in cases: + with self.subTest(controller=controller): + self.assertEqual(await self.pump.measure_pressure(controller=controller), expected) + self.assertEqual(self.sent()[-1], (10, 30, bytes.fromhex(request))) + + async def test_request_feedback_sensor_parameters(self): + self.assertEqual( + await self.pump.request_feedback_sensor_parameters(controller="vacuum"), + RECORDED_CALIBRATION["vacuum"], + ) + self.assertEqual(self.sent(), [CALIBRATION_READS[2]]) + + async def test_pump_off_reads(self): + self.assertEqual(await self.pump.request_controller_status(controller="bottle"), 0) + self.assertFalse(await self.pump.request_controller_enabled(controller="bottle")) + self.assertFalse(await self.pump.request_pump_on()) + self.assertEqual(self.sent(), [(10, 16, bytes(2)), (10, 12, bytes(2)), (10, 14, b"")]) + + RUNNING_READS = ( + CHIP_CONTROLLER_ENABLED, + PUMP_ON, + CHIP_CONTROLLER_SETTLED, + VACUUM_CONTROLLER_UNSETTLED, + ) + + async def test_pump_running_reads(self): + for board, command, request, reply in self.RUNNING_READS: + self.link.answers[(board, command, request)] = [reply] + self.assertTrue(await self.pump.request_controller_enabled(controller="chip")) + self.assertTrue(await self.pump.request_pump_on()) + self.assertEqual(await self.pump.request_controller_status(controller="chip"), 1) + self.assertEqual(await self.pump.request_controller_status(controller="vacuum"), 2) + self.assertEqual( + self.sent(), + [(board, command, request) for board, command, request, _ in self.RUNNING_READS], + ) + + async def test_unknown_controller_sends_nothing(self): + for read in ( + self.pump.measure_pressure, + self.pump.request_feedback_sensor_parameters, + self.pump.request_controller_status, + self.pump.request_controller_enabled, + ): + with self.subTest(read=read.__name__): + with self.assertRaises(ValueError): + await read(controller="air") # type: ignore[arg-type] + self.assertEqual(self.sent(), []) + + +class PumpDiscoveryTests(DriverTestCase): + async def test_built_only_when_the_pressure_board_answers(self): + for absent_boards, built in (((), True), ((10,), False)): + with self.subTest(absent_boards=absent_boards): + link = FakeLink(answers=make_answers(absent_boards=absent_boards)) + driver = await self.set_up_driver(link=link) + self.assertEqual(driver.pump is not None, built) + sent = [parse_frame(frame=frame)[1:] for frame in link.written] + self.assertEqual(any(read in sent for read in CALIBRATION_READS), built) + + async def test_calibration_not_answered_leaves_it_empty(self): + answers = make_answers() + for board, command, request in CALIBRATION_READS: + del answers[(board, command, request)] + driver = await self.set_up_driver(link=FakeLink(answers=answers)) + self.assertIsNotNone(driver.pump) + self.assertEqual(driver.configuration.feedback_sensor_parameters, {}) + + async def test_declared_calibration_is_ignored_and_not_saved(self): + calibration = {"bottle": [0.01, -14.0], "chip": [0.01, -14.0], "vacuum": [0.01, -38.0]} + device = RECORDING["device"].items() + without = {field: value for field, value in device if field != "feedback_sensor_parameters"} + for device in (without, {**without, "feedback_sensor_parameters": calibration}): + with self.subTest(declared="feedback_sensor_parameters" in device): + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "declared.json") + with open(path, "w", encoding="utf-8") as f: + json.dump({**RECORDING, "device": device}, f) + with self.assertNoLogs("pylabrobot.formulatrix.mantis", level="WARNING"): + driver = await self.set_up_driver( + link=FakeLink(answers=make_answers()), declared_configuration_json=path + ) + self.assertEqual(driver.configuration.feedback_sensor_parameters, RECORDED_CALIBRATION) + saved = os.path.join(directory, "saved.json") + driver.save_configuration(path=saved) + with open(saved, encoding="utf-8") as f: + self.assertNotIn("feedback_sensor_parameters", json.load(f)["device"]) + + async def test_recording_holds_the_recorded_calibration(self): + with open(RECORDING_MANTIS_V4, encoding="utf-8") as f: + declared = json.load(f)["device"]["feedback_sensor_parameters"] + self.assertEqual( + declared, {controller: list(value) for controller, value in RECORDED_CALIBRATION.items()} + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/formulatrix/mantis/driver/master.py b/pylabrobot/formulatrix/mantis/driver/master.py index 94837d4895a..db9047c4ab9 100644 --- a/pylabrobot/formulatrix/mantis/driver/master.py +++ b/pylabrobot/formulatrix/mantis/driver/master.py @@ -1,15 +1,16 @@ """The Mantis driver: the link on the FTDI's interface B, discovery and the device reads. Every command sent here reads the device; the features in ``features/`` send the commands that -act. Reads return the device's units. The motor and pressure reads arrive with the arm and pump -features. +act, and the LED bar shows setup and stop. Reads return the device's units. The motor reads are +here until the arm feature takes them; the pressure reads are on the pump. """ import asyncio +import contextlib import json import logging import struct -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Any, AsyncContextManager, Dict, List, Optional, Set, Tuple from pylabrobot.io.ftdi import FTDI @@ -19,6 +20,7 @@ RECORDING_MANTIS_V4, UNIT_BLOCKS, DeviceConfiguration, + _fill_declared_configuration, compute_generation, configuration_to_json, read_configuration, @@ -32,7 +34,9 @@ MantisNotConnectedError, MantisTimeoutError, ) +from .features.led_bar import LedBar from .features.plate_clamp import UNCLAMPING_DONE, PlateClamp +from .features.pump import Pump from .features.rfid import RFID_BAUD_RATE, RFID_FTDI_INTERFACE, Rfid, _initialize_rfid_module from .framing import ( ABSOLUTE_POSITION_SENSOR_BOARD, @@ -50,6 +54,11 @@ logger = logging.getLogger(__name__) MOTORS = (0, 1, 2) +# The arm motors; board 11 reads their absolute encoders. Z has none. +ABSOLUTE_ENCODER_MOTORS = (0, 1) +# How often an absolute encoder is read for a valid position; each motor's first after connecting +# is not. +ABSOLUTE_ENCODER_READS = 5 class MantisDriver: @@ -89,6 +98,7 @@ def __init__( declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, ftdi_library: Optional[str] = None, + show_actions: bool = True, ): """ Args: @@ -102,10 +112,13 @@ def __init__( oem_config_directory: the unit's vendor ``Data/Device/Configs`` folder, read at every setup. ftdi_library: "libftdi" or "d2xx", for both links; None picks D2XX on Windows when FTDI's D2XX library loads, else libftdi. Ignored for a link given as ``io`` or ``rfid_io``. + show_actions: whether the v4's LED bar shows each action's colour, and the idle animation + between actions; False shows the idle animation from setup on. The unit's own values (``UNIT_BLOCKS``) come from ``oem_config_directory`` and the declared - file; where both have a block, they must agree. A block neither has comes from the - generation's recording, with a warning: another unit's values, less accurate on this one. + file; where both have a block, they must agree. A unit block or non-fitted device field the + declared file lacks comes from the folder, else the generation's recording (another unit's + values, less accurate on this one), with one warning naming each and where it came from. """ self.io: FTDI = ( io if io is not None else self._build_io(device_id=device_id, ftdi_library=ftdi_library) @@ -129,7 +142,12 @@ def __init__( else self._build_rfid_io(device_id=device_id, ftdi_library=ftdi_library) ) self.plate_clamp: Optional[PlateClamp] = None + self.pump: Optional[Pump] = None self.rfid: Optional[Rfid] = None + self.led_bar: Optional[LedBar] = None + self._show_actions = show_actions + # The latest sequencer item the device reported played; the sequencer sets it. + self._sequence_progress: Optional[int] = None self._rfid_link_open = False self._connected = False @@ -362,6 +380,128 @@ async def request_motor_limits(self, *, motor: int) -> Tuple[float, float]: lower, upper = struct.unpack_from(" Tuple[float, float, float]: + """A motor's three currents, in the order the device sends them. + + 1.5, 1.0 and 0.6 on the v4's arms. + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=12, payload=struct.pack(" Tuple[int, float, float, float, int, int]: + """A motor's configuration, its six fields as the device sends them. + + A flag (1 on motor 1, 0 on motors 0 and 2 of the v4), three values (2000.0, 200.0, 0.0 on its + arms), the microsteps (100) and a last word (1 on every motor). + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=14, payload=struct.pack(" int: + """A motor's status word; ``MOTOR_STATUS_ERROR_BITS`` names the bits that report a fault. + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=20, payload=struct.pack(" float: + """A motor's demand position, in the device's units: where the controller drives it. + + It is not measured. It reads back the last position declared to the motor until the motor + moves, wherever the motor sits; an arm moved by hand still reads it. + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=27, payload=struct.pack(" Tuple[bool, float]: + """A motor's following-error check: (enabled, the largest following error), device units. + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=82, payload=struct.pack(" float: + """A motor's servo encoder position, in the device's units. + + It counts only while the device is powered: after power-on it reads about 0 wherever the motor + sits. ``request_absolute_encoder_position`` reads where an arm is. + + Args: + motor: 0, 1 (arm) or 2 (Z). + """ + self._check_index(name="motor", value=motor, allowed=MOTORS) + reply = await self._send_command( + board=MOTION_BOARD, command=84, payload=struct.pack(" float: + """An arm motor's raw absolute encoder reading, from board 11; kept across power-off. + + Not an angle yet: the unit's encoder correction and absolute homing position make it one. A + read the device marks not valid, as each motor's first after connecting is, is read again. + + Args: + motor: 0 or 1 (arm). + + Raises: + MantisTimeoutError: if none of ``ABSOLUTE_ENCODER_READS`` reads is valid. + """ + self._check_index(name="motor", value=motor, allowed=ABSOLUTE_ENCODER_MOTORS) + for _ in range(ABSOLUTE_ENCODER_READS): + reply = await self._send_command( + board=ABSOLUTE_POSITION_SENSOR_BOARD, + command=30, + payload=struct.pack(" DeviceConfiguration: @@ -392,7 +532,7 @@ def _discover_features(self) -> None: A feature already built is kept. Raises: - MantisConfigurationError: if the ``plate_clamp`` block is malformed. + MantisConfigurationError: if the ``plate_clamp`` or ``led_bar`` block is malformed. """ if self.configuration.firmware_versions.get(LC3_BOARD) is not None and self.plate_clamp is None: levels: Optional[Dict[int, Tuple[float, int]]] = None @@ -406,6 +546,43 @@ def _discover_features(self) -> None: except (KeyError, TypeError, ValueError) as error: raise MantisConfigurationError(f"malformed plate_clamp block: {error!r}") from error self.plate_clamp = PlateClamp(driver=self, levels=levels) + if self.configuration.firmware_versions.get(PRESSURE_BOARD) is not None and self.pump is None: + self.pump = Pump(driver=self) + # The LED bar is the v4's, on board 12: a v3 has none. + if ( + self.configuration.generation == "v4" + and self.configuration.firmware_versions.get(LC3_BOARD) is not None + and self.led_bar is None + ): + led_bar_block = self._unit_blocks.get("led_bar") + if led_bar_block is None: + raise MantisConfigurationError(f"{self.name}: no led_bar block for this v4") + try: + number_of_leds = int(led_bar_block["number_of_leds"]) + except (KeyError, TypeError, ValueError) as error: + raise MantisConfigurationError(f"malformed led_bar block: {error!r}") from error + if number_of_leds < 1: + raise MantisConfigurationError(f"malformed led_bar block: {number_of_leds} LEDs") + self.led_bar = LedBar( + driver=self, number_of_leds=number_of_leds, show_actions=self._show_actions + ) + + def _show_action( + self, *, action: str, show_inner_actions: bool = True, only_when_idle: bool = False + ) -> AsyncContextManager[None]: + """Show ``action``'s colour on the LED bar while the block runs; nothing without a bar or with + its ``show_actions`` off. + + Args: + action: a key ``LedBar._show`` takes, other than "idle". + show_inner_actions: whether actions inside the block change the bar. + only_when_idle: change the bar only when no other action runs. + """ + if self.led_bar is None or not self.led_bar.show_actions: + return contextlib.nullcontext() + return self.led_bar._show_action( + action=action, show_inner_actions=show_inner_actions, only_when_idle=only_when_idle + ) def _check_declared_configuration(self) -> None: """Compare what is fitted with the declared configuration's ``device`` block, if there is one. @@ -435,33 +612,28 @@ def _check_declared_configuration(self) -> None: f"{self.name} does not match {self.declared_configuration_json}: " + "; ".join(differences) ) - async def discover(self) -> None: - """Read the configuration into ``configuration``, check it against the declared one, then - build the features it lists, and the RFID reader if it answers on interface A. + def _load_unit_values( + self, *, generation: Optional[str] + ) -> Tuple[Dict[str, Any], Dict[str, str]]: + """The declared file, or without one the folder's blocks, with each value it lacks taken from + the folder, else the generation's recording; and where each taken one came from, by key. - The reader is set up again after every opening of the link; one that fails is absent. + Args: + generation: picks the recording. Raises: - MantisConfigurationError: if the device does not match the declared configuration, the - unit's folder and declared file disagree on a block, or a block is malformed. + MantisConfigurationError: if the folder and the declared file disagree on a block. """ - self.configuration = await self._request_configuration() - self._check_declared_configuration() - # The unit's blocks: its folder and its declared file must agree; a recording fills the rest. folder = ( _load_unit_configuration(oem_config_directory=self.oem_config_directory) if self.oem_config_directory is not None else {} ) - declared = ( - read_configuration(path=self.declared_configuration_json) - if self.declared_configuration_json is not None - else {} - ) - recording_path = ( - RECORDING_MANTIS_V4 if self.configuration.generation == "v4" else RECORDING_MANTIS_V3 - ) + recording_path = RECORDING_MANTIS_V4 if generation == "v4" else RECORDING_MANTIS_V3 recording = read_configuration(path=recording_path) + if self.declared_configuration_json is None: + return _fill_declared_configuration(declared=folder, sources=[(recording_path, recording)]) + declared = read_configuration(path=self.declared_configuration_json) differences = [ f"{block}: declared {declared[block]!r}, folder {folder[block]!r}" for block in UNIT_BLOCKS @@ -474,20 +646,49 @@ async def discover(self) -> None: f"{self.oem_config_directory} and {self.declared_configuration_json} disagree: " + "; ".join(differences) ) - self._unit_blocks, self._unit_blocks_from_recording = {}, set() - for block in UNIT_BLOCKS: - if block in folder or block in declared: - self._unit_blocks[block] = folder[block] if block in folder else declared[block] - elif block in recording: - self._unit_blocks[block] = recording[block] - self._unit_blocks_from_recording.add(block) - logger.warning( - "[%s] %s: none of this unit's values given; using %s's, less accurate on this unit", - self.name, - block, - recording_path, - ) + sources = [(recording_path, recording)] + if self.oem_config_directory is not None: + sources.insert(0, (self.oem_config_directory, folder)) + return _fill_declared_configuration(declared=declared, sources=sources) + + async def discover(self) -> None: + """Read the configuration into ``configuration``, check it against the declared one, then + build the features it lists, read the pump's sensor calibration into ``configuration``, and + build the RFID reader if it answers on interface A. + + The reader is set up again after every opening of the link; one that fails is absent. A + calibration the pressure board does not answer is left empty. + + Raises: + MantisConfigurationError: if the device does not match the declared configuration, the + unit's folder and declared file disagree on a block, or a block is malformed. + """ + self.configuration = await self._request_configuration() + self._check_declared_configuration() + values, taken = self._load_unit_values(generation=self.configuration.generation) + recordings = (RECORDING_MANTIS_V3, RECORDING_MANTIS_V4) + if taken: + given = self.declared_configuration_json or self.oem_config_directory + logger.warning( + "[%s] %s: %s%s", + self.name, + f"not in {given}" if given is not None else "none given", + "; ".join(f"{key} from {where}" for key, where in taken.items()), + ", another unit's values, less accurate on this unit" + if any(where in recordings for where in taken.values()) + else "", + ) + self._unit_blocks = {block: values[block] for block in UNIT_BLOCKS if block in values} + self._unit_blocks_from_recording = { + block for block in UNIT_BLOCKS if taken.get(block) in recordings + } self._discover_features() + if self.pump is not None: + # Nothing uses the calibration yet: a board 10 that does not answer it does not fail setup. + try: + self.configuration.feedback_sensor_parameters = await self.pump._request_calibration() + except (TimeoutError, CommandRefusedError, FrameError, struct.error) as error: + logger.warning("[%s] the pressure sensors' calibration not read: %r", self.name, error) if self.rfid is None and self.rfid_io is not None and self._rfid_link_open: try: await _initialize_rfid_module(io=self.rfid_io) @@ -523,7 +724,8 @@ def _check_firmware_compatibility(self) -> None: def save_configuration(self, path: str, indent: Optional[int] = 2) -> None: """Write what this device reported to a file, to be declared later - (``declared_configuration_json``), with the unit's blocks; none taken from a recording. + (``declared_configuration_json``), with the unit's blocks; none taken from a recording. The + sensor calibration is not written: it is read at every setup. Args: path: where to write it. @@ -562,7 +764,10 @@ async def _open(self) -> None: async def _close(self) -> None: """Stop reading the link and close it, then close the RFID reader's link.""" reader, self._reader = self._reader, None - if reader is not None: + if reader is not None and reader.done(): + if not reader.cancelled(): + reader.exception() # a failed read was logged when it failed + elif reader is not None: reader.cancel() try: await reader @@ -582,8 +787,16 @@ async def _close(self) -> None: logger.warning("[%s] the RFID reader's link did not close", self.name, exc_info=True) async def stop(self) -> None: - """Fail the command in flight and every event waiter, release the plate clamp, stop reading - the link and close it. Each step runs whether or not the one before it succeeded.""" + """Fail the command in flight and every event waiter, release the plate clamp, turn the LED + bar off, stop reading the link and close it. + + Each step runs whether or not the one before it succeeded. After reading the link failed, + nothing is sent: the link is closed. + """ + if self._reader is not None and self._reader.done() and self._connected: + failure = None if self._reader.cancelled() else self._reader.exception() + logger.warning("[%s] the link was lost (%r); closing it, sending nothing", self.name, failure) + self._connected = False stopped = f"{self.name} stopped" if self._pending is not None: _, _, future = self._pending @@ -599,12 +812,19 @@ async def stop(self) -> None: await self.plate_clamp.unlock() except Exception: logger.warning("[%s] could not release the plate clamp", self.name, exc_info=True) + if self.led_bar is not None and self._connected: + try: + await self.led_bar.turn_off() + except Exception: + logger.warning("[%s] could not turn the LED bar off", self.name, exc_info=True) self._connected = False await self._close() async def setup(self) -> None: """Open the link if it is not open, read the device's configuration and check it. + On a v4 the LED bar shows the setup, then "setup_done" for ``SETUP_DONE_HOLD``, then idle. + Raises: MantisConfigurationError: if the device does not match the declared configuration. """ @@ -615,6 +835,11 @@ async def setup(self) -> None: await self._open() self._connected = True await self.discover() + if self.led_bar is not None and not self.led_bar.show_actions: + await self.led_bar._show(key="idle") # the idle animation from setup on + async with self._show_action(action="setup", show_inner_actions=False): + if self.led_bar is not None and self.led_bar.show_actions: + await self.led_bar._show_setup_done() except BaseException: try: await self.stop() diff --git a/pylabrobot/formulatrix/mantis/driver/master_tests.py b/pylabrobot/formulatrix/mantis/driver/master_tests.py index bde33dfcd37..48d176bb490 100644 --- a/pylabrobot/formulatrix/mantis/driver/master_tests.py +++ b/pylabrobot/formulatrix/mantis/driver/master_tests.py @@ -17,13 +17,15 @@ MantisNotConnectedError, MantisTimeoutError, ) +from pylabrobot.formulatrix.mantis.driver.features.led_bar import LedBar +from pylabrobot.formulatrix.mantis.driver.features.pump import PressureController from pylabrobot.formulatrix.mantis.driver.features.rfid import ( RFID_PING_DEVICE_COMMAND, RFID_READ_TAG_COMMAND, ) from pylabrobot.formulatrix.mantis.driver.framing import build_frame, parse_frame -from pylabrobot.formulatrix.mantis.driver.master import MantisDriver -from pylabrobot.io.ftdi import FTDI +from pylabrobot.formulatrix.mantis.driver.master import ABSOLUTE_ENCODER_READS, MantisDriver +from pylabrobot.io.ftdi import FTDI, D2XXError with open(RECORDING_MANTIS_V4, encoding="utf-8") as f: RECORDING = json.load(f) @@ -43,6 +45,44 @@ (0, 10, "0000"): "000000000040afc0000000000040af40", (0, 10, "0100"): "000000000040afc0000000000040af40", (0, 10, "0200"): "00000000000064c00000000000f8a140", + # The motor reads: currents, configuration, status, position, following error, encoder. + (0, 12, "0000"): "000000000000f83f000000000000f03f000000403333e33f", + (0, 12, "0200"): "000000000000f03f000000000000f03f000000606666e63f", + (0, 14, "0100"): "01000000000000409f400000000000006940000000000000000064000100", + (0, 14, "0200"): "00000000000000409f400000000000005940000000000000000064000100", + (0, 20, "0000"): "0000", + (0, 20, "0100"): "8000", + (0, 20, "0200"): "0600", + ( + 0, + 27, + "0000", + ): "00000000000000000000000000000000000000000000000000000000000000000000000000000000" + "0000000000407f40", + ( + 0, + 27, + "0200", + ): "00000000fc0c7840000000a076d385c00000004005c2bec000000000d0e74ec000000000b8eaa040" + "0000004005c2be40", + (0, 82, "0000"): "01000000000000001040", + (0, 82, "0200"): "01000000000000005940", + (0, 84, "0000"): "00000000b42350bf", + (0, 84, "0100"): "0000000090b8d73f", + # The absolute encoders, valid. + (11, 30, "0000"): "010000000000f43f6740", + (11, 30, "0100"): "010000000000a0d73940", + # The pressure board: each controller's sensor calibration, its reading and its status, and + # whether a controller is enabled and the pump on, with the pump off. + (10, 20, "0000"): "0000004049d0853f000000e06f292bc0", + (10, 20, "0100"): "00000080d1d5853f00000000b4572bc0", + (10, 20, "0200"): "000000001723853f00000020486343c0", + (10, 30, "0000"): "000000003abcf23f", + (10, 30, "0100"): "000000004f60f23f", + (10, 30, "0200"): "00000000c0d7e1bf", + (10, 16, "0000"): "0000", + (10, 12, "0000"): "0000", + (10, 14, ""): "0000", # The release and the lock at levels 1, 2 and 3. (12, 31, "000000000000e83f000000000000000088130000"): "", (12, 30, "0000000000000000105839b4c876be3fd0070000"): "", @@ -51,7 +91,22 @@ (12, 33, ""): "0600", # The RFID antenna switch to each of the six antennas. **{(0, 41, f"78000100{antenna:02x}00"): "" for antenna in (0, 1, 4, 5, 6, 7)}, + # The LED bar: the animation stopped, the idle animation, and setup's blue and green and stop's + # off on all 20 LEDs. + (12, 16, "0000"): "", + (12, 17, "0000000056000100"): "0100", + **{(12, 13, "00001400" + color * 20): "0100" for color in ("1f00", "e007", "0000")}, } +# The sensor calibration the recorded replies carry, by controller. +RECORDED_CALIBRATION: Dict[PressureController, Tuple[float, float]] = { + "bottle": (0.010651180520653725, -13.580931663513184), + "chip": (0.010661732405424118, -13.671295166015625), + "vacuum": (0.010320834815502167, -38.775638580322266), +} +# Motor 0's absolute encoder after connecting: its first read not valid, the next valid. +ABSOLUTE_ENCODER_REPLIES_AFTER_POWER_ON = ["00000000000000000000", "010000000000f43f6740"] +# The link reader's failure when the USB cable was pulled from a Mantis v4, through D2XX. +RECORDED_LINK_FAILURE = D2XXError("FT_GetQueueStatus returned FT_IO_ERROR (4)", status=4) # Interface A: what the RFID reader on a Mantis v4 sent, chunked as it arrived. READER_PING_REPLY = [b"TRF7970A EVM \r\n\r\n"] READER_EMPTY_SLOT_REPLY = [b"Request mode.\r\n", b"[]\r\n"] @@ -138,9 +193,16 @@ async def read(self, num_bytes: int = 1) -> bytes: class DriverTestCase(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # The LED bar's holds are taken, not waited: setup holds "setup_done" for a second. + patcher = mock.patch.object(LedBar, "_sleep", autospec=True) + self.led_bar_sleep = patcher.start() + self.addCleanup(patcher.stop) + def make_driver(self, *, link: FakeLink, **kwargs) -> MantisDriver: driver = MantisDriver(io=link.io, rfid_io=link.reader.io, **kwargs) - driver.default_read_timeout = 0.05 + # short, as a board that never answers waits it out; long enough for a slow test runner + driver.default_read_timeout = 0.25 return driver async def set_up_driver(self, *, link: FakeLink, **kwargs) -> MantisDriver: @@ -171,6 +233,7 @@ async def test_v4_discovery(self): generation="v4", firmware_versions={0: "4.20.6.2", 10: "1.0.8", 11: "0.1.7 DBG", 12: "0.1.8 DBG"}, motor_limits={0: (-4000.0, 4000.0), 1: (-4000.0, 4000.0), 2: (-160.0, 2300.0)}, + feedback_sensor_parameters=RECORDED_CALIBRATION, ), ) @@ -223,6 +286,44 @@ async def test_a_second_setup_opens_the_link_once(self): link.io.set_baudrate.assert_awaited_once() self.assertIs(driver._reader, reader) + async def test_setup_after_a_lost_link_closes_it_quietly_and_reopens_it(self): + link = FakeLink(answers=make_answers()) + driver = await self.set_up_driver(link=link, declared_configuration_json=RECORDING_MANTIS_V4) + read = link.io.read.side_effect + link.io.read.side_effect = RECORDED_LINK_FAILURE + with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="ERROR"): + while driver._reader is not None and not driver._reader.done(): + await asyncio.sleep(0.01) + link.io.read.side_effect = read + sent_before = len(link.written) + with self.assertLogs("pylabrobot.formulatrix.mantis", level="INFO") as logs: + await driver.setup() + warnings = [record for record in logs.records if record.levelname != "INFO"] + self.assertEqual(len(warnings), 1) + self.assertIn("the link was lost", warnings[0].getMessage()) + self.assertTrue(all(record.exc_info is None for record in logs.records)) + link.io.stop.assert_awaited_once() + self.assertEqual(link.io.setup.await_count, 2) + # The first frame after the failure is setup's own firmware read: nothing went to the dead link. + self.assertEqual(parse_frame(frame=link.written[sent_before])[1:], (0, 1, b"")) + assert driver._reader is not None + self.assertFalse(driver._reader.done()) + + async def test_an_older_declared_file_takes_what_it_lacks_from_the_recording(self): + older = {block: values for block, values in RECORDING.items() if block != "led_bar"} + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "older.json") + with open(path, "w", encoding="utf-8") as f: + json.dump(older, f) + link = FakeLink(answers=make_answers()) + with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="WARNING") as logs: + driver = await self.set_up_driver(link=link, declared_configuration_json=path) + self.assertEqual(len(logs.output), 1) + self.assertIn(f"led_bar from {RECORDING_MANTIS_V4}", logs.output[0]) + assert driver.led_bar is not None + self.assertEqual(driver.led_bar.number_of_leds, 20) + self.assertEqual(driver.configuration.feedback_sensor_parameters, RECORDED_CALIBRATION) + async def test_saved_configuration_declares_the_same_device(self): link = FakeLink(answers=make_answers()) driver = await self.set_up_driver(link=link, declared_configuration_json=RECORDING_MANTIS_V4) @@ -231,7 +332,9 @@ async def test_saved_configuration_declares_the_same_device(self): driver.save_configuration(path=path) await driver.stop() with open(path, encoding="utf-8") as f: - self.assertEqual(json.load(f)["plate_clamp"], RECORDING["plate_clamp"]) + saved = json.load(f) + for block in ("plate_clamp", "led_bar"): + self.assertEqual(saved[block], RECORDING[block]) link = FakeLink(answers=make_answers()) await self.set_up_driver(link=link, declared_configuration_json=path) @@ -318,5 +421,80 @@ async def test_unanswered_read_is_resent_with_its_packet_number_then_times_out(s self.assertNotEqual(self.link.written[-4][4:6], self.link.written[-1][4:6]) +class MotorReadTests(DriverTestCase): + async def asyncSetUp(self): + self.link = FakeLink(answers=make_answers()) + self.driver = await self.set_up_driver(link=self.link) + + async def test_reads(self): + driver = self.driver + cases = [ + ( + 0, + 12, + "0000", + lambda: driver.request_motor_currents(motor=0), + (1.5, 1.0, 0.6000000238418579), + ), + ( + 0, + 14, + "0100", + lambda: driver.request_motor_configuration(motor=1), + (1, 2000.0, 200.0, 0.0, 100, 1), + ), + (0, 20, "0100", lambda: driver.request_motor_status(motor=1), 0x80), + (0, 20, "0200", lambda: driver.request_motor_status(motor=2), 6), + (0, 27, "0000", lambda: driver.request_motor_position(motor=0), 0.0), + (0, 27, "0200", lambda: driver.request_motor_position(motor=2), 384.8115234375), + (0, 82, "0000", lambda: driver.request_following_error_configuration(motor=0), (True, 4.0)), + (0, 82, "0200", lambda: driver.request_following_error_configuration(motor=2), (True, 100.0)), + (0, 84, "0100", lambda: driver.request_encoder_position(motor=1), 0.3706398010253906), + (11, 30, "0000", lambda: driver.request_absolute_encoder_position(motor=0), 185.99853515625), + ] + for board, command, request, read, expected in cases: + with self.subTest(board=board, command=command, request=request): + result = await read() + self.assertEqual( + parse_frame(frame=self.link.written[-1])[1:], (board, command, bytes.fromhex(request)) + ) + self.assertEqual(result, expected) + + async def test_reads_refuse_unseen_motors(self): + driver = self.driver + reads = [ + lambda: driver.request_motor_currents(motor=3), + lambda: driver.request_motor_configuration(motor=3), + lambda: driver.request_motor_status(motor=3), + lambda: driver.request_motor_position(motor=3), + lambda: driver.request_following_error_configuration(motor=3), + lambda: driver.request_encoder_position(motor=3), + lambda: driver.request_absolute_encoder_position(motor=2), + ] + sent_before = len(self.link.written) + for read in reads: + with self.assertRaises(ValueError): + await read() + self.assertEqual(len(self.link.written), sent_before) + + async def test_absolute_encoder_read_again_until_valid(self): + self.link.answers[(11, 30, b"\x00\x00")] = [ + bytes.fromhex(reply) for reply in ABSOLUTE_ENCODER_REPLIES_AFTER_POWER_ON + ] + sent_before = len(self.link.written) + self.assertEqual(await self.driver.request_absolute_encoder_position(motor=0), 185.99853515625) + sent = [parse_frame(frame=frame)[1:] for frame in self.link.written[sent_before:]] + self.assertEqual(sent, [(11, 30, b"\x00\x00")] * 2) + + async def test_absolute_encoder_never_valid_raises(self): + self.link.answers[(11, 30, b"\x01\x00")] = [ + bytes.fromhex(ABSOLUTE_ENCODER_REPLIES_AFTER_POWER_ON[0]) + ] + sent_before = len(self.link.written) + with self.assertRaisesRegex(MantisTimeoutError, "no valid position in 5 reads"): + await self.driver.request_absolute_encoder_position(motor=1) + self.assertEqual(len(self.link.written) - sent_before, ABSOLUTE_ENCODER_READS) + + if __name__ == "__main__": unittest.main() diff --git a/pylabrobot/formulatrix/mantis/driver/oem_config.py b/pylabrobot/formulatrix/mantis/driver/oem_config.py index 4c4a924dc6c..d20adf20106 100644 --- a/pylabrobot/formulatrix/mantis/driver/oem_config.py +++ b/pylabrobot/formulatrix/mantis/driver/oem_config.py @@ -38,6 +38,7 @@ def _load_unit_configuration(*, oem_config_directory: str) -> Dict[str, Any]: ``PlateClampTimer``), only for a clamp driven as a DC motor (``PlateClampController``). ``rfid``: each input slot's antenna, from every ``RFIDAntennaId.ACC`` of ``RFID.config``, only when it exists; with ``Device.config``'s ``RfidMultiplexerDelay`` as written. + ``led_bar``: the LED count, from ``Device.config``'s ``NumberOfLed``, only where it is set. Args: oem_config_directory: the unit's ``Data/Device/Configs`` folder. @@ -85,6 +86,13 @@ def setting(*, path: str, settings: Dict[str, str], key: str) -> str: } } + if "NumberOfLed" in values: + blocks["led_bar"] = { + "number_of_leds": int( + number(path=device_config, key="NumberOfLed", text=values["NumberOfLed"]) + ) + } + rfid_config = os.path.join(oem_config_directory, "RFID.config") if os.path.isfile(rfid_config): rfid_values = _parse_config_entries(path=rfid_config) diff --git a/pylabrobot/formulatrix/mantis/driver/oem_config_tests.py b/pylabrobot/formulatrix/mantis/driver/oem_config_tests.py index c18b046413a..2ef5a68b4d5 100644 --- a/pylabrobot/formulatrix/mantis/driver/oem_config_tests.py +++ b/pylabrobot/formulatrix/mantis/driver/oem_config_tests.py @@ -29,6 +29,7 @@ ' \r\n' ' \r\n' ' \r\n' + ' \r\n' ) + TAIL ) @@ -69,13 +70,15 @@ def write_files(*, directory: str, files: Dict[str, str]) -> str: class LoadUnitConfigurationTests(unittest.TestCase): - def test_a_v4_folder_gives_its_lock_levels_and_antennas(self): + def test_a_v4_folder_gives_its_lock_levels_antennas_and_led_count(self): with tempfile.TemporaryDirectory() as directory: files = {"Device.config": V4_DEVICE_CONFIG, "RFID.config": V4_RFID_CONFIG} blocks = _load_unit_configuration( oem_config_directory=write_files(directory=directory, files=files) ) - self.assertEqual(blocks, {block: RECORDING[block] for block in ("plate_clamp", "rfid")}) + self.assertEqual( + blocks, {block: RECORDING[block] for block in ("plate_clamp", "rfid", "led_bar")} + ) def test_a_v3_folder_gives_neither(self): with tempfile.TemporaryDirectory() as directory: @@ -116,7 +119,9 @@ async def test_the_folder_alone_gives_the_blocks_and_they_are_saved(self): oem_config_directory=write_files(directory=directory, files=files) ) saved = self.saved_blocks(driver=driver, directory=directory) - self.assertEqual(saved, {block: RECORDING[block] for block in ("plate_clamp", "rfid")}) + self.assertEqual( + saved, {block: RECORDING[block] for block in ("plate_clamp", "rfid", "led_bar")} + ) async def test_a_folder_and_declared_file_that_agree_set_up(self): with tempfile.TemporaryDirectory() as directory: @@ -140,6 +145,37 @@ async def test_a_folder_and_declared_file_that_disagree_refuse_setup(self): with self.assertRaisesRegex(MantisConfigurationError, "plate_clamp: declared"): await driver.setup() + async def test_a_folder_and_declared_file_that_disagree_on_the_led_count_refuse_setup(self): + with tempfile.TemporaryDirectory() as directory: + text = V4_DEVICE_CONFIG.replace('"NumberOfLed" value="20"', '"NumberOfLed" value="24"') + write_files(directory=directory, files={"Device.config": text}) + driver = self.make_driver( + link=FakeLink(answers=make_answers()), + oem_config_directory=directory, + declared_configuration_json=RECORDING_MANTIS_V4, + ) + with self.assertRaisesRegex(MantisConfigurationError, "led_bar: declared"): + await driver.setup() + self.assertIsNone(driver.led_bar) + + async def test_an_older_declared_file_takes_what_it_lacks_from_the_folder(self): + older = {block: values for block, values in RECORDING.items() if block != "led_bar"} + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "older.json") + with open(path, "w", encoding="utf-8") as f: + json.dump(older, f) + text = V4_DEVICE_CONFIG.replace('"NumberOfLed" value="20"', '"NumberOfLed" value="24"') + folder = write_files(directory=directory, files={"Device.config": text}) + with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="WARNING") as logs: + driver = await self.set_up_with( + oem_config_directory=folder, declared_configuration_json=path + ) + self.assertEqual(len(logs.output), 1) + self.assertIn(f"not in {path}: led_bar from {folder}", logs.output[0]) + self.assertNotIn("less accurate", logs.output[0]) + assert driver.led_bar is not None + self.assertEqual(driver.led_bar.number_of_leds, 24) + async def test_neither_takes_the_recording_warns_and_saves_none_of_it(self): with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="WARNING") as logs: driver = await self.set_up_with() diff --git a/pylabrobot/formulatrix/mantis/driver/recordings/README.md b/pylabrobot/formulatrix/mantis/driver/recordings/README.md index aaa0ac22dad..b14a3fd2488 100644 --- a/pylabrobot/formulatrix/mantis/driver/recordings/README.md +++ b/pylabrobot/formulatrix/mantis/driver/recordings/README.md @@ -19,9 +19,10 @@ needed. | block | from | a **simulated** Mantis | a **physical** Mantis | |---|---|---|---| -| `device` | the device, at setup | takes its configuration from it | is checked against it at setup, on `FITTED_FIELDS`; setup refuses if they disagree | +| `device` | the device, at setup | takes its configuration from it; its pump answers the generation's recording's sensor calibration (`feedback_sensor_parameters`) | is checked against it at setup, on `FITTED_FIELDS`; setup refuses if they disagree. The sensor calibration is read off the device at every setup and never saved; one in a declared file is ignored | | `plate_clamp` | `Device.config`: `PlateClampInisiatePWM`, `PlateClampTimer`; only for a clamp driven as a DC motor | takes its lock levels from it | takes its lock levels from it | | `rfid` | `RFID.config`: every `RFIDAntennaId.ACC`; `Device.config`: `RfidMultiplexerDelay`, as written (`multiplexer_delay_raw`) | takes its slots from it; every slot reads empty | takes each slot's antenna; a delay other than 0 refuses the reads, as its unit is unconfirmed | +| `led_bar` | `Device.config`: `NumberOfLed`; set on a v4 only | takes its LED count from it | takes its LED count from it: each colour write sets that many LEDs | For each unit block, setup takes: @@ -31,14 +32,18 @@ For each unit block, setup takes: 3. the generation's recording's, when neither has it, with a warning: another unit's values, less accurate on this one. `save_configuration` does not save a block taken from a recording. +A declared file saved before a block or a non-fitted `device` field existed still loads: each +value it lacks comes from the folder, else the recording, on a device and in simulation, with one +warning naming each and where it came from. + | file | device | |---|---| -| `mantis_v4.json` | Mantis v4, motion firmware 4.20.6.2, with that unit's lock levels and RFID antennas (vendor software 5.5.2). The default for a simulated `Mantis`. | +| `mantis_v4.json` | Mantis v4, motion firmware 4.20.6.2, with that unit's lock levels, RFID antennas, LED count (20) and pressure sensor calibration. The default for a simulated `Mantis`. | | `mantis_v3.json` | Mantis v3 (vendor software 4.8.2.5): the generation only. No v3 has run setup, so its firmware is not recorded; its install has no DC-motor clamp and no `RFID.config`, so it has no unit blocks yet. | ## What is not in one - **The serial number.** Never saved. - **The device name.** Read with `request_name`; a simulated Mantis does not answer it. -- **State.** Where the arm is, the clamp's state, which chips are loaded, anything that changes - during a run. +- **State.** Where the arm is, the clamp's state, what the LED bar shows, which chips are loaded, + anything that changes during a run. diff --git a/pylabrobot/formulatrix/mantis/driver/recordings/mantis_v4.json b/pylabrobot/formulatrix/mantis/driver/recordings/mantis_v4.json index 407629a7be1..8c17e404828 100644 --- a/pylabrobot/formulatrix/mantis/driver/recordings/mantis_v4.json +++ b/pylabrobot/formulatrix/mantis/driver/recordings/mantis_v4.json @@ -20,6 +20,20 @@ -160.0, 2300.0 ] + }, + "feedback_sensor_parameters": { + "bottle": [ + 0.010651180520653725, + -13.580931663513184 + ], + "chip": [ + 0.010661732405424118, + -13.671295166015625 + ], + "vacuum": [ + 0.010320834815502167, + -38.775638580322266 + ] } }, "plate_clamp": { @@ -48,5 +62,8 @@ "6": 5 }, "multiplexer_delay_raw": 0.0 + }, + "led_bar": { + "number_of_leds": 20 } } diff --git a/pylabrobot/formulatrix/mantis/driver/simulator.py b/pylabrobot/formulatrix/mantis/driver/simulator.py index f3e0fa54057..5fe33512264 100644 --- a/pylabrobot/formulatrix/mantis/driver/simulator.py +++ b/pylabrobot/formulatrix/mantis/driver/simulator.py @@ -6,16 +6,20 @@ from pylabrobot.io.ftdi import FTDI from .configuration import ( + RECORDING_MANTIS_V3, + RECORDING_MANTIS_V4, DeviceConfiguration, compute_generation, configuration_from_json, read_configuration, ) from .errors import MantisConfigurationError, MantisTimeoutError +from .features.led_bar import LedBar from .features.plate_clamp import CLAMPING_DONE, UNCLAMPING_DONE, PlateClamp +from .features.pump import PressureController, Pump from .features.rfid import Rfid from .framing import ABSOLUTE_POSITION_SENSOR_BOARD, LC3_BOARD, MOTION_BOARD, PRESSURE_BOARD -from .master import MOTORS, MantisDriver +from .master import ABSOLUTE_ENCODER_MOTORS, MOTORS, MantisDriver logger = logging.getLogger(__name__) @@ -47,6 +51,58 @@ async def request_state(self, *, attempts: int = 3) -> int: return self._state +class MantisSimulationPump(Pump): + """A pump with no device behind it, answering from the declared configuration. + + Off, its controllers off and disabled, every line at 0, with the recorded sensor calibration. + """ + + async def measure_pressure(self, *, controller: PressureController) -> float: + """0: no line holds pressure.""" + self._get_controller_index(controller=controller) + return 0.0 + + async def request_feedback_sensor_parameters( + self, *, controller: PressureController + ) -> Tuple[float, float]: + """The recorded calibration, as setup read it. + + Raises: + MantisConfigurationError: if the generation's recording holds none for ``controller``. + """ + self._get_controller_index(controller=controller) + parameters = self._driver.configuration.feedback_sensor_parameters.get(controller) + if parameters is None: + raise MantisConfigurationError( + f"{self._driver.name}: no feedback_sensor_parameters recorded for the {controller} " + "controller" + ) + return parameters + + async def request_controller_status(self, *, controller: PressureController) -> int: + """0: off.""" + self._get_controller_index(controller=controller) + return 0 + + async def request_controller_enabled(self, *, controller: PressureController) -> bool: + """False: no controller regulates.""" + self._get_controller_index(controller=controller) + return False + + async def request_pump_on(self) -> bool: + """False: the pump is off.""" + return False + + async def _request_calibration(self) -> Dict[PressureController, Tuple[float, float]]: + """The generation's recording's calibration; none recorded is none.""" + generation = self._driver.configuration.generation + path = RECORDING_MANTIS_V4 if generation == "v4" else RECORDING_MANTIS_V3 + recorded = read_configuration(path=path)["device"].get("feedback_sensor_parameters", {}) + return { + controller: (float(scale), float(offset)) for controller, (scale, offset) in recorded.items() + } + + class MantisSimulationRfid(Rfid): """An RFID reader with no device behind it: every slot reads empty.""" @@ -58,12 +114,30 @@ async def _read_tag(self) -> Optional[bytes]: return None +class MantisSimulationLedBar(LedBar): + """An LED bar with no device behind it: every command is taken, and no hold is waited.""" + + async def _unchecked_fw_stop_animator(self) -> None: + """Send nothing.""" + + async def _unchecked_fw_start_animator(self, *, lowest_index: int, highest_index: int) -> None: + """Send nothing.""" + + async def _unchecked_fw_set_rgb_led(self, *, colors: List[int]) -> None: + """Send nothing.""" + + async def _sleep(self, *, seconds: float) -> None: + """Return at once: a simulated bar has nothing to show.""" + + class MantisSimulationDriver(MantisDriver): """A Mantis driver with no device behind it. Nothing reaches a wire. The configuration comes from ``declared_configuration_json``, and the reads answer from it; a board it leaves ``None`` does not answer. The plate clamp's state changes - at once, and every RFID slot reads empty. The device name is not simulated. + at once, every RFID slot reads empty, and the LED bar takes every command and holds nothing. The + motors read idle at 0, as after power-on. The device name and the motor reads the configuration + does not hold are not simulated. """ def __init__( @@ -72,6 +146,7 @@ def __init__( name: str = "Mantis", *, oem_config_directory: Optional[str] = None, + show_actions: bool = True, ): """ Args: @@ -79,11 +154,13 @@ def __init__( configuration. name: the device name, heading its log messages. oem_config_directory: a unit's vendor ``Data/Device/Configs`` folder, as for a device. + show_actions: whether the LED bar shows each action, as for a device. """ super().__init__( name=name, declared_configuration_json=declared_configuration_json, oem_config_directory=oem_config_directory, + show_actions=show_actions, ) def _build_io(self, *, device_id: Optional[str], ftdi_library: Optional[str]) -> FTDI: @@ -133,13 +210,53 @@ async def request_motor_limits(self, *, motor: int) -> Tuple[float, float]: self._check_index(name="motor", value=motor, allowed=MOTORS) return self.configuration.motor_limits[motor] + async def request_motor_currents(self, *, motor: int) -> Tuple[float, float, float]: + """Raise: the currents are not part of the configuration.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + raise NotImplementedError(f"{self.name}: the motor currents are not simulated") + + async def request_motor_configuration( + self, *, motor: int + ) -> Tuple[int, float, float, float, int, int]: + """Raise: a motor's configuration is not part of the device configuration.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + raise NotImplementedError(f"{self.name}: the motor configuration is not simulated") + + async def request_motor_status(self, *, motor: int) -> int: + """0: idle, no fault, as after power-on.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + return 0 + + async def request_motor_position(self, *, motor: int) -> float: + """0, as after power-on.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + return 0.0 + + async def request_following_error_configuration(self, *, motor: int) -> Tuple[bool, float]: + """Raise: the following-error check is not part of the configuration.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + raise NotImplementedError(f"{self.name}: the following-error check is not simulated") + + async def request_encoder_position(self, *, motor: int) -> float: + """0, as after power-on.""" + self._check_index(name="motor", value=motor, allowed=MOTORS) + return 0.0 + + async def request_absolute_encoder_position(self, *, motor: int) -> float: + """Raise: a raw reading means nothing without the unit's encoder correction.""" + self._check_index(name="motor", value=motor, allowed=ABSOLUTE_ENCODER_MOTORS) + raise NotImplementedError(f"{self.name}: the absolute encoders are not simulated") + async def _request_configuration(self) -> DeviceConfiguration: - """The declared configuration's ``device`` block; the generation from its boards, as a - device's is.""" - configuration = configuration_from_json( + """The declared configuration's ``device`` block, a field it lacks filled as for the unit + blocks; the generation from its boards, as a device's is.""" + declared = configuration_from_json( fields=read_configuration(path=self.declared_configuration_json or "")["device"] ) - configuration.generation = compute_generation(firmware_versions=configuration.firmware_versions) + generation = compute_generation(firmware_versions=declared.firmware_versions) + values, _ = self._load_unit_values(generation=generation) + configuration = configuration_from_json(fields=values["device"]) + configuration.generation = generation return configuration def _discover_features(self) -> None: @@ -153,6 +270,14 @@ def _discover_features(self) -> None: self.plate_clamp, MantisSimulationPlateClamp ): self.plate_clamp = MantisSimulationPlateClamp(driver=self, levels=self.plate_clamp.levels) + if self.pump is not None and not isinstance(self.pump, MantisSimulationPump): + self.pump = MantisSimulationPump(driver=self) + if self.led_bar is not None and not isinstance(self.led_bar, MantisSimulationLedBar): + self.led_bar = MantisSimulationLedBar( + driver=self, + number_of_leds=self.led_bar.number_of_leds, + show_actions=self.led_bar.show_actions, + ) if self.rfid is None: rfid_block = self._unit_blocks.get("rfid") slot_to_antenna: Optional[Dict[int, int]] = None diff --git a/pylabrobot/formulatrix/mantis/driver/simulator_tests.py b/pylabrobot/formulatrix/mantis/driver/simulator_tests.py index 5d4fd38ad36..d67b2480d6a 100644 --- a/pylabrobot/formulatrix/mantis/driver/simulator_tests.py +++ b/pylabrobot/formulatrix/mantis/driver/simulator_tests.py @@ -8,6 +8,7 @@ from pylabrobot.formulatrix.mantis import Mantis from pylabrobot.formulatrix.mantis.device import RECORDING_MANTIS_V4 from pylabrobot.formulatrix.mantis.driver.errors import MantisTimeoutError +from pylabrobot.formulatrix.mantis.driver.simulator import MantisSimulationLedBar class SimulationTests(unittest.IsolatedAsyncioTestCase): @@ -48,6 +49,7 @@ async def test_board_the_recording_declares_absent_does_not_answer(self): await mantis.setup() self.addAsyncCleanup(mantis.stop) self.assertEqual(mantis.driver.configuration.generation, "v3") + self.assertIsNone(mantis.led_bar) with self.assertRaises(MantisTimeoutError): await mantis.driver.request_firmware_version(board=11) @@ -62,6 +64,72 @@ async def test_reads_answer_from_the_configuration(self): with self.assertRaisesRegex(NotImplementedError, "name is not simulated"): await self.mantis.driver.request_name() + async def test_motors_read_idle_at_0(self): + driver = self.mantis.driver + for motor in (0, 1, 2): + with self.subTest(motor=motor): + self.assertEqual(await driver.request_motor_status(motor=motor), 0) + self.assertEqual(await driver.request_motor_position(motor=motor), 0.0) + self.assertEqual(await driver.request_encoder_position(motor=motor), 0.0) + with self.assertRaisesRegex(NotImplementedError, "absolute encoders are not simulated"): + await driver.request_absolute_encoder_position(motor=0) + with self.assertRaises(ValueError): + await driver.request_motor_status(motor=3) + + async def test_pump_reads_off_with_the_recorded_calibration(self): + pump = self.mantis.pump + assert pump is not None + with open(RECORDING_MANTIS_V4, encoding="utf-8") as f: + recorded = json.load(f)["device"]["feedback_sensor_parameters"] + self.assertEqual( + self.mantis.driver.configuration.feedback_sensor_parameters, + {controller: tuple(value) for controller, value in recorded.items()}, + ) + self.assertEqual( + await pump.request_feedback_sensor_parameters(controller="vacuum"), + tuple(recorded["vacuum"]), + ) + self.assertEqual(await pump.measure_pressure(controller="chip"), 0.0) + self.assertEqual(await pump.request_controller_status(controller="chip"), 0) + self.assertFalse(await pump.request_controller_enabled(controller="chip")) + self.assertFalse(await pump.request_pump_on()) + with self.assertRaises(ValueError): + await pump.measure_pressure(controller="air") # type: ignore[arg-type] + + async def test_an_older_declared_file_takes_what_it_lacks_from_the_recording(self): + with open(RECORDING_MANTIS_V4, encoding="utf-8") as f: + recording = json.load(f) + del recording["device"]["feedback_sensor_parameters"] + del recording["led_bar"] + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "older.json") + with open(path, "w", encoding="utf-8") as f: + json.dump(recording, f) + mantis = Mantis(simulation=True, declared_configuration_json=path) + with self.assertLogs("pylabrobot.formulatrix.mantis.driver.master", level="WARNING") as logs: + await mantis.setup() + self.addAsyncCleanup(mantis.stop) + self.assertEqual(len(logs.output), 1) + self.assertIn(f"led_bar from {RECORDING_MANTIS_V4}", logs.output[0]) + assert mantis.pump is not None and mantis.led_bar is not None + self.assertEqual(mantis.led_bar.number_of_leds, 20) + self.assertEqual( + await mantis.pump.request_feedback_sensor_parameters(controller="bottle"), + (0.010651180520653725, -13.580931663513184), + ) + + async def test_led_bar_takes_every_command(self): + bar = self.mantis.led_bar + assert isinstance(bar, MantisSimulationLedBar) + self.assertEqual(bar.number_of_leds, 20) + await bar.set_color(color="#FF0080") + await bar.set_color(color=["green"] * 10 + ["off"] * 10) + await bar.start_animation(effect="wave") + await bar.stop_animation() + await bar.turn_off() + with self.assertRaises(ValueError): + await bar.set_color(color="purple") + async def test_simulated_mantis_sends_nothing(self): with self.assertRaisesRegex(NotImplementedError, "sends nothing"): await self.mantis.driver._send_command(board=0, command=3)