diff --git a/src/ophyd_async/core/_detector.py b/src/ophyd_async/core/_detector.py
index 7d22fa6ace..45d48fa43f 100644
--- a/src/ophyd_async/core/_detector.py
+++ b/src/ophyd_async/core/_detector.py
@@ -51,20 +51,24 @@ class TriggerInfo(BaseModel):
     """Minimal set of information required to setup triggering on a detector"""
 
     #: Number of triggers that will be sent, 0 means infinite
-    number: int = Field(gt=0)
+    number: int = Field(ge=0)
     #: Sort of triggers that will be sent
-    trigger: DetectorTrigger = Field()
+    trigger: DetectorTrigger = Field(default=DetectorTrigger.internal)
     #: What is the minimum deadtime between triggers
-    deadtime: float | None = Field(ge=0)
+    deadtime: float | None = Field(default=None, ge=0)
     #: What is the maximum high time of the triggers
-    livetime: float | None = Field(ge=0)
+    livetime: float | None = Field(default=None, ge=0)
     #: What is the maximum timeout on waiting for a frame
-    frame_timeout: float | None = Field(None, gt=0)
+    frame_timeout: float | None = Field(default=None, gt=0)
     #: How many triggers make up a single StreamDatum index, to allow multiple frames
     #: from a faster detector to be zipped with a single frame from a slow detector
     #: e.g. if num=10 and multiplier=5 then the detector will take 10 frames,
     #: but publish 2 indices, and describe() will show a shape of (5, h, w)
     multiplier: int = 1
+    #: The number of times the detector can go through a complete cycle of kickoff and
+    #: complete without needing to re-arm. This is important for detectors where the
+    #: process of arming is expensive in terms of time
+    iteration: int = 1
 
 
 class DetectorControl(ABC):
@@ -78,27 +82,35 @@ def get_deadtime(self, exposure: float | None) -> float:
         """For a given exposure, how long should the time between exposures be"""
 
     @abstractmethod
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
+    async def prepare(self, trigger_info: TriggerInfo):
         """
-        Arm detector, do all necessary steps to prepare detector for triggers.
+        Do all necessary steps to prepare the detector for triggers.
 
         Args:
-            num: Expected number of frames
-            trigger: Type of trigger for which to prepare the detector. Defaults to
-            DetectorTrigger.internal.
-            exposure: Exposure time with which to set up the detector. Defaults to None
-            if not applicable or the detector is expected to use its previously-set
-            exposure time.
+            trigger_info: This is a Pydantic model which contains
+                number Expected number of frames.
+                trigger Type of trigger for which to prepare the detector. Defaults
+                to DetectorTrigger.internal.
+                livetime Livetime / Exposure time with which to set up the detector.
+                Defaults to None
+                if not applicable or the detector is expected to use its previously-set
+                exposure time.
+                deadtime Defaults to None. This is the minimum deadtime between
+                triggers.
+                multiplier The number of triggers grouped into a single StreamDatum
+                index.
+        """
 
-        Returns:
-            AsyncStatus: Status representing the arm operation. This function returning
-            represents the start of the arm. The returned status completing means
-            the detector is now armed.
+    @abstractmethod
+    async def arm(self) -> None:
+        """
+        Arm the detector
+        """
+
+    @abstractmethod
+    async def wait_for_idle(self):
+        """
+        This will wait on the internal _arm_status and wait for it to get disarmed/idle
         """
 
     @abstractmethod
@@ -186,7 +198,7 @@ def __init__(
         self._watchers: List[Callable] = []
         self._fly_status: Optional[WatchableAsyncStatus] = None
         self._fly_start: float
-
+        self._iterations_completed: int = 0
         self._intial_frame: int
         self._last_frame: int
         super().__init__(name)
@@ -248,15 +260,15 @@ async def trigger(self) -> None:
                     trigger=DetectorTrigger.internal,
                     deadtime=None,
                     livetime=None,
+                    frame_timeout=None,
                 )
             )
+        assert self._trigger_info
+        assert self._trigger_info.trigger is DetectorTrigger.internal
         # Arm the detector and wait for it to finish.
         indices_written = await self.writer.get_indices_written()
-        written_status = await self.controller.arm(
-            num=self._trigger_info.number,
-            trigger=self._trigger_info.trigger,
-        )
-        await written_status
+        await self.controller.arm()
+        await self.controller.wait_for_idle()
         end_observation = indices_written + 1
 
         async for index in self.writer.observe_indices_written(
@@ -283,35 +295,35 @@ async def prepare(self, value: TriggerInfo) -> None:
         Args:
             value: TriggerInfo describing how to trigger the detector
         """
-        self._trigger_info = value
         if value.trigger != DetectorTrigger.internal:
             assert (
                 value.deadtime
             ), "Deadtime must be supplied when in externally triggered mode"
         if value.deadtime:
-            required = self.controller.get_deadtime(self._trigger_info.livetime)
+            required = self.controller.get_deadtime(value.livetime)
             assert required <= value.deadtime, (
                 f"Detector {self.controller} needs at least {required}s deadtime, "
                 f"but trigger logic provides only {value.deadtime}s"
             )
+        self._trigger_info = value
         self._initial_frame = await self.writer.get_indices_written()
         self._last_frame = self._initial_frame + self._trigger_info.number
-        self._arm_status = await self.controller.arm(
-            num=self._trigger_info.number,
-            trigger=self._trigger_info.trigger,
-            exposure=self._trigger_info.livetime,
+        self._describe, _ = await asyncio.gather(
+            self.writer.open(value.multiplier), self.controller.prepare(value)
         )
-        self._fly_start = time.monotonic()
-        self._describe = await self.writer.open(value.multiplier)
+        if value.trigger != DetectorTrigger.internal:
+            await self.controller.arm()
+            self._fly_start = time.monotonic()
 
     @AsyncStatus.wrap
     async def kickoff(self):
-        if not self._arm_status:
-            raise Exception("Detector not armed!")
+        assert self._trigger_info, "Prepare must be called before kickoff!"
+        if self._iterations_completed >= self._trigger_info.iteration:
+            raise Exception(f"Kickoff called more than {self._trigger_info.iteration}")
+        self._iterations_completed += 1
 
     @WatchableAsyncStatus.wrap
     async def complete(self):
-        assert self._arm_status, "Prepare not run"
         assert self._trigger_info
         async for index in self.writer.observe_indices_written(
             self._trigger_info.frame_timeout
@@ -332,6 +344,8 @@ async def complete(self):
             )
             if index >= self._trigger_info.number:
                 break
+        if self._iterations_completed == self._trigger_info.iteration:
+            await self.controller.wait_for_idle()
 
     async def describe_collect(self) -> Dict[str, DataKey]:
         return self._describe
diff --git a/src/ophyd_async/core/_signal.py b/src/ophyd_async/core/_signal.py
index 503c1f9ca3..340298160b 100644
--- a/src/ophyd_async/core/_signal.py
+++ b/src/ophyd_async/core/_signal.py
@@ -505,7 +505,7 @@ async def wait_for_value(self, signal: SignalR[T], timeout: Optional[float]):
         try:
             await asyncio.wait_for(self._wait_for_value(signal), timeout)
         except asyncio.TimeoutError as e:
-            raise TimeoutError(
+            raise asyncio.TimeoutError(
                 f"{signal.name} didn't match {self._matcher_name} in {timeout}s, "
                 f"last value {self._last_value!r}"
             ) from e
diff --git a/src/ophyd_async/epics/adaravis/_aravis_controller.py b/src/ophyd_async/epics/adaravis/_aravis_controller.py
index 6349d111b1..894a46c008 100644
--- a/src/ophyd_async/epics/adaravis/_aravis_controller.py
+++ b/src/ophyd_async/epics/adaravis/_aravis_controller.py
@@ -1,12 +1,13 @@
 import asyncio
-from typing import Literal, Optional, Tuple
+from typing import Literal, Tuple
 
 from ophyd_async.core import (
-    AsyncStatus,
     DetectorControl,
     DetectorTrigger,
+    TriggerInfo,
     set_and_wait_for_value,
 )
+from ophyd_async.core._status import AsyncStatus
 from ophyd_async.epics import adcore
 
 from ._aravis_io import AravisDriverIO, AravisTriggerMode, AravisTriggerSource
@@ -23,24 +24,20 @@ class AravisController(DetectorControl):
     def __init__(self, driver: AravisDriverIO, gpio_number: GPIO_NUMBER) -> None:
         self._drv = driver
         self.gpio_number = gpio_number
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return _HIGHEST_POSSIBLE_DEADTIME
 
-    async def arm(
-        self,
-        num: int = 0,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
-        if num == 0:
+    async def prepare(self, trigger_info: TriggerInfo):
+        if (num := trigger_info.number) == 0:
             image_mode = adcore.ImageMode.continuous
         else:
             image_mode = adcore.ImageMode.multiple
-        if exposure is not None:
+        if (exposure := trigger_info.livetime) is not None:
             await self._drv.acquire_time.set(exposure)
 
-        trigger_mode, trigger_source = self._get_trigger_info(trigger)
+        trigger_mode, trigger_source = self._get_trigger_info(trigger_info.trigger)
         # trigger mode must be set first and on it's own!
         await self._drv.trigger_mode.set(trigger_mode)
 
@@ -50,8 +47,12 @@ async def arm(
             self._drv.image_mode.set(image_mode),
         )
 
-        status = await set_and_wait_for_value(self._drv.acquire, True)
-        return status
+    async def arm(self):
+        self._arm_status = await set_and_wait_for_value(self._drv.acquire, True)
+
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
 
     def _get_trigger_info(
         self, trigger: DetectorTrigger
diff --git a/src/ophyd_async/epics/adkinetix/_kinetix_controller.py b/src/ophyd_async/epics/adkinetix/_kinetix_controller.py
index 9469fda68a..70d32e6a78 100644
--- a/src/ophyd_async/epics/adkinetix/_kinetix_controller.py
+++ b/src/ophyd_async/epics/adkinetix/_kinetix_controller.py
@@ -1,7 +1,8 @@
 import asyncio
-from typing import Optional
 
-from ophyd_async.core import AsyncStatus, DetectorControl, DetectorTrigger
+from ophyd_async.core import DetectorControl, DetectorTrigger
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.core._status import AsyncStatus
 from ophyd_async.epics import adcore
 
 from ._kinetix_io import KinetixDriverIO, KinetixTriggerMode
@@ -20,27 +21,31 @@ def __init__(
         driver: KinetixDriverIO,
     ) -> None:
         self._drv = driver
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return 0.001
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
+    async def prepare(self, trigger_info: TriggerInfo):
         await asyncio.gather(
-            self._drv.trigger_mode.set(KINETIX_TRIGGER_MODE_MAP[trigger]),
-            self._drv.num_images.set(num),
+            self._drv.trigger_mode.set(KINETIX_TRIGGER_MODE_MAP[trigger_info.trigger]),
+            self._drv.num_images.set(trigger_info.number),
             self._drv.image_mode.set(adcore.ImageMode.multiple),
         )
-        if exposure is not None and trigger not in [
+        if trigger_info.livetime is not None and trigger_info.trigger not in [
             DetectorTrigger.variable_gate,
             DetectorTrigger.constant_gate,
         ]:
-            await self._drv.acquire_time.set(exposure)
-        return await adcore.start_acquiring_driver_and_ensure_status(self._drv)
+            await self._drv.acquire_time.set(trigger_info.livetime)
+
+    async def arm(self):
+        self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
+            self._drv
+        )
+
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
 
     async def disarm(self):
         await adcore.stop_busy_record(self._drv.acquire, False, timeout=1)
diff --git a/src/ophyd_async/epics/adpilatus/_pilatus_controller.py b/src/ophyd_async/epics/adpilatus/_pilatus_controller.py
index dd48eaf50c..54e0d41d5d 100644
--- a/src/ophyd_async/epics/adpilatus/_pilatus_controller.py
+++ b/src/ophyd_async/epics/adpilatus/_pilatus_controller.py
@@ -1,13 +1,13 @@
 import asyncio
-from typing import Optional
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
-    AsyncStatus,
     DetectorControl,
     DetectorTrigger,
     wait_for_value,
 )
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.core._status import AsyncStatus
 from ophyd_async.epics import adcore
 
 from ._pilatus_io import PilatusDriverIO, PilatusTriggerMode
@@ -27,29 +27,29 @@ def __init__(
     ) -> None:
         self._drv = driver
         self._readout_time = readout_time
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return self._readout_time
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
-        if exposure is not None:
+    async def prepare(self, trigger_info: TriggerInfo):
+        if trigger_info.livetime is not None:
             await adcore.set_exposure_time_and_acquire_period_if_supplied(
-                self, self._drv, exposure
+                self, self._drv, trigger_info.livetime
             )
         await asyncio.gather(
-            self._drv.trigger_mode.set(self._get_trigger_mode(trigger)),
-            self._drv.num_images.set(999_999 if num == 0 else num),
+            self._drv.trigger_mode.set(self._get_trigger_mode(trigger_info.trigger)),
+            self._drv.num_images.set(
+                999_999 if trigger_info.number == 0 else trigger_info.number
+            ),
             self._drv.image_mode.set(adcore.ImageMode.multiple),
         )
 
+    async def arm(self):
         # Standard arm the detector and wait for the acquire PV to be True
-        idle_status = await adcore.start_acquiring_driver_and_ensure_status(self._drv)
-
+        self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
+            self._drv
+        )
         # The pilatus has an additional PV that goes True when the camserver
         # is actually ready. Should wait for that too or we risk dropping
         # a frame
@@ -59,7 +59,9 @@ async def arm(
             timeout=DEFAULT_TIMEOUT,
         )
 
-        return idle_status
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
 
     @classmethod
     def _get_trigger_mode(cls, trigger: DetectorTrigger) -> PilatusTriggerMode:
diff --git a/src/ophyd_async/epics/adsimdetector/_sim_controller.py b/src/ophyd_async/epics/adsimdetector/_sim_controller.py
index 789f89701c..6561ee24f1 100644
--- a/src/ophyd_async/epics/adsimdetector/_sim_controller.py
+++ b/src/ophyd_async/epics/adsimdetector/_sim_controller.py
@@ -1,12 +1,13 @@
 import asyncio
-from typing import Optional, Set
+from typing import Set
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
-    AsyncStatus,
     DetectorControl,
     DetectorTrigger,
 )
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.core._status import AsyncStatus
 from ophyd_async.epics import adcore
 
 
@@ -18,28 +19,33 @@ def __init__(
     ) -> None:
         self.driver = driver
         self.good_states = good_states
+        self.frame_timeout: float
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return 0.002
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
+    async def prepare(self, trigger_info: TriggerInfo):
         assert (
-            trigger == DetectorTrigger.internal
+            trigger_info.trigger == DetectorTrigger.internal
         ), "fly scanning (i.e. external triggering) is not supported for this device"
-        frame_timeout = DEFAULT_TIMEOUT + await self.driver.acquire_time.get_value()
+        self.frame_timeout = (
+            DEFAULT_TIMEOUT + await self.driver.acquire_time.get_value()
+        )
         await asyncio.gather(
-            self.driver.num_images.set(num),
+            self.driver.num_images.set(trigger_info.number),
             self.driver.image_mode.set(adcore.ImageMode.multiple),
         )
-        return await adcore.start_acquiring_driver_and_ensure_status(
-            self.driver, good_states=self.good_states, timeout=frame_timeout
+
+    async def arm(self):
+        self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
+            self.driver, good_states=self.good_states, timeout=self.frame_timeout
         )
 
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
+
     async def disarm(self):
         # We can't use caput callback as we already used it in arm() and we can't have
         # 2 or they will deadlock
diff --git a/src/ophyd_async/epics/advimba/_vimba_controller.py b/src/ophyd_async/epics/advimba/_vimba_controller.py
index fa0232dd2a..9b87d37872 100644
--- a/src/ophyd_async/epics/advimba/_vimba_controller.py
+++ b/src/ophyd_async/epics/advimba/_vimba_controller.py
@@ -1,7 +1,8 @@
 import asyncio
-from typing import Optional
 
-from ophyd_async.core import AsyncStatus, DetectorControl, DetectorTrigger
+from ophyd_async.core import DetectorControl, DetectorTrigger
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.core._status import AsyncStatus
 from ophyd_async.epics import adcore
 
 from ._vimba_io import VimbaDriverIO, VimbaExposeOutMode, VimbaOnOff, VimbaTriggerSource
@@ -27,32 +28,36 @@ def __init__(
         driver: VimbaDriverIO,
     ) -> None:
         self._drv = driver
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return 0.001
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
+    async def prepare(self, trigger_info: TriggerInfo):
         await asyncio.gather(
-            self._drv.trigger_mode.set(TRIGGER_MODE[trigger]),
-            self._drv.exposure_mode.set(EXPOSE_OUT_MODE[trigger]),
-            self._drv.num_images.set(num),
+            self._drv.trigger_mode.set(TRIGGER_MODE[trigger_info.trigger]),
+            self._drv.exposure_mode.set(EXPOSE_OUT_MODE[trigger_info.trigger]),
+            self._drv.num_images.set(trigger_info.number),
             self._drv.image_mode.set(adcore.ImageMode.multiple),
         )
-        if exposure is not None and trigger not in [
+        if trigger_info.livetime is not None and trigger_info.trigger not in [
             DetectorTrigger.variable_gate,
             DetectorTrigger.constant_gate,
         ]:
-            await self._drv.acquire_time.set(exposure)
-        if trigger != DetectorTrigger.internal:
+            await self._drv.acquire_time.set(trigger_info.livetime)
+        if trigger_info.trigger != DetectorTrigger.internal:
             self._drv.trigger_source.set(VimbaTriggerSource.line1)
         else:
             self._drv.trigger_source.set(VimbaTriggerSource.freerun)
-        return await adcore.start_acquiring_driver_and_ensure_status(self._drv)
+
+    async def arm(self):
+        self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
+            self._drv
+        )
+
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
 
     async def disarm(self):
         await adcore.stop_busy_record(self._drv.acquire, False, timeout=1)
diff --git a/src/ophyd_async/epics/eiger/_eiger_controller.py b/src/ophyd_async/epics/eiger/_eiger_controller.py
index fa37bbebaf..c7542bc741 100644
--- a/src/ophyd_async/epics/eiger/_eiger_controller.py
+++ b/src/ophyd_async/epics/eiger/_eiger_controller.py
@@ -1,13 +1,12 @@
 import asyncio
-from typing import Optional
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
-    AsyncStatus,
     DetectorControl,
     DetectorTrigger,
     set_and_wait_for_other_value,
 )
+from ophyd_async.core._detector import TriggerInfo
 
 from ._eiger_io import EigerDriverIO, EigerTriggerMode
 
@@ -37,30 +36,31 @@ async def set_energy(self, energy: float, tolerance: float = 0.1):
         if abs(current_energy - energy) > tolerance:
             await self._drv.photon_energy.set(energy)
 
-    @AsyncStatus.wrap
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ):
+    async def prepare(self, trigger_info: TriggerInfo):
         coros = [
-            self._drv.trigger_mode.set(EIGER_TRIGGER_MODE_MAP[trigger].value),
-            self._drv.num_images.set(num),
+            self._drv.trigger_mode.set(
+                EIGER_TRIGGER_MODE_MAP[trigger_info.trigger].value
+            ),
+            self._drv.num_images.set(trigger_info.number),
         ]
-        if exposure is not None:
+        if trigger_info.livetime is not None:
             coros.extend(
                 [
-                    self._drv.acquire_time.set(exposure),
-                    self._drv.acquire_period.set(exposure),
+                    self._drv.acquire_time.set(trigger_info.livetime),
+                    self._drv.acquire_period.set(trigger_info.livetime),
                 ]
             )
         await asyncio.gather(*coros)
 
+    async def arm(self):
         # TODO: Detector state should be an enum see https://github.com/DiamondLightSource/eiger-fastcs/issues/43
-        await set_and_wait_for_other_value(
+        self._arm_status = set_and_wait_for_other_value(
             self._drv.arm, 1, self._drv.state, "ready", timeout=DEFAULT_TIMEOUT
         )
 
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
+
     async def disarm(self):
         await self._drv.disarm.set(1)
diff --git a/src/ophyd_async/fastcs/panda/_control.py b/src/ophyd_async/fastcs/panda/_control.py
index 08c17df1ef..aeb8e750cd 100644
--- a/src/ophyd_async/fastcs/panda/_control.py
+++ b/src/ophyd_async/fastcs/panda/_control.py
@@ -1,12 +1,12 @@
 import asyncio
-from typing import Optional
 
 from ophyd_async.core import (
-    AsyncStatus,
     DetectorControl,
     DetectorTrigger,
     wait_for_value,
 )
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.core._status import AsyncStatus
 
 from ._block import PcapBlock
 
@@ -14,25 +14,24 @@
 class PandaPcapController(DetectorControl):
     def __init__(self, pcap: PcapBlock) -> None:
         self.pcap = pcap
+        self._arm_status: AsyncStatus | None = None
 
     def get_deadtime(self, exposure: float) -> float:
         return 0.000000008
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.constant_gate,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
-        assert trigger in (
+    async def prepare(self, trigger_info: TriggerInfo):
+        assert trigger_info.trigger in (
             DetectorTrigger.constant_gate,
-            trigger == DetectorTrigger.variable_gate,
+            DetectorTrigger.variable_gate,
         ), "Only constant_gate and variable_gate triggering is supported on the PandA"
-        await asyncio.gather(self.pcap.arm.set(True))
+
+    async def arm(self):
+        self._arm_status = self.pcap.arm.set(True)
         await wait_for_value(self.pcap.active, True, timeout=1)
-        return AsyncStatus(wait_for_value(self.pcap.active, False, timeout=None))
 
-    async def disarm(self) -> AsyncStatus:
+    async def wait_for_idle(self):
+        pass
+
+    async def disarm(self):
         await asyncio.gather(self.pcap.arm.set(False))
         await wait_for_value(self.pcap.active, False, timeout=1)
-        return AsyncStatus(wait_for_value(self.pcap.active, False, timeout=None))
diff --git a/src/ophyd_async/plan_stubs/_fly.py b/src/ophyd_async/plan_stubs/_fly.py
index daa686b477..2cf6f5499e 100644
--- a/src/ophyd_async/plan_stubs/_fly.py
+++ b/src/ophyd_async/plan_stubs/_fly.py
@@ -46,6 +46,7 @@ def prepare_static_seq_table_flyer_and_detectors_with_same_trigger(
     repeats: int = 1,
     period: float = 0.0,
     frame_timeout: Optional[float] = None,
+    iteration: int = 1,
 ):
     """Prepare a hardware triggered flyable and one or more detectors.
 
@@ -68,6 +69,7 @@ def prepare_static_seq_table_flyer_and_detectors_with_same_trigger(
         deadtime=deadtime,
         livetime=exposure,
         frame_timeout=frame_timeout,
+        iteration=iteration,
     )
     trigger_time = number_of_frames * (exposure + deadtime)
     pre_delay = max(period - 2 * shutter_time - trigger_time, 0)
diff --git a/src/ophyd_async/sim/demo/_pattern_detector/_pattern_detector_controller.py b/src/ophyd_async/sim/demo/_pattern_detector/_pattern_detector_controller.py
index 064bea873f..039ddb066c 100644
--- a/src/ophyd_async/sim/demo/_pattern_detector/_pattern_detector_controller.py
+++ b/src/ophyd_async/sim/demo/_pattern_detector/_pattern_detector_controller.py
@@ -1,7 +1,10 @@
 import asyncio
 from typing import Optional
 
-from ophyd_async.core import AsyncStatus, DetectorControl, DetectorTrigger, PathProvider
+from pydantic import Field
+
+from ophyd_async.core import DetectorControl, PathProvider
+from ophyd_async.core._detector import TriggerInfo
 
 from ._pattern_generator import PatternGenerator
 
@@ -11,30 +14,36 @@ def __init__(
         self,
         pattern_generator: PatternGenerator,
         path_provider: PathProvider,
-        exposure: float = 0.1,
+        exposure: float = Field(default=0.1),
     ) -> None:
         self.pattern_generator: PatternGenerator = pattern_generator
-        if exposure is None:
-            exposure = 0.1
         self.pattern_generator.set_exposure(exposure)
         self.path_provider: PathProvider = path_provider
         self.task: Optional[asyncio.Task] = None
         super().__init__()
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = 0.01,
-    ) -> AsyncStatus:
-        if exposure is None:
-            exposure = 0.1
-        period: float = exposure + self.get_deadtime(exposure)
-        task = asyncio.create_task(
-            self._coroutine_for_image_writing(exposure, period, num)
+    async def prepare(
+        self, trigger_info: TriggerInfo = TriggerInfo(number=1, livetime=0.01)
+    ):
+        self._trigger_info = trigger_info
+        if self._trigger_info.livetime is None:
+            self._trigger_info.livetime = 0.01
+        self.period: float = self._trigger_info.livetime + self.get_deadtime(
+            trigger_info.livetime
         )
-        self.task = task
-        return AsyncStatus(task)
+
+    async def arm(self):
+        assert self._trigger_info.livetime
+        assert self.period
+        self.task = asyncio.create_task(
+            self._coroutine_for_image_writing(
+                self._trigger_info.livetime, self.period, self._trigger_info.number
+            )
+        )
+
+    async def wait_for_idle(self):
+        if self.task:
+            await self.task
 
     async def disarm(self):
         if self.task:
diff --git a/tests/core/test_flyer.py b/tests/core/test_flyer.py
index 9d968526d9..6d9c9142aa 100644
--- a/tests/core/test_flyer.py
+++ b/tests/core/test_flyer.py
@@ -117,7 +117,7 @@ async def detectors(RE: RunEngine) -> tuple[StandardDetector, StandardDetector]:
     await writers[0].dummy_signal.connect(mock=True)
     await writers[1].dummy_signal.connect(mock=True)
 
-    async def dummy_arm_1(self=None, trigger=None, num=0, exposure=None):
+    def dummy_arm_1(self=None, trigger=None, num=0, exposure=None):
         return writers[0].dummy_signal.set(1)
 
     async def dummy_arm_2(self=None, trigger=None, num=0, exposure=None):
diff --git a/tests/core/test_signal.py b/tests/core/test_signal.py
index ab5c02cffe..d498e67531 100644
--- a/tests/core/test_signal.py
+++ b/tests/core/test_signal.py
@@ -240,7 +240,7 @@ async def test_wait_for_value_with_value():
     await signal.set("blah")
 
     with pytest.raises(
-        TimeoutError,
+        asyncio.TimeoutError,
         match="signal didn't match 'something' in 0.1s, last value 'blah'",
     ):
         await wait_for_value(signal, "something", timeout=0.1)
@@ -263,7 +263,7 @@ def less_than_42(v):
         return v < 42
 
     with pytest.raises(
-        TimeoutError,
+        asyncio.TimeoutError,
         match="signal didn't match less_than_42 in 0.1s, last value 45.8",
     ):
         await wait_for_value(signal, less_than_42, timeout=0.1)
diff --git a/tests/epics/adaravis/test_aravis.py b/tests/epics/adaravis/test_aravis.py
index efefae2eb2..3c34fa49eb 100644
--- a/tests/epics/adaravis/test_aravis.py
+++ b/tests/epics/adaravis/test_aravis.py
@@ -36,7 +36,15 @@ async def test_trigger_source_set_to_gpio_line(test_adaravis: adaravis.AravisDet
     set_mock_value(test_adaravis.drv.trigger_source, "Freerun")
 
     async def trigger_and_complete():
-        await test_adaravis.controller.arm(num=1, trigger=DetectorTrigger.edge_trigger)
+        await test_adaravis.controller.prepare(
+            TriggerInfo(
+                number=1,
+                trigger=DetectorTrigger.edge_trigger,
+                livetime=None,
+                deadtime=None,
+                frame_timeout=None,
+            )
+        )
         # Prevent timeouts
         set_mock_value(test_adaravis.drv.acquire, True)
 
@@ -158,7 +166,7 @@ async def test_unsupported_trigger_excepts(test_adaravis: adaravis.AravisDetecto
     ):
         await test_adaravis.prepare(
             TriggerInfo(
-                number=1,
+                number=0,
                 trigger=DetectorTrigger.variable_gate,
                 deadtime=1,
                 livetime=1,
diff --git a/tests/epics/adcore/test_scans.py b/tests/epics/adcore/test_scans.py
index 3936885083..6889fbbe3b 100644
--- a/tests/epics/adcore/test_scans.py
+++ b/tests/epics/adcore/test_scans.py
@@ -37,14 +37,14 @@ async def stop(self): ...
 
 class DummyController(DetectorControl):
     def __init__(self) -> None: ...
+    async def prepare(self, trigger_info: TriggerInfo):
+        return AsyncStatus(asyncio.sleep(0.01))
 
-    async def arm(
-        self,
-        num: int,
-        trigger: DetectorTrigger = DetectorTrigger.internal,
-        exposure: Optional[float] = None,
-    ) -> AsyncStatus:
-        return AsyncStatus(asyncio.sleep(0.1))
+    async def arm(self):
+        self._arm_status = AsyncStatus(asyncio.sleep(0.01))
+
+    async def wait_for_idle(self):
+        await self._arm_status
 
     async def disarm(self): ...
 
diff --git a/tests/epics/adkinetix/test_kinetix.py b/tests/epics/adkinetix/test_kinetix.py
index 3a53091f40..a17be5e5b3 100644
--- a/tests/epics/adkinetix/test_kinetix.py
+++ b/tests/epics/adkinetix/test_kinetix.py
@@ -33,7 +33,11 @@ async def test_trigger_modes(test_adkinetix: adkinetix.KinetixDetector):
     set_mock_value(test_adkinetix.drv.trigger_mode, "Internal")
 
     async def setup_trigger_mode(trig_mode: DetectorTrigger):
-        await test_adkinetix.controller.arm(num=1, trigger=trig_mode)
+        await test_adkinetix.controller.prepare(
+            TriggerInfo(number=1, trigger=trig_mode)
+        )
+        await test_adkinetix.controller.arm()
+        await test_adkinetix.controller.wait_for_idle()
         # Prevent timeouts
         set_mock_value(test_adkinetix.drv.acquire, True)
 
diff --git a/tests/epics/adpilatus/test_pilatus.py b/tests/epics/adpilatus/test_pilatus.py
index 00b3c119a8..72145ff0cc 100644
--- a/tests/epics/adpilatus/test_pilatus.py
+++ b/tests/epics/adpilatus/test_pilatus.py
@@ -1,3 +1,4 @@
+import asyncio
 from typing import Awaitable, Callable
 from unittest.mock import patch
 
@@ -61,11 +62,11 @@ async def test_trigger_mode_set(
 ):
     async def trigger_and_complete():
         set_mock_value(test_adpilatus.drv.armed, True)
-        status = await test_adpilatus.controller.arm(
-            num=1,
-            trigger=detector_trigger,
+        await test_adpilatus.controller.prepare(
+            TriggerInfo(number=1, trigger=detector_trigger)
         )
-        await status
+        await test_adpilatus.controller.arm()
+        await test_adpilatus.controller.wait_for_idle()
 
     await _trigger(test_adpilatus, expected_trigger_mode, trigger_and_complete)
 
@@ -74,17 +75,17 @@ async def test_trigger_mode_set_without_armed_pv(
     test_adpilatus: adpilatus.PilatusDetector,
 ):
     async def trigger_and_complete():
-        status = await test_adpilatus.controller.arm(
-            num=1,
-            trigger=DetectorTrigger.internal,
+        await test_adpilatus.controller.prepare(
+            TriggerInfo(number=1, trigger=DetectorTrigger.internal)
         )
-        await status
+        await test_adpilatus.controller.arm()
+        await test_adpilatus.controller.wait_for_idle()
 
     with patch(
         "ophyd_async.epics.adpilatus._pilatus_controller.DEFAULT_TIMEOUT",
         0.1,
     ):
-        with pytest.raises(TimeoutError):
+        with pytest.raises(asyncio.TimeoutError):
             await _trigger(
                 test_adpilatus,
                 adpilatus.PilatusTriggerMode.internal,
diff --git a/tests/epics/adpilatus/test_pilatus_controller.py b/tests/epics/adpilatus/test_pilatus_controller.py
index 00194fdb94..bb825b8744 100644
--- a/tests/epics/adpilatus/test_pilatus_controller.py
+++ b/tests/epics/adpilatus/test_pilatus_controller.py
@@ -1,6 +1,7 @@
 import pytest
 
 from ophyd_async.core import DetectorTrigger, DeviceCollector, set_mock_value
+from ophyd_async.core._detector import TriggerInfo
 from ophyd_async.epics import adcore, adpilatus
 
 
@@ -28,8 +29,9 @@ async def test_pilatus_controller(
     pilatus_driver: adpilatus.PilatusDriverIO,
 ):
     set_mock_value(pilatus_driver.armed, True)
-    status = await pilatus.arm(num=1, trigger=DetectorTrigger.constant_gate)
-    await status
+    await pilatus.prepare(TriggerInfo(number=1, trigger=DetectorTrigger.constant_gate))
+    await pilatus.arm()
+    await pilatus.wait_for_idle()
 
     assert await pilatus_driver.num_images.get_value() == 1
     assert await pilatus_driver.image_mode.get_value() == adcore.ImageMode.multiple
diff --git a/tests/epics/adsimdetector/test_adsim_controller.py b/tests/epics/adsimdetector/test_adsim_controller.py
index 8a7c33516b..64d53ce590 100644
--- a/tests/epics/adsimdetector/test_adsim_controller.py
+++ b/tests/epics/adsimdetector/test_adsim_controller.py
@@ -1,6 +1,9 @@
+from unittest.mock import patch
+
 import pytest
 
 from ophyd_async.core import DeviceCollector
+from ophyd_async.core._detector import DetectorTrigger, TriggerInfo
 from ophyd_async.epics import adcore, adsimdetector
 
 
@@ -14,7 +17,10 @@ async def ad(RE) -> adsimdetector.SimController:
 
 
 async def test_ad_controller(RE, ad: adsimdetector.SimController):
-    await ad.arm(num=1)
+    with patch("ophyd_async.core._signal.wait_for_value", return_value=None):
+        await ad.prepare(TriggerInfo(number=1, trigger=DetectorTrigger.internal))
+        await ad.arm()
+        await ad.wait_for_idle()
 
     driver = ad.driver
     assert await driver.num_images.get_value() == 1
diff --git a/tests/epics/adsimdetector/test_sim.py b/tests/epics/adsimdetector/test_sim.py
index 149835aa49..891d89c33c 100644
--- a/tests/epics/adsimdetector/test_sim.py
+++ b/tests/epics/adsimdetector/test_sim.py
@@ -114,9 +114,6 @@ async def test_two_detectors_fly_different_rate(
     trigger_info = TriggerInfo(
         number=15,
         trigger=DetectorTrigger.internal,
-        deadtime=None,
-        livetime=None,
-        frame_timeout=None,
     )
     docs = defaultdict(list)
 
diff --git a/tests/epics/advimba/test_vimba.py b/tests/epics/advimba/test_vimba.py
index 0bc32b887b..ec93cc07d3 100644
--- a/tests/epics/advimba/test_vimba.py
+++ b/tests/epics/advimba/test_vimba.py
@@ -35,7 +35,9 @@ async def test_arming_trig_modes(test_advimba: advimba.VimbaDetector):
     set_mock_value(test_advimba.drv.exposure_mode, "Timed")
 
     async def setup_trigger_mode(trig_mode: DetectorTrigger):
-        await test_advimba.controller.arm(num=1, trigger=trig_mode)
+        await test_advimba.controller.prepare(TriggerInfo(number=1, trigger=trig_mode))
+        await test_advimba.controller.arm()
+        await test_advimba.controller.wait_for_idle()
         # Prevent timeouts
         set_mock_value(test_advimba.drv.acquire, True)
 
diff --git a/tests/epics/eiger/test_eiger_controller.py b/tests/epics/eiger/test_eiger_controller.py
index 70f3078163..39204142e5 100644
--- a/tests/epics/eiger/test_eiger_controller.py
+++ b/tests/epics/eiger/test_eiger_controller.py
@@ -8,6 +8,7 @@
     get_mock_put,
     set_mock_value,
 )
+from ophyd_async.core._detector import TriggerInfo
 from ophyd_async.epics.eiger._eiger_controller import EigerController
 from ophyd_async.epics.eiger._eiger_io import EigerDriverIO
 
@@ -43,7 +44,9 @@ async def test_when_arm_with_exposure_then_time_and_period_set(
 ):
     driver, controller = eiger_driver_and_controller
     test_exposure = 0.002
-    await controller.arm(10, exposure=test_exposure)
+    await controller.prepare(TriggerInfo(number=10, livetime=test_exposure))
+    await controller.arm()
+    await controller.wait_for_idle()
     assert (await driver.acquire_period.get_value()) == test_exposure
     assert (await driver.acquire_time.get_value()) == test_exposure
 
@@ -52,7 +55,9 @@ async def test_when_arm_with_no_exposure_then_arm_set_correctly(
     eiger_driver_and_controller: DriverAndController,
 ):
     driver, controller = eiger_driver_and_controller
-    await controller.arm(10, exposure=None)
+    await controller.prepare(TriggerInfo(number=10))
+    await controller.arm()
+    await controller.wait_for_idle()
     get_mock_put(driver.arm).assert_called_once_with(1, wait=ANY, timeout=ANY)
 
 
@@ -61,7 +66,9 @@ async def test_when_arm_with_number_of_images_then_number_of_images_set_correctl
 ):
     driver, controller = eiger_driver_and_controller
     test_number_of_images = 40
-    await controller.arm(test_number_of_images, exposure=None)
+    await controller.prepare(TriggerInfo(number=test_number_of_images))
+    await controller.arm()
+    await controller.wait_for_idle()
     get_mock_put(driver.num_images).assert_called_once_with(
         test_number_of_images, wait=ANY, timeout=ANY
     )
@@ -73,7 +80,9 @@ async def test_given_detector_fails_to_go_ready_when_arm_called_then_fails(
 ):
     driver, controller = eiger_driver_and_controller_no_arm
     with raises(TimeoutError):
-        await controller.arm(10)
+        await controller.prepare(TriggerInfo(number=10))
+        await controller.arm()
+        await controller.wait_for_idle()
 
 
 async def test_when_disarm_called_on_controller_then_disarm_called_on_driver(
diff --git a/tests/fastcs/panda/test_hdf_panda.py b/tests/fastcs/panda/test_hdf_panda.py
index 43b969ab5e..ab5acd5c20 100644
--- a/tests/fastcs/panda/test_hdf_panda.py
+++ b/tests/fastcs/panda/test_hdf_panda.py
@@ -2,28 +2,28 @@
 from typing import Dict
 from unittest.mock import ANY
 
+import bluesky.plan_stubs as bps
 import numpy as np
 import pytest
-from bluesky import plan_stubs as bps
-from bluesky.run_engine import RunEngine
+from bluesky import RunEngine
 
 from ophyd_async.core import (
     Device,
     SignalR,
-    StandardFlyer,
     StaticFilenameProvider,
     StaticPathProvider,
-    assert_emitted,
     callback_on_mock_put,
     set_mock_value,
 )
+from ophyd_async.core._flyer import StandardFlyer
+from ophyd_async.core._signal import assert_emitted
 from ophyd_async.fastcs.panda import (
     DatasetTable,
     HDFPanda,
     PandaHdf5DatasetType,
-    StaticSeqTableTriggerLogic,
 )
-from ophyd_async.plan_stubs import (
+from ophyd_async.fastcs.panda._trigger import StaticSeqTableTriggerLogic
+from ophyd_async.plan_stubs._fly import (
     prepare_static_seq_table_flyer_and_detectors_with_same_trigger,
 )
 
@@ -111,7 +111,7 @@ def flying_plan():
         set_mock_value(flyer.trigger_logic.seq.active, 1)
 
         yield from bps.kickoff(flyer, wait=True)
-        yield from bps.kickoff(mock_hdf_panda)
+        yield from bps.kickoff(mock_hdf_panda, wait=True)
 
         yield from bps.complete(flyer, wait=False, group="complete")
         yield from bps.complete(mock_hdf_panda, wait=False, group="complete")
@@ -191,3 +191,123 @@ def assert_resource_document():
         assert stream_datum["stream_resource"] in [
             sd["uid"].split("/")[0] for sd in docs["stream_datum"]
         ]
+
+
+async def test_hdf_panda_hardware_triggered_flyable_with_iterations(
+    RE: RunEngine,
+    mock_hdf_panda,
+    tmp_path,
+):
+    docs = {}
+
+    def append_and_print(name, doc):
+        if name not in docs:
+            docs[name] = []
+        docs[name] += [doc]
+
+    RE.subscribe(append_and_print)
+
+    shutter_time = 0.004
+    exposure = 1
+
+    trigger_logic = StaticSeqTableTriggerLogic(mock_hdf_panda.seq[1])
+    flyer = StandardFlyer(trigger_logic, [], name="flyer")
+
+    def flying_plan():
+        iteration = 2
+        yield from bps.stage_all(mock_hdf_panda, flyer)
+        yield from bps.open_run()
+        yield from prepare_static_seq_table_flyer_and_detectors_with_same_trigger(  # noqa: E501
+            flyer,
+            [mock_hdf_panda],
+            number_of_frames=1,
+            exposure=exposure,
+            shutter_time=shutter_time,
+            iteration=iteration,
+        )
+
+        yield from bps.declare_stream(mock_hdf_panda, name="main_stream", collect=True)
+
+        for i in range(iteration):
+            set_mock_value(flyer.trigger_logic.seq.active, 1)
+            yield from bps.kickoff(flyer, wait=True)
+            yield from bps.kickoff(mock_hdf_panda)
+
+            yield from bps.complete(flyer, wait=False, group="complete")
+            yield from bps.complete(mock_hdf_panda, wait=False, group="complete")
+
+            # Manually incremenet the index as if a frame was taken
+            set_mock_value(mock_hdf_panda.data.num_captured, 1)
+            set_mock_value(flyer.trigger_logic.seq.active, 0)
+
+            done = False
+            while not done:
+                try:
+                    yield from bps.wait(group="complete", timeout=0.5)
+                except TimeoutError:
+                    pass
+                else:
+                    done = True
+                yield from bps.collect(
+                    mock_hdf_panda,
+                    return_payload=False,
+                    name="main_stream",
+                )
+            yield from bps.wait(group="complete")
+        yield from bps.close_run()
+
+        yield from bps.unstage_all(flyer, mock_hdf_panda)
+        yield from bps.wait_for([lambda: mock_hdf_panda.controller.disarm()])
+
+    # fly scan
+    RE(flying_plan())
+
+    assert_emitted(
+        docs, start=1, descriptor=1, stream_resource=2, stream_datum=2, stop=1
+    )
+
+    # test descriptor
+    data_key_names: Dict[str, str] = docs["descriptor"][0]["object_keys"]["panda"]
+    assert data_key_names == ["x", "y"]
+    for data_key_name in data_key_names:
+        assert (
+            docs["descriptor"][0]["data_keys"][data_key_name]["source"]
+            == "mock+soft://panda-data-hdf_directory"
+        )
+
+    # test stream resources
+    for dataset_name, stream_resource, data_key_name in zip(
+        ("x", "y"), docs["stream_resource"], data_key_names
+    ):
+
+        def assert_resource_document():
+            assert stream_resource == {
+                "run_start": docs["start"][0]["uid"],
+                "uid": ANY,
+                "data_key": data_key_name,
+                "mimetype": "application/x-hdf5",
+                "uri": "file://localhost" + str(tmp_path / "test-panda.h5"),
+                "parameters": {
+                    "dataset": f"/{dataset_name}",
+                    "swmr": False,
+                    "multiplier": 1,
+                },
+            }
+            assert "test-panda.h5" in stream_resource["uri"]
+
+        assert_resource_document()
+
+    # test stream datum
+    for stream_datum in docs["stream_datum"]:
+        assert stream_datum["descriptor"] == docs["descriptor"][0]["uid"]
+        assert stream_datum["seq_nums"] == {
+            "start": 1,
+            "stop": 2,
+        }
+        assert stream_datum["indices"] == {
+            "start": 0,
+            "stop": 1,
+        }
+        assert stream_datum["stream_resource"] in [
+            sd["uid"].split("/")[0] for sd in docs["stream_datum"]
+        ]
diff --git a/tests/fastcs/panda/test_panda_control.py b/tests/fastcs/panda/test_panda_control.py
index b73d907f89..6c0abf298f 100644
--- a/tests/fastcs/panda/test_panda_control.py
+++ b/tests/fastcs/panda/test_panda_control.py
@@ -5,6 +5,7 @@
 import pytest
 
 from ophyd_async.core import DEFAULT_TIMEOUT, DetectorTrigger, Device, DeviceCollector
+from ophyd_async.core._detector import TriggerInfo
 from ophyd_async.epics.pvi import fill_pvi_entries
 from ophyd_async.epics.signal import epics_signal_rw
 from ophyd_async.fastcs.panda import CommonPandaBlocks, PandaPcapController
@@ -36,18 +37,27 @@ class PcapBlock(Device):
     pandaController = PandaPcapController(pcap=PcapBlock())
     with patch("ophyd_async.fastcs.panda._control.wait_for_value", return_value=None):
         with pytest.raises(AttributeError) as exc:
-            await pandaController.arm(num=1, trigger=DetectorTrigger.constant_gate)
+            await pandaController.prepare(
+                TriggerInfo(number=1, trigger=DetectorTrigger.constant_gate)
+            )
+            await pandaController.arm()
     assert ("'PcapBlock' object has no attribute 'arm'") in str(exc.value)
 
 
 async def test_panda_controller_arm_disarm(mock_panda):
     pandaController = PandaPcapController(mock_panda.pcap)
     with patch("ophyd_async.fastcs.panda._control.wait_for_value", return_value=None):
-        await pandaController.arm(num=1, trigger=DetectorTrigger.constant_gate)
+        await pandaController.prepare(
+            TriggerInfo(number=1, trigger=DetectorTrigger.constant_gate)
+        )
+        await pandaController.arm()
+        await pandaController.wait_for_idle()
     await pandaController.disarm()
 
 
 async def test_panda_controller_wrong_trigger():
     pandaController = PandaPcapController(None)
     with pytest.raises(AssertionError):
-        await pandaController.arm(num=1, trigger=DetectorTrigger.internal)
+        await pandaController.prepare(
+            TriggerInfo(number=1, trigger=DetectorTrigger.internal)
+        )
diff --git a/tests/fastcs/panda/test_writer.py b/tests/fastcs/panda/test_writer.py
index 7bdca1a31b..dc26787cbf 100644
--- a/tests/fastcs/panda/test_writer.py
+++ b/tests/fastcs/panda/test_writer.py
@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 from pathlib import Path
@@ -187,7 +188,7 @@ async def test_wait_for_index(mock_writer: PandaHDFWriter):
     set_mock_value(mock_writer.panda_data_block.num_captured, 3)
     await mock_writer.wait_for_index(3, timeout=1)
     set_mock_value(mock_writer.panda_data_block.num_captured, 2)
-    with pytest.raises(TimeoutError):
+    with pytest.raises(asyncio.TimeoutError):
         await mock_writer.wait_for_index(3, timeout=0.1)
 
 
diff --git a/tests/plan_stubs/test_fly.py b/tests/plan_stubs/test_fly.py
index 791935c111..06e40e29db 100644
--- a/tests/plan_stubs/test_fly.py
+++ b/tests/plan_stubs/test_fly.py
@@ -115,7 +115,6 @@ def __init__(
 
     @WatchableAsyncStatus.wrap
     async def complete(self):
-        assert self._arm_status, "Prepare not run"
         assert self._trigger_info
         self.writer.increment_index()
         async for index in self.writer.observe_indices_written(
@@ -145,10 +144,10 @@ async def detectors(RE: RunEngine) -> tuple[MockDetector, MockDetector]:
     await writers[0].dummy_signal.connect(mock=True)
     await writers[1].dummy_signal.connect(mock=True)
 
-    async def dummy_arm_1(self=None, trigger=None, num=0, exposure=None):
+    def dummy_arm_1(self=None):
         return writers[0].dummy_signal.set(1)
 
-    async def dummy_arm_2(self=None, trigger=None, num=0, exposure=None):
+    def dummy_arm_2(self=None):
         return writers[1].dummy_signal.set(1)
 
     detector_1 = MockDetector(