diff --git a/src/fastcs_pandablocks/__init__.py b/src/fastcs_pandablocks/__init__.py index 4f6a708..2c02802 100644 --- a/src/fastcs_pandablocks/__init__.py +++ b/src/fastcs_pandablocks/__init__.py @@ -10,13 +10,12 @@ from ._version import __version__ from .gui import PandaGUIOptions from .panda.controller import PandaController -from .types import EpicsName DEFAULT_POLL_PERIOD = 0.1 def ioc( - epics_prefix: EpicsName, + epics_prefix: str, hostname: str, screens_directory: Path | None = None, clear_bobfiles: bool = False, @@ -31,7 +30,7 @@ def ioc( controller = PandaController(hostname, poll_period) backend = EpicsBackend( - controller, pv_prefix=str(epics_prefix), ioc_options=epics_ioc_options + controller, pv_prefix=epics_prefix, ioc_options=epics_ioc_options ) if clear_bobfiles and not screens_directory: diff --git a/src/fastcs_pandablocks/__main__.py b/src/fastcs_pandablocks/__main__.py index a0c4ae3..4a5fd3d 100644 --- a/src/fastcs_pandablocks/__main__.py +++ b/src/fastcs_pandablocks/__main__.py @@ -7,7 +7,6 @@ from fastcs.backends.epics.util import PvNamingConvention from fastcs_pandablocks import DEFAULT_POLL_PERIOD, ioc -from fastcs_pandablocks.types import EpicsName from . import __version__ @@ -83,7 +82,7 @@ def main(): logging.basicConfig(format="%(levelname)s:%(message)s", level=level) ioc( - EpicsName(prefix=parsed_args.prefix), + parsed_args.prefix, parsed_args.hostname, screens_directory=Path(parsed_args.screens_dir), clear_bobfiles=parsed_args.clear_bobfiles, diff --git a/src/fastcs_pandablocks/panda/blocks.py b/src/fastcs_pandablocks/panda/blocks.py deleted file mode 100644 index 2ce23e9..0000000 --- a/src/fastcs_pandablocks/panda/blocks.py +++ /dev/null @@ -1,146 +0,0 @@ -from collections.abc import Generator - -from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW -from fastcs.controller import SubController - -from fastcs_pandablocks.types import ( - EpicsName, - PandaName, - RawBlocksType, - RawFieldsType, - RawInitialValuesType, -) -from fastcs_pandablocks.types.annotations import ResponseType - -from .fields import ( - FieldControllerType, - get_field_controller_from_field_info, -) - - -def _def_pop_up_to_block_or_field(name: PandaName, dictionary: RawInitialValuesType): - extracted_members = {} - resolution_method = ( - PandaName.up_to_block if name.field is None else PandaName.up_to_field - ) - - # So the dictionary can be changed during iteration. - for sub_name in list(dictionary): - if resolution_method(sub_name) == name: - extracted_members[sub_name] = dictionary.pop(sub_name) - return extracted_members - - -class BlockController(SubController): - fields: dict[PandaName, FieldControllerType] - - def __init__( - self, - panda_name: PandaName, - description: str | None | None, - field_infos: dict[PandaName, ResponseType], - initial_values: RawInitialValuesType, - label: str | None, - ): - self.panda_name = panda_name - self.description = description - self.label = label - - self._additional_attributes: dict[str, Attribute] = {} - self.fields: dict[PandaName, FieldControllerType] = {} - - for field_name, field_info in field_infos.items(): - field_name = panda_name + field_name - field_initial_values = _def_pop_up_to_block_or_field( - field_name, initial_values - ) - self.fields[field_name] = get_field_controller_from_field_info( - field_name, field_info, field_initial_values, label - ) - - super().__init__() - - def initialise(self): - for field_name, field in self.fields.items(): - if field.additional_attributes: - self.register_sub_controller( - field_name.attribute_name, sub_controller=field - ) - field.initialise() # Registers `field.sub_contollers`. - if field.top_level_attribute: - assert field_name.field is not None - self._additional_attributes[field_name.field] = ( - field.top_level_attribute - ) - - @property - def additional_attributes(self) -> dict[str, Attribute]: - return self._additional_attributes - - -class Blocks: - _blocks: dict[PandaName, BlockController] - epics_prefix: EpicsName - - def __init__(self): - self._blocks = {} - - def parse_introspected_data( - self, - blocks: RawBlocksType, - field_infos: RawFieldsType, - labels: RawInitialValuesType, - initial_values: RawInitialValuesType, - ): - self._blocks = {} - - for (block_name, block_info), field_info in zip( - blocks.items(), field_infos, strict=True - ): - numbered_block_names = ( - [block_name] - if block_info.number in (None, 1) - else [ - block_name + PandaName(block_number=number) - for number in range(1, block_info.number + 1) - ] - ) - - for numbered_block_name in numbered_block_names: - block_initial_values = _def_pop_up_to_block_or_field( - numbered_block_name, initial_values - ) - label = labels.get(numbered_block_name, None) - - self._blocks[numbered_block_name] = BlockController( - numbered_block_name, - block_info.description, - field_info, - block_initial_values, - label, - ) - - async def update_field_value(self, panda_name: PandaName, value: str): - attribute = self[panda_name] - - if isinstance(attribute, AttrW): - await attribute.process(value) - elif isinstance(attribute, (AttrRW | AttrR)): - await attribute.set(value) - else: - raise RuntimeError(f"Couldn't find panda field for {panda_name}.") - - def flattened_attribute_tree( - self, - ) -> Generator[tuple[str, BlockController], None, None]: - for block in self._blocks.values(): - yield (block.panda_name.attribute_name, block) - - def __getitem__(self, name: PandaName) -> BlockController | Attribute | None: - block = self._blocks[name.up_to_block()] - if name.field is None: - return block - field = block.fields[name.up_to_field()] - if name.sub_field is None: - return field.top_level_attribute - return field.additional_attributes[name.sub_field] diff --git a/src/fastcs_pandablocks/panda/client_wrapper.py b/src/fastcs_pandablocks/panda/client_wrapper.py index 067cf76..49f9959 100644 --- a/src/fastcs_pandablocks/panda/client_wrapper.py +++ b/src/fastcs_pandablocks/panda/client_wrapper.py @@ -14,11 +14,11 @@ ) from fastcs_pandablocks.types import ( + PandaName, RawBlocksType, RawFieldsType, RawInitialValuesType, ) -from fastcs_pandablocks.types.string_types import PandaName class RawPanda: diff --git a/src/fastcs_pandablocks/panda/controller.py b/src/fastcs_pandablocks/panda/controller.py index b0092da..487d2c0 100644 --- a/src/fastcs_pandablocks/panda/controller.py +++ b/src/fastcs_pandablocks/panda/controller.py @@ -1,49 +1,119 @@ import asyncio +from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW from fastcs.controller import Controller from fastcs.wrappers import scan -from fastcs_pandablocks.types import PandaName +from fastcs_pandablocks.types import ( + PandaName, + RawBlocksType, + RawFieldsType, + RawInitialValuesType, +) -from .blocks import Blocks from .client_wrapper import RawPanda +from .fields import FieldController + + +def _parse_introspected_data( + blocks: RawBlocksType, + field_infos: RawFieldsType, + labels: RawInitialValuesType, + initial_values: RawInitialValuesType, +): + block_controllers: dict[PandaName, FieldController] = {} + for (block_name, block_info), field_info in zip( + blocks.items(), field_infos, strict=True + ): + numbered_block_names = ( + [block_name] + if block_info.number in (None, 1) + else [ + block_name + PandaName(block_number=number) + for number in range(1, block_info.number + 1) + ] + ) + for numbered_block_name in numbered_block_names: + block_initial_values = { + key: value + for key, value in initial_values.items() + if key in numbered_block_name + } + label = labels.get(numbered_block_name, None) + block = FieldController( + numbered_block_name, + label=block_info.description or label, + ) + block.make_sub_fields(field_info, block_initial_values) + block_controllers[numbered_block_name] = block + + return block_controllers class PandaController(Controller): def __init__(self, hostname: str, poll_period: float) -> None: + # TODO https://github.com/DiamondLightSource/FastCS/issues/62 + self.poll_period = poll_period + + self._additional_attributes: dict[str, Attribute] = {} self._raw_panda = RawPanda(hostname) - self._blocks = Blocks() - self.is_connected = False + _blocks: dict[PandaName, FieldController] = {} super().__init__() - async def connect(self) -> None: - if self.is_connected: - return + @property + def additional_attributes(self): + return self._additional_attributes + async def connect(self) -> None: await self._raw_panda.connect() blocks, fields, labels, initial_values = await self._raw_panda.introspect() - - self._blocks.parse_introspected_data(blocks, fields, labels, initial_values) - self.is_connected = True + self._blocks = _parse_introspected_data(blocks, fields, labels, initial_values) async def initialise(self) -> None: await self.connect() + for block_name, block in self._blocks.items(): + if block.top_level_attribute is not None: + self._additional_attributes[block_name.attribute_name] = ( + block.top_level_attribute + ) + if block.additional_attributes or block.sub_fields: + self.register_sub_controller(block_name.attribute_name, block) + await block.initialise() + + def get_attribute(self, panda_name: PandaName) -> Attribute: + assert panda_name.block + block_controller = self._blocks[panda_name.up_to_block()] + if panda_name.field is None: + assert block_controller.top_level_attribute is not None + return block_controller.top_level_attribute + + field_controller = block_controller.sub_fields[panda_name.up_to_field()] + if panda_name.sub_field is None: + assert field_controller.top_level_attribute is not None + return field_controller.top_level_attribute - for attr_name, controller in self._blocks.flattened_attribute_tree(): - self.register_sub_controller(attr_name, controller) - controller.initialise() + sub_field_controller = field_controller.sub_fields[panda_name] + assert sub_field_controller.top_level_attribute is not None + return sub_field_controller.top_level_attribute + + async def update_field_value(self, panda_name: PandaName, value: str): + attribute = self.get_attribute(panda_name) + + if isinstance(attribute, AttrW): + await attribute.process(value) + elif isinstance(attribute, (AttrRW | AttrR)): + await attribute.set(value) + else: + raise RuntimeError(f"Couldn't find panda field for {panda_name}.") - # TODO https://github.com/DiamondLightSource/FastCS/issues/62 @scan(0.1) async def update(self): await self._raw_panda.get_changes() assert self._raw_panda.changes await asyncio.gather( *[ - self._blocks.update_field_value( - PandaName.from_string(raw_panda_name), value - ) + self.update_field_value(PandaName.from_string(raw_panda_name), value) for raw_panda_name, value in self._raw_panda.changes.items() ] ) diff --git a/src/fastcs_pandablocks/panda/fields.py b/src/fastcs_pandablocks/panda/fields.py index 99c20bb..b656c15 100644 --- a/src/fastcs_pandablocks/panda/fields.py +++ b/src/fastcs_pandablocks/panda/fields.py @@ -1,7 +1,5 @@ from __future__ import annotations -from enum import Enum - from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW from fastcs.controller import SubController from fastcs.datatypes import Bool, Float, Int, String @@ -29,61 +27,78 @@ DefaultFieldUpdater, EguSender, ) -from fastcs_pandablocks.types.annotations import RawInitialValuesType, ResponseType -from fastcs_pandablocks.types.string_types import PandaName - - -class WidgetGroup(Enum): - """Purposely not an enum since we only ever want the string.""" - - NONE = None - PARAMETERS = "Parameters" - OUTPUTS = "Outputs" - INPUTS = "Inputs" - READBACKS = "Readbacks" - CAPTURE = "Capture" - +from fastcs_pandablocks.types import ( + PandaName, + RawInitialValuesType, + ResponseType, + WidgetGroup, +) # EPICS hardcoded. TODO: remove once we switch to pvxs. MAXIMUM_DESCRIPTION_LENGTH = 40 -def _strip_description(description: str | None) -> str: +def _strip_description(description: str | None) -> str | None: if description is None: - return "" + return description return description[:MAXIMUM_DESCRIPTION_LENGTH] class FieldController(SubController): - def __init__(self): - """ - Since fields contain an attribute for the field itself - `PREFIX:BLOCK:FIELD`, but also subfields, `PREFIX:BLOCK:FIELD:SUB_FIELD`, - have a top level attribute set in the `BlockController`, and - further attributes which are used in the field as a `SubController`. - """ - + def __init__( + self, + panda_name: PandaName, + label: str | None = None, + ): + self.panda_name = panda_name self.top_level_attribute: Attribute | None = None - self._additional_attributes = {} - self.sub_controllers: dict[str, FieldController] = {} + + # Sub fields eg `PGEN.OUT` and `PGEN.TRIGGER` + self.sub_fields: dict[PandaName, FieldController] = {} + + self._additional_attributes: dict[str, Attribute] = {} + super().__init__(search_device_for_attributes=False) + def make_sub_fields( + self, + field_infos: dict[PandaName, ResponseType], + initial_values: RawInitialValuesType, + ): + for sub_field_name, field_info in field_infos.items(): + full_sub_field_name = self.panda_name + sub_field_name + field_initial_values = { + key: value + for key, value in initial_values.items() + if key in sub_field_name + } + self.sub_fields[full_sub_field_name] = get_field_controller_from_field_info( + full_sub_field_name, field_info, field_initial_values + ) + + async def initialise(self): + for field_name, field in self.sub_fields.items(): + self.register_sub_controller( + field_name.attribute_name, sub_controller=field + ) + await field.initialise() + if field.top_level_attribute: + self._additional_attributes[field_name.attribute_name] = ( + field.top_level_attribute + ) + @property def additional_attributes(self) -> dict[str, Attribute]: + """ + Used by the FastCS mapping parser to get attributes since + we're not searching for device attributes. + """ return self._additional_attributes - def initialise(self): - for sub_field_name, sub_field_controller in self.sub_controllers.items(): - self.register_sub_controller(sub_field_name, sub_field_controller) - sub_field_controller.initialise() - self._additional_attributes[sub_field_name] = ( - sub_field_controller.top_level_attribute - ) - class TableFieldController(FieldController): def __init__(self, panda_name: PandaName, field_info: TableFieldInfo): - super().__init__() + super().__init__(panda_name, field_info.description) self.top_level_attribute = AttrR( Float(), @@ -99,7 +114,7 @@ def __init__( panda_name: PandaName, field_info: SubtypeTimeFieldInfo | TimeFieldInfo, ): - super().__init__() + super().__init__(panda_name) self.top_level_attribute = AttrRW( Float(), handler=DefaultFieldHandler(panda_name), @@ -120,7 +135,7 @@ def __init__( panda_name: PandaName, field_info: SubtypeTimeFieldInfo, ): - super().__init__() + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(), handler=DefaultFieldUpdater( @@ -143,7 +158,7 @@ def __init__( panda_name: PandaName, field_info: SubtypeTimeFieldInfo, ): - super().__init__() + super().__init__(panda_name) self.top_level_attribute = AttrW( Float(), handler=DefaultFieldSender(panda_name), @@ -159,8 +174,8 @@ def __init__( class BitOutFieldController(FieldController): - def __init__(self, field_info: BitOutFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, field_info: BitOutFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Bool(znam="0", onam="1"), description=_strip_description(field_info.description), @@ -170,7 +185,7 @@ def __init__(self, field_info: BitOutFieldInfo): class PosOutFieldController(FieldController): def __init__(self, panda_name: PandaName, field_info: PosOutFieldInfo): - super().__init__() + super().__init__(panda_name) top_level_attribute = AttrR( Float(), description=_strip_description(field_info.description), @@ -219,8 +234,8 @@ async def updated_scaled_on_offset_change(*_): class ExtOutFieldController(FieldController): - def __init__(self, field_info: ExtOutFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, field_info: ExtOutFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(), description=_strip_description(field_info.description), @@ -241,8 +256,8 @@ def __init__(self, field_info: ExtOutFieldInfo): class _BitsSubFieldController(FieldController): - def __init__(self, label: str): - super().__init__() + def __init__(self, panda_name: PandaName, label: str): + super().__init__(panda_name, label=label) self.top_level_attribute = AttrR( Bool(znam="0", onam="1"), @@ -259,20 +274,24 @@ def __init__(self, label: str): class ExtOutBitsFieldController(ExtOutFieldController): def __init__( self, + panda_name: PandaName, field_info: ExtOutBitsFieldInfo, ): - super().__init__(field_info) + super().__init__(panda_name, field_info) for bit_number, label in enumerate(field_info.bits, start=1): if label == "": continue # Some rows are empty, do not create records. - self.sub_controllers[f"BIT{bit_number}"] = _BitsSubFieldController(label) + sub_field_panda_name = panda_name + PandaName(sub_field=f"bit{bit_number}") + self.sub_fields[sub_field_panda_name] = _BitsSubFieldController( + sub_field_panda_name, label=label + ) class BitMuxFieldController(FieldController): def __init__(self, panda_name: PandaName, bit_mux_field_info: BitMuxFieldInfo): - super().__init__() + super().__init__(panda_name) self.top_level_attribute = AttrRW( String(), description=_strip_description(bit_mux_field_info.description), @@ -291,8 +310,8 @@ def __init__(self, panda_name: PandaName, bit_mux_field_info: BitMuxFieldInfo): class PosMuxFieldController(FieldController): - def __init__(self, pos_mux_field_info: PosMuxFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, pos_mux_field_info: PosMuxFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrRW( String(), group=WidgetGroup.INPUTS.value, @@ -300,8 +319,8 @@ def __init__(self, pos_mux_field_info: PosMuxFieldInfo): class UintParamFieldController(FieldController): - def __init__(self, uint_param_field_info: UintFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, uint_param_field_info: UintFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(prec=0), group=WidgetGroup.PARAMETERS.value, @@ -309,8 +328,8 @@ def __init__(self, uint_param_field_info: UintFieldInfo): class UintReadFieldController(FieldController): - def __init__(self, uint_read_field_info: UintFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, uint_read_field_info: UintFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(prec=0), group=WidgetGroup.READBACKS.value, @@ -320,8 +339,8 @@ def __init__(self, uint_read_field_info: UintFieldInfo): class UintWriteFieldController(FieldController): - def __init__(self, uint_write_field_info: UintFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, uint_write_field_info: UintFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrW( Float(prec=0), group=WidgetGroup.OUTPUTS.value, @@ -329,8 +348,8 @@ def __init__(self, uint_write_field_info: UintFieldInfo): class IntParamFieldController(FieldController): - def __init__(self, int_param_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, int_param_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrRW( Float(prec=0), group=WidgetGroup.PARAMETERS.value, @@ -338,8 +357,8 @@ def __init__(self, int_param_field_info: FieldInfo): class IntReadFieldController(FieldController): - def __init__(self, int_read_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, int_read_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Int(), group=WidgetGroup.READBACKS.value, @@ -347,8 +366,8 @@ def __init__(self, int_read_field_info: FieldInfo): class IntWriteFieldController(FieldController): - def __init__(self, int_write_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, int_write_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrW( Int(), group=WidgetGroup.PARAMETERS.value, @@ -356,8 +375,8 @@ def __init__(self, int_write_field_info: FieldInfo): class ScalarParamFieldController(FieldController): - def __init__(self, scalar_param_field_info: ScalarFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, scalar_param_field_info: ScalarFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrRW( Float(), group=WidgetGroup.PARAMETERS.value, @@ -365,8 +384,8 @@ def __init__(self, scalar_param_field_info: ScalarFieldInfo): class ScalarReadFieldController(FieldController): - def __init__(self, scalar_read_field_info: ScalarFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, scalar_read_field_info: ScalarFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(), group=WidgetGroup.READBACKS.value, @@ -374,8 +393,8 @@ def __init__(self, scalar_read_field_info: ScalarFieldInfo): class ScalarWriteFieldController(FieldController): - def __init__(self, scalar_write_field_info: ScalarFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, scalar_write_field_info: ScalarFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Float(), group=WidgetGroup.PARAMETERS.value, @@ -383,8 +402,8 @@ def __init__(self, scalar_write_field_info: ScalarFieldInfo): class BitParamFieldController(FieldController): - def __init__(self, bit_param_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, bit_param_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrRW( Bool(znam="0", onam="1"), group=WidgetGroup.PARAMETERS.value, @@ -392,8 +411,8 @@ def __init__(self, bit_param_field_info: FieldInfo): class BitReadFieldController(FieldController): - def __init__(self, bit_read_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, bit_read_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Bool(znam="0", onam="1"), group=WidgetGroup.READBACKS.value, @@ -401,8 +420,8 @@ def __init__(self, bit_read_field_info: FieldInfo): class BitWriteFieldController(FieldController): - def __init__(self, bit_write_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, bit_write_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrW( Bool(znam="0", onam="1"), group=WidgetGroup.OUTPUTS.value, @@ -410,8 +429,8 @@ def __init__(self, bit_write_field_info: FieldInfo): class ActionReadFieldController(FieldController): - def __init__(self, action_read_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, action_read_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( Bool(znam="0", onam="1"), group=WidgetGroup.READBACKS.value, @@ -419,8 +438,8 @@ def __init__(self, action_read_field_info: FieldInfo): class ActionWriteFieldController(FieldController): - def __init__(self, action_write_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, action_write_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrW( Bool(znam="0", onam="1"), group=WidgetGroup.OUTPUTS.value, @@ -428,8 +447,8 @@ def __init__(self, action_write_field_info: FieldInfo): class LutParamFieldController(FieldController): - def __init__(self, lut_param_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, lut_param_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrRW( String(), group=WidgetGroup.PARAMETERS.value, @@ -437,8 +456,8 @@ def __init__(self, lut_param_field_info: FieldInfo): class LutReadFieldController(FieldController): - def __init__(self, lut_read_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, lut_read_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( String(), group=WidgetGroup.READBACKS.value, @@ -446,8 +465,8 @@ def __init__(self, lut_read_field_info: FieldInfo): class LutWriteFieldController(FieldController): - def __init__(self, lut_read_field_info: FieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, lut_read_field_info: FieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( String(), group=WidgetGroup.OUTPUTS.value, @@ -455,8 +474,8 @@ def __init__(self, lut_read_field_info: FieldInfo): class EnumParamFieldController(FieldController): - def __init__(self, enum_param_field_info: EnumFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, enum_param_field_info: EnumFieldInfo): + super().__init__(panda_name) self.allowed_values = enum_param_field_info.labels self.top_level_attribute = AttrRW( String(), @@ -465,8 +484,8 @@ def __init__(self, enum_param_field_info: EnumFieldInfo): class EnumReadFieldController(FieldController): - def __init__(self, enum_read_field_info: EnumFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, enum_read_field_info: EnumFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrR( String(), group=WidgetGroup.READBACKS.value, @@ -474,8 +493,8 @@ def __init__(self, enum_read_field_info: EnumFieldInfo): class EnumWriteFieldController(FieldController): - def __init__(self, enum_write_field_info: EnumFieldInfo): - super().__init__() + def __init__(self, panda_name: PandaName, enum_write_field_info: EnumFieldInfo): + super().__init__(panda_name) self.top_level_attribute = AttrW( String(), group=WidgetGroup.OUTPUTS.value, @@ -520,7 +539,6 @@ def get_field_controller_from_field_info( panda_name: PandaName, field_info: ResponseType, initial_values: RawInitialValuesType, - label: str | None, ) -> FieldControllerType: match field_info: case TableFieldInfo(): @@ -537,79 +555,79 @@ def get_field_controller_from_field_info( # Bit types case BitOutFieldInfo(): - return BitOutFieldController(field_info) + return BitOutFieldController(panda_name, field_info) case ExtOutBitsFieldInfo(subtype="timestamp"): - return ExtOutFieldController(field_info) + return ExtOutFieldController(panda_name, field_info) case ExtOutBitsFieldInfo(): - return ExtOutBitsFieldController(field_info) + return ExtOutBitsFieldController(panda_name, field_info) case ExtOutFieldInfo(): - return ExtOutFieldController(field_info) + return ExtOutFieldController(panda_name, field_info) case BitMuxFieldInfo(): return BitMuxFieldController(panda_name, field_info) case FieldInfo(type="param", subtype="bit"): - return BitParamFieldController(field_info) + return BitParamFieldController(panda_name, field_info) case FieldInfo(type="read", subtype="bit"): - return BitReadFieldController(field_info) + return BitReadFieldController(panda_name, field_info) case FieldInfo(type="write", subtype="bit"): - return BitWriteFieldController(field_info) + return BitWriteFieldController(panda_name, field_info) # Pos types case PosOutFieldInfo(): return PosOutFieldController(panda_name, field_info) case PosMuxFieldInfo(): - return PosMuxFieldController(field_info) + return PosMuxFieldController(panda_name, field_info) # Uint types case UintFieldInfo(type="param"): - return UintParamFieldController(field_info) + return UintParamFieldController(panda_name, field_info) case UintFieldInfo(type="read"): - return UintReadFieldController(field_info) + return UintReadFieldController(panda_name, field_info) case UintFieldInfo(type="write"): - return UintWriteFieldController(field_info) + return UintWriteFieldController(panda_name, field_info) # Scalar types case ScalarFieldInfo(subtype="param"): - return ScalarParamFieldController(field_info) + return ScalarParamFieldController(panda_name, field_info) case ScalarFieldInfo(type="read"): - return ScalarReadFieldController(field_info) + return ScalarReadFieldController(panda_name, field_info) case ScalarFieldInfo(type="write"): - return ScalarWriteFieldController(field_info) + return ScalarWriteFieldController(panda_name, field_info) # Int types case FieldInfo(type="param", subtype="int"): - return IntParamFieldController(field_info) + return IntParamFieldController(panda_name, field_info) case FieldInfo(type="read", subtype="int"): - return IntReadFieldController(field_info) + return IntReadFieldController(panda_name, field_info) case FieldInfo(type="write", subtype="int"): - return IntWriteFieldController(field_info) + return IntWriteFieldController(panda_name, field_info) # Action types case FieldInfo( type="read", subtype="action", ): - return ActionReadFieldController(field_info) + return ActionReadFieldController(panda_name, field_info) case FieldInfo( type="write", subtype="action", ): - return ActionWriteFieldController(field_info) + return ActionWriteFieldController(panda_name, field_info) # Lut types case FieldInfo(type="param", subtype="lut"): - return LutParamFieldController(field_info) + return LutParamFieldController(panda_name, field_info) case FieldInfo(type="read", subtype="lut"): - return LutReadFieldController(field_info) + return LutReadFieldController(panda_name, field_info) case FieldInfo(type="write", subtype="lut"): - return LutWriteFieldController(field_info) + return LutWriteFieldController(panda_name, field_info) # Enum types case EnumFieldInfo(type="param"): - return EnumParamFieldController(field_info) + return EnumParamFieldController(panda_name, field_info) case EnumFieldInfo(type="read"): - return EnumReadFieldController(field_info) + return EnumReadFieldController(panda_name, field_info) case EnumFieldInfo(type="write"): - return EnumWriteFieldController(field_info) + return EnumWriteFieldController(panda_name, field_info) case _: raise ValueError(f"Unknown field type: {type(field_info).__name__}.") diff --git a/src/fastcs_pandablocks/types/__init__.py b/src/fastcs_pandablocks/types/__init__.py index b124a72..efaf40d 100644 --- a/src/fastcs_pandablocks/types/__init__.py +++ b/src/fastcs_pandablocks/types/__init__.py @@ -1,23 +1,34 @@ -from .annotations import ( +from enum import Enum + +from ._annotations import ( RawBlocksType, RawFieldsType, RawInitialValuesType, ResponseType, ) -from .string_types import ( +from ._string_types import ( EPICS_SEPARATOR, PANDA_SEPARATOR, - EpicsName, PandaName, ) + +class WidgetGroup(Enum): + NONE = None + PARAMETERS = "Parameters" + OUTPUTS = "Outputs" + INPUTS = "Inputs" + READBACKS = "Readbacks" + CAPTURE = "Capture" + + __all__ = [ "EPICS_SEPARATOR", - "EpicsName", "PANDA_SEPARATOR", "PandaName", "ResponseType", "RawBlocksType", "RawFieldsType", "RawInitialValuesType", + "WidgetGroup", ] diff --git a/src/fastcs_pandablocks/types/annotations.py b/src/fastcs_pandablocks/types/_annotations.py similarity index 95% rename from src/fastcs_pandablocks/types/annotations.py rename to src/fastcs_pandablocks/types/_annotations.py index aca1cab..4d50a33 100644 --- a/src/fastcs_pandablocks/types/annotations.py +++ b/src/fastcs_pandablocks/types/_annotations.py @@ -17,7 +17,7 @@ UintFieldInfo, ) -from .string_types import PandaName +from ._string_types import PandaName # Pyright gives us variable not allowed in type expression error # if we try to use the new (|) syntax diff --git a/src/fastcs_pandablocks/types/string_types.py b/src/fastcs_pandablocks/types/_string_types.py similarity index 54% rename from src/fastcs_pandablocks/types/string_types.py rename to src/fastcs_pandablocks/types/_string_types.py index 1a2482b..1b3107e 100644 --- a/src/fastcs_pandablocks/types/string_types.py +++ b/src/fastcs_pandablocks/types/_string_types.py @@ -95,15 +95,6 @@ def from_string(cls, name: str): block=block, block_number=block_number, field=field, sub_field=sub_field ) - @cached_property - def epics_name(self): - return EpicsName( - block=self.block, - block_number=self.block_number, - field=self.field, - sub_field=self.sub_field, - ) - def __add__(self, other: PandaName) -> PandaName: return PandaName( block=_choose_sub_pv(self.block, other.block), @@ -124,89 +115,10 @@ def attribute_name(self) -> str: ) return "" - -@dataclass(frozen=True) -class EpicsName: - prefix: str | None = None - block: str | None = None - block_number: int | None = None - field: str | None = None - sub_field: str | None = None - - @cached_property - def _string_form(self) -> str: - return _format_with_separator( - EPICS_SEPARATOR, - self.prefix, - (self.block, self.block_number), - self.field, - self.sub_field, - ) - - def __str__(self) -> str: - return self._string_form - - @classmethod - def from_string(cls, name: str) -> EpicsName: - """Converts a string to an EPICS name, must contain a prefix.""" - split_name = name.split(EPICS_SEPARATOR) - if len(split_name) < 3: - raise ValueError( - f"Received a a pv string `{name}` which isn't of the form " - "`PREFIX:BLOCK:FIELD` or `PREFIX:BLOCK:FIELD:SUB_FIELD`." - ) - split_name = name.split(EPICS_SEPARATOR) - prefix, block_with_number, field = split_name[:3] - block, block_number = _extract_number_at_of_string(block_with_number) - sub_field = split_name[3] if len(split_name) == 4 else None - - return EpicsName( - prefix=prefix, - block=block, - block_number=block_number, - field=field, - sub_field=sub_field, - ) - - @cached_property - def panda_name(self) -> PandaName: - return PandaName( - block=self.block, - block_number=self.block_number, - field=self.field, - sub_field=self.sub_field, - ) - - def __add__(self, other: EpicsName) -> EpicsName: - """ - Returns the sum of PVs: - - EpicsName(prefix="PREFIX", block="BLOCK") + EpicsName(field="FIELD") - == EpicsName.from_string("PREFIX:BLOCK:FIELD") - """ - - return EpicsName( - prefix=_choose_sub_pv(self.prefix, other.prefix), - block=_choose_sub_pv(self.block, other.block), - block_number=_choose_sub_pv(self.block_number, other.block_number), - field=_choose_sub_pv(self.field, other.field), - sub_field=_choose_sub_pv(self.sub_field, other.sub_field), - ) - - def __contains__(self, other: EpicsName) -> bool: - """Checks to see if a given epics name is a subset of another one. - - Examples - -------- - - (EpicsName(block="field1") in EpicsName("prefix:block1:field1")) == True - (EpicsName(block="field1") in EpicsName("prefix:block1:field2")) == False - """ - + def __contains__(self, other: PandaName) -> bool: return ( - _check_eq(self.prefix, other.prefix) - and _check_eq(self.block, other.block) - and _check_eq(self.block_number, other.block_number) - and _check_eq(self.field, other.field) - and _check_eq(self.sub_field, other.sub_field) + _check_eq(other.block, self.block) + and _check_eq(other.block_number, self.block_number) + and _check_eq(other.field, self.field) + and _check_eq(other.sub_field, self.sub_field) ) diff --git a/tests/test_types.py b/tests/test_types.py deleted file mode 100644 index 3a68023..0000000 --- a/tests/test_types.py +++ /dev/null @@ -1,66 +0,0 @@ -from dataclasses import FrozenInstanceError - -import pytest - -from fastcs_pandablocks.types import EpicsName, PandaName - - -@pytest.mark.parametrize( - "name_factory", - [ - lambda: EpicsName.from_string("PREFIX:BLOCK:FIELD"), - lambda: PandaName.from_string("BLOCK.FIELD"), - ], -) -def test_names_are_frozen(name_factory): - name = name_factory() - with pytest.raises(FrozenInstanceError): - name.block = "hello" - - -def test_epics_name(): - string_form = "prefix:block1:field:sub_field" - name1 = EpicsName.from_string(string_form) - assert name1.prefix == "prefix" - assert name1.block == "block" - assert name1.block_number == 1 - assert name1.field == "field" - assert name1.sub_field == "sub_field" - assert str(name1) == string_form - assert name1 == EpicsName( - prefix="prefix", - block="block", - block_number=1, - field="field", - sub_field="sub_field", - ) - - -def test_epics_name_add(): - assert ( - EpicsName.from_string("prefix:block1:field") - + EpicsName.from_string("prefix:block1:field") - ) == EpicsName.from_string("prefix:block1:field") - assert EpicsName(block="block") + EpicsName(block_number=1) == EpicsName( - block="block", block_number=1 - ) - with pytest.raises(TypeError) as error: - _ = EpicsName(block="block", block_number=1, field="field") + EpicsName( - block="block", block_number=2, field="field" - ) - assert str(error.value) == "Ambiguous pv elements on add 1 and 2" - - -def test_epics_name_contains(): - parent_name = EpicsName(prefix="prefix", block="block") - assert parent_name in parent_name - assert EpicsName(prefix="prefix", block="block", block_number=1) in parent_name - assert ( - EpicsName(prefix="prefix", block="block", block_number=1, field="field") - in parent_name - ) - assert parent_name not in EpicsName(block="block", block_number=1) - assert parent_name not in EpicsName(prefix="prefix", block="block", block_number=2) - assert parent_name not in EpicsName( - prefix="prefix", block="block", block_number=1, field="field" - )