Added BaseDetector and StepDetector#1062
Conversation
|
@coretl as we have decided to use Should I remove the tests, or should I change it to Or is this check not necessary and we just remove those tests all together? Is there not a simpler way to check something is connected like: if not signal.connected:
raise Error(...) |
…between detectors
|
@coretl I've done the core logic for this now and also add some comments to get some feedback on to see if you agree. I haven't added any new tests yet as I want to get feedback on design first before I commit to writing tests. |
coretl
left a comment
There was a problem hiding this comment.
Please can we discuss ReadableDetector in person. I'm pretty convinced this should share more with StandardReadable than with StandardDetector.
There was a problem hiding this comment.
Ok, having seen this, I'm more convinced that I prefer the ReadableDetector approach. With a few tweaks we can make it look more readable:
# StandardReadable mechanisms for read, describe, read_configuration, describe_configuration, stage, unstage
# Defer to the controller for arm and prepare
class ReadableDetector(StandardReadable, Triggerable):
def __init__(
self,
controller: DetectorController,
name: str = "",
):
self._controller = controller
self._trigger_info: TriggerInfo | None = None
# Hook in the controller functions, requires change
# to to StandardDetector
self._stage_funcs += (self._controller.disarm,)
self._unstage_funcs += (self._controller.disarm,)
super().__init__(name=name)
@AsyncStatus.wrap
async def prepare(self, value: TriggerInfo) -> None:
self._trigger_info = await self._controller.prepare(value)
@AsyncStatus.wrap
async def trigger(self) -> None:
if self._trigger_info is None:
await self.prepare(TriggerInfo())
await self._controller.arm()
await self._controller.wait_for_idle()
class MyReadableDetector(ReadableDetector):
def __init__(self, prefix: str, name=""):
self.drv = MyDetectorDriver(prefix + "DRV:")
self.array = NDStdArray(prefix + "IMAGE:")
# Add the detector config to read_configuration()
self.add_readables([self.drv.acquire_time, self.drv.gain], Format.CONFIG_SIGNAL)
# Add the array data to read()
self.add_readables(self.array.array_data, Format.HINTED_SIGNAL)
super().__init__(name=name, controller=MyController())There was a problem hiding this comment.
Okay, as discussed IRL, I have added an initial attempt at this. We have created a _ContextController class for handling the stage/unstage of a controller, but I've gone a bit further and also made it handle prepare and trigger. Let me know what you think of this implementation.
| class _ControllerContext(Device, Stageable, Triggerable, Preparable): | ||
| """Controller and trigger_info as a wrapped device.""" | ||
|
|
||
| def __init__(self, controller: DetectorController, name: str = ""): | ||
| self.controller = controller | ||
| self.trigger_info: TriggerInfo | None = None | ||
| super().__init__(name) | ||
|
|
||
| @AsyncStatus.wrap | ||
| async def stage(self) -> None: | ||
| await self.controller.disarm() | ||
|
|
||
| @AsyncStatus.wrap | ||
| async def unstage(self) -> None: | ||
| await self.controller.disarm() | ||
| self.trigger_info = None | ||
|
|
||
| @AsyncStatus.wrap | ||
| async def prepare(self, value: TriggerInfo) -> None: | ||
| await self.controller.prepare(value) | ||
| self.trigger_info = value | ||
|
|
||
| @AsyncStatus.wrap | ||
| async def trigger(self) -> None: | ||
| await self.controller.arm() | ||
| await self.controller.wait_for_idle() |
There was a problem hiding this comment.
I like the encapsulation of prepare, but I think you could suck in ensure_trigger_info in as well and the software triggering part too. How about:
| class _ControllerContext(Device, Stageable, Triggerable, Preparable): | |
| """Controller and trigger_info as a wrapped device.""" | |
| def __init__(self, controller: DetectorController, name: str = ""): | |
| self.controller = controller | |
| self.trigger_info: TriggerInfo | None = None | |
| super().__init__(name) | |
| @AsyncStatus.wrap | |
| async def stage(self) -> None: | |
| await self.controller.disarm() | |
| @AsyncStatus.wrap | |
| async def unstage(self) -> None: | |
| await self.controller.disarm() | |
| self.trigger_info = None | |
| @AsyncStatus.wrap | |
| async def prepare(self, value: TriggerInfo) -> None: | |
| await self.controller.prepare(value) | |
| self.trigger_info = value | |
| @AsyncStatus.wrap | |
| async def trigger(self) -> None: | |
| await self.controller.arm() | |
| await self.controller.wait_for_idle() | |
| class _ControllerContext(Device, Stageable, Triggerable, Preparable): | |
| """Controller and trigger_info as a wrapped device.""" | |
| def __init__(self, controller: DetectorController, name: str = ""): | |
| self._controller = controller | |
| self._trigger_info: TriggerInfo | None = None | |
| super().__init__(name) | |
| @AsyncStatus.wrap | |
| async def stage(self) -> None: | |
| await self._controller.disarm() | |
| self._trigger_info = None | |
| unstage = stage | |
| @property | |
| def trigger_info(self) -> TriggerInfo: | |
| """The TriggerInfo the detector has been prepared with, raising RuntimeError if prepare not called.""" | |
| if self._trigger_info is None: | |
| raise RuntimeError("Detector has not been prepared with a TriggerInfo.") | |
| return self._trigger_info | |
| @AsyncStatus.wrap | |
| async def prepare(self, value: TriggerInfo) -> None: | |
| await self._controller.prepare(value) | |
| self._trigger_info = value | |
| @AsyncStatus.wrap | |
| async def trigger(self) -> None: | |
| if self._trigger_info is None: | |
| await self.prepare(TriggerInfo()) | |
| await self.controller.arm() | |
| await self.controller.wait_for_idle() |
|
Superseded by #1177 |
Fixes #1060