Source code for ophyd_async.epics.adkinetix._kinetix_controller
import asyncio
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
KINETIX_TRIGGER_MODE_MAP = {
DetectorTrigger.internal: KinetixTriggerMode.internal,
DetectorTrigger.constant_gate: KinetixTriggerMode.gate,
DetectorTrigger.variable_gate: KinetixTriggerMode.gate,
DetectorTrigger.edge_trigger: KinetixTriggerMode.edge,
}
[docs]
class KinetixController(DetectorControl):
def __init__(
self,
driver: KinetixDriverIO,
) -> None:
self._drv = driver
self._arm_status: AsyncStatus | None = None
[docs]
def get_deadtime(self, exposure: float) -> float:
return 0.001
[docs]
async def prepare(self, trigger_info: TriggerInfo):
await asyncio.gather(
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 trigger_info.livetime is not None and trigger_info.trigger not in [
DetectorTrigger.variable_gate,
DetectorTrigger.constant_gate,
]:
await self._drv.acquire_time.set(trigger_info.livetime)
[docs]
async def arm(self):
self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
self._drv
)
[docs]
async def wait_for_idle(self):
if self._arm_status:
await self._arm_status
[docs]
async def disarm(self):
await adcore.stop_busy_record(self._drv.acquire, False, timeout=1)