Support ophyd-async v0.6.0

src/mx_bluesky/hyperion/device_setup_plans/setup_panda.py

@@ -11,9 +11,7 @@
 from ophyd_async.fastcs.panda import (
     HDFPanda,
     SeqTable,
-    SeqTableRow,
     SeqTrigger,
-    seq_table_from_rows,
 )
 
 import mx_bluesky.hyperion.resources.panda as panda_resource
@@ -85,10 +83,10 @@ def _get_seq_table(
     # BITA_1 trigger wired from TTLIN1, this is the trigger input
 
     # +ve direction scan
-    rows = [SeqTableRow(trigger=SeqTrigger.BITA_1, time2=1)]
 
-    rows.append(
-        SeqTableRow(
+    table = (
+        SeqTable.row(trigger=SeqTrigger.BITA_1, time2=1)
+        + SeqTable.row(
             repeats=num_pulses,
             trigger=SeqTrigger.POSA_GT,
             position=start_of_grid_x_counts,
@@ -97,13 +95,10 @@ def _get_seq_table(
             time2=delay_between_pulses - PULSE_WIDTH_US,
             outa2=False,
         )
-    )
-
-    # -ve direction scan
-    rows.append(SeqTableRow(trigger=SeqTrigger.BITA_1, time2=1))
-
-    rows.append(
-        SeqTableRow(
+        +
+        # -ve direction scan
+        SeqTable.row(trigger=SeqTrigger.BITA_1, time2=1)
+        + SeqTable.row(
             repeats=num_pulses,
             trigger=SeqTrigger.POSA_LT,
             position=end_of_grid_x_counts + exposure_distance_x_counts,
@@ -114,8 +109,6 @@ def _get_seq_table(
         )
     )
 
-    table = seq_table_from_rows(*rows)
-
     return table
 
 

tests/unit_tests/hyperion/device_setup_plans/test_setup_panda.py

@@ -1,5 +1,4 @@
 from datetime import datetime
-from typing import NamedTuple
 from unittest.mock import MagicMock, patch
 
 import numpy as np
@@ -9,7 +8,7 @@
 from bluesky.simulators import RunEngineSimulator, assert_message_and_return_remaining
 from dodal.common.types import UpdatingPathProvider
 from dodal.devices.fast_grid_scan import PandAGridScanParams
-from ophyd_async.fastcs.panda import SeqTrigger
+from ophyd_async.fastcs.panda import SeqTable, SeqTrigger
 
 from mx_bluesky.hyperion.device_setup_plans.setup_panda import (
     MM_TO_ENCODER_COUNTS,
@@ -69,16 +68,6 @@ def test_setup_panda_performs_correct_plans(mock_load_device, sim_run_engine):
     assert num_of_waits == 3
 
 
-class SeqRow(NamedTuple):
-    repeats: int
-    trigger: SeqTrigger
-    position: int
-    time1: int
-    outa1: int
-    time2: int
-    outa2: int
-
-
 @pytest.mark.parametrize(
     "x_steps, x_step_size, x_start, run_up_distance_mm, time_between_x_steps_ms, exposure_time_s",
     [
@@ -142,44 +131,54 @@ def test_setup_panda_correctly_configures_table(
 
     PULSE_WIDTH_US = 1
     SPACE_WIDTH_US = int(time_between_x_steps_ms * 1000 - PULSE_WIDTH_US)
-    expected_seq_rows: list[SeqRow] = [
-        SeqRow(1, SeqTrigger.BITA_1, 0, 0, 0, 1, 0),
-        SeqRow(
-            x_steps,
-            SeqTrigger.POSA_GT,
-            int(params.x_start * MM_TO_ENCODER_COUNTS),
-            PULSE_WIDTH_US,
-            1,
-            SPACE_WIDTH_US,
-            0,
-        ),
-    ]
 
     exposure_distance_counts = exposure_distance_mm * MM_TO_ENCODER_COUNTS
-    expected_seq_rows.extend(
-        [
-            SeqRow(1, SeqTrigger.BITA_1, 0, 0, 0, 1, 0),
-            SeqRow(
-                x_steps,
-                SeqTrigger.POSA_LT,
-                int(
-                    (params.x_start + (params.x_steps - 1) * params.x_step_size)
-                    * MM_TO_ENCODER_COUNTS
-                    + exposure_distance_counts
-                ),
-                PULSE_WIDTH_US,
-                1,
-                SPACE_WIDTH_US,
-                0,
+    expected_seq_table: SeqTable = (
+        SeqTable.row(
+            repeats=1,
+            trigger=SeqTrigger.BITA_1,
+            position=0,
+            time1=False,
+            outa1=False,
+            time2=True,
+            outa2=False,
+        )
+        + SeqTable.row(
+            repeats=x_steps,
+            trigger=SeqTrigger.POSA_GT,
+            position=int(params.x_start * MM_TO_ENCODER_COUNTS),
+            time1=PULSE_WIDTH_US,
+            outa1=True,
+            time2=SPACE_WIDTH_US,
+            outa2=False,
+        )
+        + SeqTable.row(
+            repeats=1,
+            trigger=SeqTrigger.BITA_1,
+            position=0,
+            time1=False,
+            outa1=False,
+            time2=True,
+            outa2=False,
+        )
+        + SeqTable.row(
+            repeats=x_steps,
+            trigger=SeqTrigger.POSA_LT,
+            position=int(
+                (params.x_start + (params.x_steps - 1) * params.x_step_size)
+                * MM_TO_ENCODER_COUNTS
+                + exposure_distance_counts
             ),
-        ]
+            time1=PULSE_WIDTH_US,
+            outa1=True,
+            time2=SPACE_WIDTH_US,
+            outa2=False,
+        )
     )
 
-    for key in SeqRow._fields:
+    for attr_name in table.__annotations__.keys():
         np.testing.assert_array_equal(
-            table.get(key),
-            [getattr(row, key) for row in expected_seq_rows],
-            f"Sequence table for field {key} does not match",
+            getattr(table, attr_name), getattr(expected_seq_table, attr_name)
         )