Added User array to control GPIO during motion program

src/ophyd_async/epics/pmac/_pmac_io.py

@@ -2,8 +2,7 @@
 
 from ophyd_async.core import Array1D, DeviceVector, StandardReadable
 from ophyd_async.epics import motor
-
-from ..signal import epics_signal_r, epics_signal_rw
+from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
 
 
 class Pmac(StandardReadable):
@@ -13,6 +12,7 @@ def __init__(self, prefix: str, name: str = "") -> None:
         self.time_array = epics_signal_rw(
             Array1D[np.float64], prefix + ":ProfileTimeArray"
         )
+        self.user_array = epics_signal_rw(Array1D[np.int32], prefix + ":UserArray")
         cs_letters = "ABCUVWXYZ"
         # 1 indexed CS axes so we can index into them from the compound motor input link
         self.positions = DeviceVector(

src/ophyd_async/epics/pmac/_pmac_trajectory.py

@@ -44,7 +44,10 @@ async def prepare(self, value: PmacTrajInfo):
         velocities: dict[int, npt.NDArray[np.float64]] = {}
         cs_axes: dict[PmacMotor, int] = {}
         time_array: npt.NDArray[np.float64] = np.empty(
-            2 * scan_size + (len(gaps) * 4) + 1, dtype=np.float64
+            2 * scan_size + ((len(gaps) + 1) * 4) + 1, dtype=np.float64
+        )
+        user_array: npt.NDArray[np.int32] = np.empty(
+            2 * scan_size + ((len(gaps) + 1) * 4) + 1, dtype=np.int32
         )
         # Which Axes are in use?
         scan_axes = chunk.axes()
@@ -54,10 +57,10 @@ async def prepare(self, value: PmacTrajInfo):
                 # Initialise numpy arrays for Positions, velocities and time within dict
                 # for each axis in scan
                 positions[cs_index] = np.empty(
-                    2 * scan_size + (len(gaps) * 4) + 1, dtype=np.float64
+                    2 * scan_size + ((len(gaps) + 1) * 4) + 1, dtype=np.float64
                 )
                 velocities[cs_index] = np.empty(
-                    2 * scan_size + (len(gaps) * 4) + 1, dtype=np.float64
+                    2 * scan_size + ((len(gaps) + 1) * 4) + 1, dtype=np.float64
                 )
                 cs_ports.add(cs_port)
                 cs_axes[axis] = cs_index
@@ -95,6 +98,7 @@ async def prepare(self, value: PmacTrajInfo):
                     time_array[profile_start:profile_gap] = np.repeat(
                         chunk.midpoints["DURATION"][start:gap] / (2 * TICK_S), 2
                     )
+                    user_array[profile_start:profile_gap] = 1
             if gap < scan_size:
                 # Create Position, velocity and time arrays for the gap
                 pos_gap, vel_gap, time_gap = await get_gap_profile(chunk, gap)
@@ -109,7 +113,7 @@ async def prepare(self, value: PmacTrajInfo):
                         ] = vel_gap[axis]
                     else:
                         time_array[profile_gap : profile_gap + len_gap] = time_gap
-
+                        user_array[profile_gap : profile_gap + len_gap] = 2
                 added_point += len_gap
             start = gap
 
@@ -154,6 +158,7 @@ async def prepare(self, value: PmacTrajInfo):
         self.scantime += run_up_time + final_time
         time_array[0] += int(run_up_time / TICK_S)
         time_array[(profile_length)] = int(final_time / TICK_S)
+        user_array[(profile_length)] = 8
         profile_length += 1
 
         for axis in scan_axes:
@@ -169,6 +174,7 @@ async def prepare(self, value: PmacTrajInfo):
                 )
             else:
                 await self.pmac.time_array.set(time_array[:profile_length])
+                await self.pmac.user_array.set(user_array[:profile_length])
 
         # MOVE TO START
         for axis in scan_axes:

tests/epics/pmac/test_pmac.py

@@ -125,6 +125,30 @@ async def test_sim_pmac_simple_trajectory(sim_x_motor, sim_pmac) -> None:
             50000,
         ]
     ).all()
+    assert (
+        await trigger_logic.pmac.user_array.get_value()
+        == [
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            8,
+        ]
+    ).all()
     assert await trigger_logic.pmac.points_to_build.get_value() == 19
     assert await sim_x_motor.user_setpoint.get_value() == 0.9875
     assert trigger_logic.scantime == 9.1
@@ -353,6 +377,49 @@ async def test_sim_grid_trajectory(sim_x_motor, sim_y_motor, sim_pmac) -> None:
             100000,
         ]
     )
+    assert await trigger_logic.pmac.user_array.get_value() == pytest.approx(
+        [
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            2,
+            2,
+            2,
+            2,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            2,
+            2,
+            2,
+            2,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            1,
+            8,
+        ]
+    )
     assert await trigger_logic.pmac.points_to_build.get_value() == 39
     assert await sim_x_motor.user_setpoint.get_value() == 0.95
     assert trigger_logic.scantime == 15.2