Make HDFPandA StandardDetector (#185)

made a new class for the PandA hdf writer and adjusted `PandAController`

src/ophyd_async/core/_providers.py

@@ -39,8 +39,10 @@ def __init__(
  directory_path: Union[str, Path],
  filename_prefix: str = "",
  filename_suffix: str = "",
- resource_dir: Path = Path("."),
+ resource_dir: Optional[Path] = None,
  ) -> None:
+ if resource_dir is None:
+ resource_dir = Path(".")
  if isinstance(directory_path, str):
  directory_path = Path(directory_path)
  self._directory_info = DirectoryInfo(

src/ophyd_async/epics/pvi/__init__.py

@@ -1,3 +1,3 @@
-from .pvi import PVIEntry, fill_pvi_entries
+from .pvi import PVIEntry, create_children_from_annotations, fill_pvi_entries
 
-__all__ = ["PVIEntry", "fill_pvi_entries"]
+__all__ = ["PVIEntry", "fill_pvi_entries", "create_children_from_annotations"]

src/ophyd_async/epics/pvi/pvi.py

@@ -1,5 +1,6 @@
 import re
 from dataclasses import dataclass
+from inspect import isclass
 from typing import (
  Any,
  Callable,
@@ -56,13 +57,14 @@ def _split_subscript(tp: T) -> Union[Tuple[Any, Tuple[Any]], Tuple[T, None]]:
  return tp, None
 
 
-def _strip_union(field: Union[Union[T], T]) -> T:
+def _strip_union(field: Union[Union[T], T]) -> Tuple[T, bool]:
  if get_origin(field) is Union:
  args = get_args(field)
+ is_optional = type(None) in args
  for arg in args:
  if arg is not type(None):
- return arg
- return field
+ return arg, is_optional
+ return field, False
 
 
 def _strip_device_vector(field: Union[Type[Device]]) -> Tuple[bool, Type[Device]]:
@@ -92,9 +94,10 @@ def _verify_common_blocks(entry: PVIEntry, common_device: Type[Device]):
  if sub_name in ("_name", "parent"):
  continue
  assert entry.sub_entries
- if sub_name not in entry.sub_entries and get_origin(sub_device) is not Optional:
+ device_t, is_optional = _strip_union(sub_device)
+ if sub_name not in entry.sub_entries and not is_optional:
  raise RuntimeError(
- f"sub device `{sub_name}:{type(sub_device)}` was not provided by pvi"
+ f"sub device `{sub_name}:{type(sub_device)}` " "was not provided by pvi"
  )
  if isinstance(entry.sub_entries[sub_name], dict):
  for sub_sub_entry in entry.sub_entries[sub_name].values(): # type: ignore
@@ -128,7 +131,7 @@ def _parse_type(
 ):
  if common_device_type:
  # pre-defined type
- device_cls = _strip_union(common_device_type)
+ device_cls, _ = _strip_union(common_device_type)
  is_device_vector, device_cls = _strip_device_vector(device_cls)
  device_cls, device_args = _split_subscript(device_cls)
  assert issubclass(device_cls, Device)
@@ -162,7 +165,7 @@ def _sim_common_blocks(device: Device, stripped_type: Optional[Type] = None):
  )
 
  for device_name, device_cls in sub_devices:
- device_cls = _strip_union(device_cls)
+ device_cls, _ = _strip_union(device_cls)
  is_device_vector, device_cls = _strip_device_vector(device_cls)
  device_cls, device_args = _split_subscript(device_cls)
  assert issubclass(device_cls, Device)
@@ -187,8 +190,7 @@ def _sim_common_blocks(device: Device, stripped_type: Optional[Type] = None):
  if is_signal:
  sub_device = device_cls(SimSignalBackend(signal_dtype))
  else:
- sub_device = device_cls()
-
+ sub_device = getattr(device, device_name, device_cls())
  _sim_common_blocks(sub_device, stripped_type=device_cls)
 
  setattr(device, device_name, sub_device)
@@ -223,7 +225,7 @@ async def _get_pvi_entries(entry: PVIEntry, timeout=DEFAULT_TIMEOUT):
  if is_signal:
  device = _pvi_mapping[frozenset(pva_entries.keys())](signal_dtype, *pvs)
  else:
- device = device_type()
+ device = getattr(entry.device, sub_name, device_type())
 
  sub_entry = PVIEntry(
  device=device, common_device_type=device_type, sub_entries={}
@@ -291,3 +293,26 @@ async def fill_pvi_entries(
  # We call set name now the parent field has been set in all of the
  # introspect-initialized devices. This will recursively set the names.
  device.set_name(device.name)
+
+
+def create_children_from_annotations(
+ device: Device, included_optional_fields: Tuple[str, ...] = ()
+):
+ """For intializing blocks at __init__ of ``device``."""
+ for name, device_type in get_type_hints(type(device)).items():
+ if name in ("_name", "parent"):
+ continue
+ device_type, is_optional = _strip_union(device_type)
+ if is_optional and name not in included_optional_fields:
+ continue
+ is_device_vector, device_type = _strip_device_vector(device_type)
+ if (
+ is_device_vector
+ or ((origin := get_origin(device_type)) and issubclass(origin, Signal))
+ or (isclass(device_type) and issubclass(device_type, Signal))
+ ):
+ continue
+
+ sub_device = device_type()
+ setattr(device, name, sub_device)
+ create_children_from_annotations(sub_device)

src/ophyd_async/panda/__init__.py

@@ -1,24 +1,25 @@
-from .panda import (
- CommonPandABlocks,
+from ._common_blocks import (
+ CommonPandaBlocks,
  DataBlock,
- PandA,
  PcapBlock,
  PulseBlock,
  SeqBlock,
  TimeUnits,
 )
-from .panda_controller import PandaPcapController
-from .table import (
+from ._hdf_panda import HDFPanda
+from ._panda_controller import PandaPcapController
+from ._table import (
  SeqTable,
  SeqTableRow,
  SeqTrigger,
  seq_table_from_arrays,
  seq_table_from_rows,
 )
-from .utils import phase_sorter
+from ._utils import phase_sorter
 
 __all__ = [
- "PandA",
+ "CommonPandaBlocks",
+ "HDFPanda",
  "PcapBlock",
  "PulseBlock",
  "seq_table_from_arrays",

src/ophyd_async/panda/_common_blocks.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from enum import Enum
+
+from ophyd_async.core import Device, DeviceVector, SignalR, SignalRW
+from ophyd_async.panda._table import SeqTable
+
+
+class DataBlock(Device):
+ # In future we may decide to make hdf_* optional
+ hdf_directory: SignalRW[str]
+ hdf_file_name: SignalRW[str]
+ num_capture: SignalRW[int]
+ num_captured: SignalR[int]
+ capture: SignalRW[bool]
+ flush_period: SignalRW[float]
+
+
+class PulseBlock(Device):
+ delay: SignalRW[float]
+ width: SignalRW[float]
+
+
+class TimeUnits(str, Enum):
+ min = "min"
+ s = "s"
+ ms = "ms"
+ us = "us"
+
+
+class SeqBlock(Device):
+ table: SignalRW[SeqTable]
+ active: SignalRW[bool]
+ repeats: SignalRW[int]
+ prescale: SignalRW[float]
+ prescale_units: SignalRW[TimeUnits]
+ enable: SignalRW[str]
+
+
+class PcapBlock(Device):
+ active: SignalR[bool]
+ arm: SignalRW[bool]
+
+
+class CommonPandaBlocks(Device):
+ pulse: DeviceVector[PulseBlock]
+ seq: DeviceVector[SeqBlock]
+ pcap: PcapBlock
+ data: DataBlock

src/ophyd_async/panda/_hdf_panda.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from typing import Sequence
+
+from ophyd_async.core import (
+ DEFAULT_TIMEOUT,
+ DirectoryProvider,
+ SignalR,
+ StandardDetector,
+)
+from ophyd_async.epics.pvi import create_children_from_annotations, fill_pvi_entries
+
+from ._common_blocks import CommonPandaBlocks
+from ._panda_controller import PandaPcapController
+from .writers._hdf_writer import PandaHDFWriter
+
+
+class HDFPanda(CommonPandaBlocks, StandardDetector):
+ def __init__(
+ self,
+ prefix: str,
+ directory_provider: DirectoryProvider,
+ config_sigs: Sequence[SignalR] = (),
+ name: str = "",
+ ):
+ self._prefix = prefix
+
+ create_children_from_annotations(self)
+ controller = PandaPcapController(pcap=self.pcap)
+ writer = PandaHDFWriter(
+ prefix=prefix,
+ directory_provider=directory_provider,
+ name_provider=lambda: name,
+ panda_device=self,
+ )
+ super().__init__(
+ controller=controller,
+ writer=writer,
+ config_sigs=config_sigs,
+ name=name,
+ writer_timeout=DEFAULT_TIMEOUT,
+ )
+
+ async def connect(
+ self, sim: bool = False, timeout: float = DEFAULT_TIMEOUT
+ ) -> None:
+ await fill_pvi_entries(self, self._prefix + "PVI", timeout=timeout, sim=sim)
+ await super().connect(sim=sim, timeout=timeout)

src/ophyd_async/panda/panda_controller.py → src/ophyd_async/panda/_panda_controller.py

@@ -7,15 +7,11 @@
  DetectorTrigger,
  wait_for_value,
 )
-
-from .panda import PcapBlock
+from ophyd_async.panda import PcapBlock
 
 
 class PandaPcapController(DetectorControl):
- def __init__(
- self,
- pcap: PcapBlock,
- ) -> None:
+ def __init__(self, pcap: PcapBlock) -> None:
  self.pcap = pcap
 
  def get_deadtime(self, exposure: float) -> float:
@@ -35,7 +31,7 @@ async def arm(
  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):
+ async def disarm(self) -> AsyncStatus:
  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))

src/ophyd_async/panda/writers/__init__.py

@@ -1,3 +1,3 @@
-from .hdf_writer import PandaHDFWriter
+from ._hdf_writer import PandaHDFWriter
 
 __all__ = ["PandaHDFWriter"]

src/ophyd_async/panda/writers/hdf_writer.py → src/ophyd_async/panda/writers/_hdf_writer.py

@@ -17,9 +17,9 @@
  wait_for_value,
 )
 from ophyd_async.core.signal import observe_value
-from ophyd_async.panda.panda import PandA
+from ophyd_async.panda import CommonPandaBlocks
 
-from .panda_hdf_file import _HDFDataset, _HDFFile
+from ._panda_hdf_file import _HDFDataset, _HDFFile
 
 
 class Capture(str, Enum):
@@ -96,7 +96,7 @@ def __init__(
  prefix: str,
  directory_provider: DirectoryProvider,
  name_provider: NameProvider,
- panda_device: PandA,
+ panda_device: CommonPandaBlocks,
  ) -> None:
  self.panda_device = panda_device
  self._prefix = prefix
@@ -142,11 +142,11 @@ async def open(self, multiplier: int = 1) -> Dict[str, Descriptor]:
  for attribute_path, capture_signal in to_capture.items():
  split_path = attribute_path.split(".")
  signal_name = split_path[-1]
+ # Get block names from numbered blocks, eg INENC[1]
  block_name = (
- split_path[-2]
- if not split_path[-2].isnumeric()
- # Get block names from numbered blocks, eg INENC[1]
- else f"{split_path[-3]}{split_path[-2]}"
+ f"{split_path[-3]}{split_path[-2]}"
+ if split_path[-2].isnumeric()
+ else split_path[-2]
  )
 
  for suffix in str(capture_signal.capture_type).split(" "):