Add stop method

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
     "colorlog",
     "pydantic>=2.0",
     "pydantic-numpy",
-    "scanspec",
+    "scanspec==0.7.2",
     "velocity-profile",
 ]
 dynamic = ["version"]

src/ophyd_async/epics/pmac/_pmacIO.py

@@ -1,13 +1,12 @@
 import numpy as np
 import numpy.typing as npt
-from bluesky.protocols import Flyable, Preparable
 
 from ophyd_async.core import DeviceVector, StandardReadable, SubsetEnum
 
 from ..signal.signal import epics_signal_r, epics_signal_rw
 
 
-class Pmac(StandardReadable, Flyable, Preparable):
+class Pmac(StandardReadable):
     """Device that moves a PMAC Motor record"""
 
     def __init__(self, prefix: str, cs="", name="") -> None:
@@ -43,5 +42,6 @@ def __init__(self, prefix: str, cs="", name="") -> None:
         self.execute_profile = epics_signal_rw(bool, prefix + ":ProfileExecute")
         self.scan_percent = epics_signal_r(float, prefix + ":TscanPercent_RBV")
         cs_names_enum = SubsetEnum[cs]
+        self.profile_abort = epics_signal_rw(bool, prefix + ":ProfileAbort")
         self.profile_cs_name = epics_signal_rw(cs_names_enum, prefix + ":ProfileCsName")
         self.profile_calc_vel = epics_signal_rw(bool, prefix + ":ProfileCalcVel")

src/ophyd_async/epics/pmac/_pmacTrajectory.py

@@ -3,7 +3,7 @@
 import numpy as np
 import numpy.typing as npt
 from pydantic import BaseModel, Field
-from scanspec.specs import Frames, Path
+from scanspec.specs import Frames, Path, Spec
 from velocity_profile import velocityprofile as vp
 
 from ophyd_async.core import TriggerLogic
@@ -17,8 +17,7 @@
 
 
 class PmacTrajInfo(BaseModel):
-    stack: list[Frames[Motor]] = Field(strict=True)
-    # pmac: Pmac = Field(strict=True)
+    spec: Spec[Motor] = Field()
 
 
 class PmacTrajectoryTriggerLogic(TriggerLogic[PmacTrajInfo]):
@@ -34,7 +33,7 @@ async def prepare(self, value: PmacTrajInfo):
         for i in range(len("ABCUVWXYZ")):
             self.pmac.use_axis[i + 1].set(False)
 
-        path = Path(value.stack)
+        path = Path(value.spec.calculate())
         chunk = path.consume()
         gaps = self._calculate_gaps(chunk)
         if gaps[0] == 0:
@@ -188,6 +187,9 @@ async def prepare(self, value: PmacTrajInfo):
     async def kickoff(self):
         self.status = self.pmac.execute_profile.set(1, timeout=self.scantime + 10)
 
+    async def stop(self):
+        await self.pmac.profile_abort.set(1)
+
     @WatchableAsyncStatus.wrap
     async def complete(self):
         async for percent in observe_value(self.scan_percent):

tests/epics/pmac/test_pmac.py

@@ -42,8 +42,7 @@ async def test_sim_pmac_simple_trajectory(sim_x_motor) -> None:
     async with DeviceCollector(mock=True):
         pmac = Pmac(prefix, "BRICK1.CS3", name="sim_pmac")
     spec = fly(Line(sim_x_motor, 1, 5, 9), 1)
-    stack = spec.calculate()
-    info = PmacTrajInfo(stack)
+    info = PmacTrajInfo(spec=spec)
     trigger_logic = PmacTrajectoryTriggerLogic(pmac)
     await trigger_logic.prepare(info)
     assert await trigger_logic.pmac.positions[9].get_value() == pytest.approx(
@@ -129,8 +128,7 @@ async def test_sim_grid_trajectory(sim_x_motor, sim_y_motor) -> None:
     async with DeviceCollector(mock=True):
         pmac = Pmac(prefix, "BRICK1.CS3", name="sim_pmac")
     spec = fly(Line(sim_y_motor, 10, 12, 3) * ~Line(sim_x_motor, 1, 5, 5), 1)
-    stack = spec.calculate()
-    info = PmacTrajInfo(stack)
+    info = PmacTrajInfo(spec=spec)
     trigger_logic = PmacTrajectoryTriggerLogic(pmac)
     await trigger_logic.prepare(info)
     assert await trigger_logic.pmac.positions[9].get_value() == pytest.approx(