diff --git a/pyproject.toml b/pyproject.toml index 49e6c8c..22903be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ classifiers = [ ] description = "A softioc to control a PandABlocks-FPGA." dependencies = [ - "fastcs~=0.6.0", + "fastcs@git+https://github.com/DiamondLightSource/FastCS@panda-conversion-improvements", "pandablocks~=0.10.0", "numpy<2", # until https://github.com/mdavidsaver/p4p/issues/145 is fixed "pydantic>2", diff --git a/src/fastcs_pandablocks/__init__.py b/src/fastcs_pandablocks/__init__.py index 2277088..2dcf1a1 100644 --- a/src/fastcs_pandablocks/__init__.py +++ b/src/fastcs_pandablocks/__init__.py @@ -4,6 +4,7 @@ from fastcs.backends.epics.backend import EpicsBackend from fastcs.backends.epics.gui import EpicsGUIFormat +from fastcs.backends.epics.ioc import EpicsIOCOptions, PvNamingConvention from ._version import __version__ from .gui import PandaGUIOptions @@ -21,7 +22,10 @@ def ioc( clear_bobfiles: bool = False, ): controller = PandaController(hostname, poll_period) - backend = EpicsBackend(controller, pv_prefix=str(prefix)) + epics_ioc_options = EpicsIOCOptions( + terminal=True, pv_naming_convention=PvNamingConvention.CAPITALIZED + ) + backend = EpicsBackend(controller, pv_prefix=str(prefix), options=epics_ioc_options) if clear_bobfiles and not screens_directory: raise ValueError("`clear_bobfiles` is True with no `screens_directory`") diff --git a/src/fastcs_pandablocks/gui.py b/src/fastcs_pandablocks/gui.py index 4285c76..c4189f0 100644 --- a/src/fastcs_pandablocks/gui.py +++ b/src/fastcs_pandablocks/gui.py @@ -1,5 +1,4 @@ from fastcs.backends.epics.gui import EpicsGUIOptions -class PandaGUIOptions(EpicsGUIOptions): - ... +class PandaGUIOptions(EpicsGUIOptions): ... diff --git a/src/fastcs_pandablocks/panda/blocks.py b/src/fastcs_pandablocks/panda/blocks.py index 19dc7df..67d8816 100644 --- a/src/fastcs_pandablocks/panda/blocks.py +++ b/src/fastcs_pandablocks/panda/blocks.py @@ -1,14 +1,15 @@ from collections.abc import Generator +from fastcs.attributes import AttrR, AttrRW, AttrW from fastcs.controller import SubController from pandablocks.responses import BlockInfo -from fastcs_pandablocks.types import EpicsName, PandaName, ResponseType +from fastcs_pandablocks.types import AttrType, EpicsName, PandaName, ResponseType from .fields import FIELD_TYPE_TO_FASTCS_TYPE, FieldType -class Block(SubController): +class BlockController(SubController): fields: dict[str, FieldType] def __init__( @@ -25,17 +26,24 @@ def __init__( self.fields = {} for field_raw_name, field_info in raw_fields.items(): - field_panda_name = PandaName(field=field_raw_name) - print(field_raw_name) + field_panda_name = self.panda_name + PandaName(field=field_raw_name) field = FIELD_TYPE_TO_FASTCS_TYPE[field_info.type][field_info.subtype]( - field_panda_name, field_info.description + # TODO make type safe after match statment + field_panda_name, + field_info, # type: ignore ) self.fields[field_raw_name] = field - self.register_sub_controller(field_panda_name.attribute_name, field) + if field.block_attribute: + setattr(self, *field.block_attribute) + if field.sub_field_controller: + self.register_sub_controller( + field_panda_name.attribute_name, field.sub_field_controller + ) + class Blocks: - _blocks: dict[str, dict[int | None, Block]] + _blocks: dict[str, dict[int | None, BlockController]] epics_prefix: EpicsName def __init__(self): @@ -51,11 +59,15 @@ def parse_introspected_data( ): iterator = ( range(1, block_info.number + 1) - if block_info.number > 1 else iter([None,]) + if block_info.number > 1 + else iter( + [ + None, + ] + ) ) self._blocks[block_name] = { - number: - Block( + number: BlockController( PandaName(block=block_name, block_number=number), block_info.number, block_info.description, @@ -65,26 +77,25 @@ def parse_introspected_data( } async def update_field_value(self, panda_name: PandaName, value: str): - assert panda_name.block - assert panda_name.field - field = ( - self._blocks[panda_name.block][panda_name.block_number].fields[panda_name.field] - ) - if panda_name.sub_field: - field = field.sub_fields[panda_name.sub_field] - await field.update_value(value) + 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, Block], None, None]: + self, + ) -> Generator[tuple[str, BlockController], None, None]: for blocks in self._blocks.values(): for block in blocks.values(): yield (block.panda_name.attribute_name, block) def __getitem__( - self, - name: EpicsName | PandaName - ) -> dict[int | None, Block] | Block | FieldType: + self, name: EpicsName | PandaName + ) -> dict[int | None, BlockController] | BlockController | AttrType: if name.block is None: raise ValueError(f"Cannot find block for name {name}.") blocks = self._blocks[name.block] @@ -94,6 +105,9 @@ def __getitem__( if name.field is None: return block field = block.fields[name.field] - if name.sub_field is None: - return field - return field.sub_fields[name.sub_field] + if not name.sub_field: + assert field.block_attribute + return field.block_attribute.attribute + + sub_field = getattr(field.sub_field_controller, name.sub_field) + return sub_field diff --git a/src/fastcs_pandablocks/panda/controller.py b/src/fastcs_pandablocks/panda/controller.py index 154b741..47fc216 100644 --- a/src/fastcs_pandablocks/panda/controller.py +++ b/src/fastcs_pandablocks/panda/controller.py @@ -3,7 +3,6 @@ from fastcs.controller import Controller from fastcs.wrappers import scan -from fastcs_pandablocks import DEFAULT_POLL_PERIOD from fastcs_pandablocks.types import PandaName from .blocks import Blocks @@ -12,16 +11,16 @@ class PandaController(Controller): def __init__(self, hostname: str, poll_period: float) -> None: - super().__init__() self._raw_panda = RawPanda(hostname) self._blocks = Blocks() + self.is_connected = False - # TODO https://github.com/DiamondLightSource/FastCS/issues/62 - #self.fastcs_method = Scan(self.update(), poll_period) - - async def initialise(self) -> None: ... + super().__init__() async def connect(self) -> None: + if self.is_connected: + return + await self._raw_panda.connect() assert self._raw_panda.blocks @@ -32,7 +31,17 @@ async def connect(self) -> None: for attr_name, controller in self._blocks.flattened_attribute_tree(): self.register_sub_controller(attr_name, controller) - @scan(DEFAULT_POLL_PERIOD) # TODO https://github.com/DiamondLightSource/FastCS/issues/62 + self.is_connected = True + + async def initialise(self) -> None: + """ + We connect in initialise since FastCS doesn't connect until + it's already parsed sub controllers. + """ + await self.connect() + + # 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 diff --git a/src/fastcs_pandablocks/panda/fields.py b/src/fastcs_pandablocks/panda/fields.py index 6bd79c3..68e29e4 100644 --- a/src/fastcs_pandablocks/panda/fields.py +++ b/src/fastcs_pandablocks/panda/fields.py @@ -1,303 +1,274 @@ from __future__ import annotations -from typing import Literal +from collections import namedtuple +from enum import Enum from fastcs.attributes import AttrR, AttrRW, AttrW from fastcs.controller import SubController from fastcs.datatypes import Bool, Float, Int, String +from pandablocks.responses import ( + BitMuxFieldInfo, + BitOutFieldInfo, + EnumFieldInfo, + ExtOutBitsFieldInfo, + ExtOutFieldInfo, + FieldInfo, + PosMuxFieldInfo, + PosOutFieldInfo, + ScalarFieldInfo, + SubtypeTimeFieldInfo, + TableFieldInfo, + TimeFieldInfo, + UintFieldInfo, +) -from fastcs_pandablocks.types import PandaName +from fastcs_pandablocks.types import AttrType, PandaName -class PviGroup: +class WidgetGroup(Enum): """Purposely not an enum since we only ever want the string.""" + + NONE = None PARAMETERS = "Parameters" OUTPUTS = "Outputs" INPUTS = "Inputs" READBACKS = "Readbacks" -PviGroupField = Literal["Parameters", "Outputs", "Inputs", "Readbacks"] +class NamedAttribute(namedtuple("NamedAttribute", "attribute_name attribute")): + attribute_name: str + attribute: AttrType -class Field(SubController): - def __init__( - self, - attribute_name: str | None, - attribute: AttrRW | AttrR | AttrW | None, - sub_fields: dict[str, FieldType] | None = None, - ): - """ - For controlling the field, sub fields can also be added. - attribute_name and attribute are optional since some fields - e.g won't contain a top level record, but only sub fields. - """ - super().__init__() - self.sub_fields = sub_fields or {} - self.attribute_name = attribute_name - - if attribute_name and attribute: + +class SubFieldController(SubController): + def __init__(self, attributes: list[NamedAttribute]): + for attribute_name, attribute in attributes: setattr(self, attribute_name, attribute) - for sub_field_name, sub_field in self.sub_fields.items(): - self.register_sub_controller( - PandaName(sub_field=sub_field_name).attribute_name, - sub_field - ) - async def update_value(self, value: str): - if self.attribute_name is None: - return +class Field: + def __init__( + self, + attribute_name: str, + attribute: AttrRW | AttrR | AttrW, + sub_field_controller: SubFieldController | None = None, + ): + self.sub_field_controller = sub_field_controller - attribute = getattr(self, self.attribute_name) - if isinstance(attribute, AttrW): - await attribute.process(value) - else: - await attribute.set(value) + self.block_attribute = ( + NamedAttribute(attribute_name=attribute_name, attribute=attribute) + if (attribute_name and attribute) + else None + ) class TableField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... + def __init__(self, panda_name: PandaName, table_field_info: TableFieldInfo): + # TODO: Make a table type. For now we'll leave this to an int. + table_field = AttrR(Float(), group=WidgetGroup.OUTPUTS.value) + super().__init__(panda_name.attribute_name, table_field) class TimeField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - - time_attr = AttrR(Float(), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, time_field_info: TimeFieldInfo): + time_attr = AttrR(Float(), group=WidgetGroup.PARAMETERS.value) # TODO: Find out how to add EGU and such super().__init__(panda_name.attribute_name, time_attr) class BitOutField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - bit_out_attr = AttrRW(Bool(znam="0", onam="1"), group=PviGroup.OUTPUTS) + def __init__(self, panda_name: PandaName, bit_out_field_info: BitOutFieldInfo): + bit_out_attr = AttrRW(Bool(znam="0", onam="1"), group=WidgetGroup.OUTPUTS.value) super().__init__(panda_name.attribute_name, bit_out_attr) class PosOutField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): + def __init__(self, panda_name: PandaName, pos_out_field_info: PosOutFieldInfo): # TODO add capture and dataset subfields - pos_out_attr = AttrR(Float(), group=PviGroup.OUTPUTS) + pos_out_attr = AttrR(Float(), group=WidgetGroup.OUTPUTS.value) super().__init__(panda_name.attribute_name, pos_out_attr) class ExtOutField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): + def __init__(self, panda_name: PandaName, ext_out_field_info: ExtOutFieldInfo): # TODO add capture and dataset subfields - super().__init__(None, None) + ext_out_field = AttrR(Float(), group=WidgetGroup.OUTPUTS.value) + super().__init__(panda_name.attribute_name, ext_out_field) class ExtOutBitsField(ExtOutField): def __init__( - self, panda_name: PandaName, description: str | None + self, panda_name: PandaName, ext_out_bits_field_info: ExtOutBitsFieldInfo ): # TODO add capture and dataset subfields - super().__init__(panda_name, description) + super().__init__(panda_name, ext_out_bits_field_info) class BitMuxField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - bit_mux_attr = AttrRW(String(), group=PviGroup.INPUTS) + def __init__(self, panda_name: PandaName, bit_mux_field_info: BitMuxFieldInfo): + bit_mux_attr = AttrRW(String(), group=WidgetGroup.INPUTS.value) super().__init__(panda_name.attribute_name, bit_mux_attr) class PosMuxField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - pos_mux_attr = AttrRW(String(), group=PviGroup.INPUTS) + def __init__(self, panda_name: PandaName, pos_mux_field_info: PosMuxFieldInfo): + pos_mux_attr = AttrRW(String(), group=WidgetGroup.INPUTS.value) super().__init__(panda_name.attribute_name, pos_mux_attr) + class UintParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - uint_param_attr = AttrR(Float(prec=0), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, uint_param_field_info: UintFieldInfo): + uint_param_attr = AttrR(Float(prec=0), group=WidgetGroup.PARAMETERS.value) super().__init__(panda_name.attribute_name, uint_param_attr) + class UintReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - uint_read_attr = AttrR(Float(prec=0), group=PviGroup.READBACKS) + def __init__(self, panda_name: PandaName, uint_read_field_info: UintFieldInfo): + uint_read_attr = AttrR(Float(prec=0), group=WidgetGroup.READBACKS.value) super().__init__(panda_name.attribute_name, uint_read_attr) class UintWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - uint_write_attr = AttrW(Float(prec=0), group=PviGroup.OUTPUTS) + def __init__(self, panda_name: PandaName, uint_write_field_info: UintFieldInfo): + uint_write_attr = AttrW(Float(prec=0), group=WidgetGroup.OUTPUTS.value) super().__init__(panda_name.attribute_name, uint_write_attr) class IntParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - uint_param_attr = AttrRW(Float(prec=0), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, int_param_field_info: FieldInfo): + uint_param_attr = AttrRW(Float(prec=0), group=WidgetGroup.PARAMETERS.value) super().__init__(panda_name.attribute_name, uint_param_attr) class IntReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - int_read_attr = AttrR(Int(), group=PviGroup.READBACKS) + def __init__(self, panda_name: PandaName, int_read_field_info: FieldInfo): + int_read_attr = AttrR(Int(), group=WidgetGroup.READBACKS.value) super().__init__(panda_name.attribute_name, int_read_attr) class IntWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - int_write_attr = AttrW(Int(), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, int_write_field_info: FieldInfo): + int_write_attr = AttrW(Int(), group=WidgetGroup.PARAMETERS.value) super().__init__(panda_name.attribute_name, int_write_attr) class ScalarParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - scalar_param_attr = AttrRW(Float(), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, scalar_param_field_info: ScalarFieldInfo): + scalar_param_attr = AttrRW(Float(), group=WidgetGroup.PARAMETERS.value) super().__init__(panda_name.attribute_name, scalar_param_attr) class ScalarReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - scalar_read_attr = AttrR(Float(), group=PviGroup.READBACKS) + def __init__(self, panda_name: PandaName, scalar_read_field_info: ScalarFieldInfo): + scalar_read_attr = AttrR(Float(), group=WidgetGroup.READBACKS.value) super().__init__(panda_name.attribute_name, scalar_read_attr) + class ScalarWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - scalar_read_attr = AttrR(Float(), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, scalar_write_field_info: ScalarFieldInfo): + scalar_read_attr = AttrR(Float(), group=WidgetGroup.PARAMETERS.value) super().__init__(panda_name.attribute_name, scalar_read_attr) class BitParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - bit_param_attr = AttrRW(Bool(znam="0", onam="1"), group=PviGroup.PARAMETERS) + def __init__(self, panda_name: PandaName, bit_param_field_info: FieldInfo): + bit_param_attr = AttrRW( + Bool(znam="0", onam="1"), group=WidgetGroup.PARAMETERS.value + ) super().__init__(panda_name.attribute_name, bit_param_attr) class BitReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - bit_read_attr = AttrR(Bool(znam="0", onam="1"), group=PviGroup.READBACKS) + def __init__(self, panda_name: PandaName, bit_read_field_info: FieldInfo): + bit_read_attr = AttrR( + Bool(znam="0", onam="1"), group=WidgetGroup.READBACKS.value + ) super().__init__(panda_name.attribute_name, bit_read_attr) + class BitWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - bit_write_attr = AttrW(Bool(znam="0", onam="1"), group=PviGroup.OUTPUTS) + def __init__(self, panda_name: PandaName, bit_write_field_info: FieldInfo): + bit_write_attr = AttrW( + Bool(znam="0", onam="1"), group=WidgetGroup.OUTPUTS.value + ) super().__init__(panda_name.attribute_name, bit_write_attr) class ActionReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): - action_read_attr = AttrW(Bool(znam="0", onam="1"), group=PviGroup.READBACKS) + def __init__(self, panda_name: PandaName, action_read_field_info: FieldInfo): + action_read_attr = AttrR( + Bool(znam="0", onam="1"), group=WidgetGroup.READBACKS.value + ) super().__init__(panda_name.attribute_name, action_read_attr) class ActionWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, action_write_field_info: FieldInfo): + action_write_attr = AttrW( + Bool(znam="0", onam="1"), group=WidgetGroup.OUTPUTS.value + ) + super().__init__(panda_name.attribute_name, action_write_attr) class LutParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, lut_param_field_info: FieldInfo): + lut_param_field = AttrRW(String(), group=WidgetGroup.PARAMETERS.value) + super().__init__(panda_name.attribute_name, lut_param_field) class LutReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, lut_read_field_info: FieldInfo): + lut_read_field = AttrR(String(), group=WidgetGroup.READBACKS.value) + super().__init__(panda_name.attribute_name, lut_read_field) class LutWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, lut_read_field_info: FieldInfo): + lut_write_field = AttrR(String(), group=WidgetGroup.OUTPUTS.value) + super().__init__(panda_name.attribute_name, lut_write_field) class EnumParamField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, enum_param_field_info: EnumFieldInfo): + self.allowed_values = enum_param_field_info.labels + enum_param_field = AttrRW(String(), group=WidgetGroup.PARAMETERS.value) + super().__init__(panda_name.attribute_name, enum_param_field) class EnumReadField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, enum_read_field_info: EnumFieldInfo): + enum_read_field = AttrR(String(), group=WidgetGroup.READBACKS.value) + super().__init__(panda_name.attribute_name, enum_read_field) class EnumWriteField(Field): - def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + def __init__(self, panda_name: PandaName, enum_write_field_info: EnumFieldInfo): + enum_write_field = AttrW(String(), group=WidgetGroup.OUTPUTS.value) + super().__init__(panda_name.attribute_name, enum_write_field) -class TimeSubTypeParamField(TimeField): +class TimeSubTypeParamField(Field): def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + self, panda_name: PandaName, time_subtype_param_field_info: SubtypeTimeFieldInfo + ): + time_subtype_param_field = AttrRW(Float(), group=WidgetGroup.PARAMETERS.value) + super().__init__(panda_name.attribute_name, time_subtype_param_field) -class TimeSubTypeReadField(TimeField): +class TimeSubTypeReadField(Field): def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + self, panda_name: PandaName, time_subtype_read_field_info: SubtypeTimeFieldInfo + ): + time_subtype_read_field = AttrR(Float(), group=WidgetGroup.READBACKS.value) + super().__init__(panda_name.attribute_name, time_subtype_read_field) -class TimeSubTypeWriteField(TimeField): +class TimeSubTypeWriteField(Field): def __init__( - self, panda_name: PandaName, description: str | None - ): ... - - ... + self, panda_name: PandaName, time_subtype_write_field_info: SubtypeTimeFieldInfo + ): + time_subtype_write_field = AttrW(Float(), group=WidgetGroup.OUTPUTS.value) + super().__init__(panda_name.attribute_name, time_subtype_write_field) FieldType = ( @@ -334,6 +305,7 @@ def __init__( | TimeSubTypeWriteField ) +# TODO: Change to a match statement so we can easily add a PCAP field type. FIELD_TYPE_TO_FASTCS_TYPE: dict[str, dict[str | None, type[FieldType]]] = { "table": {None: TableField}, "time": { diff --git a/src/fastcs_pandablocks/types/__init__.py b/src/fastcs_pandablocks/types/__init__.py index f7eadaf..2fe1977 100644 --- a/src/fastcs_pandablocks/types/__init__.py +++ b/src/fastcs_pandablocks/types/__init__.py @@ -1,4 +1,4 @@ -from .annotations import ResponseType +from .annotations import AttrType, ResponseType from .string_types import ( EPICS_SEPERATOR, PANDA_SEPERATOR, @@ -11,5 +11,6 @@ "EpicsName", "PANDA_SEPERATOR", "PandaName", + "AttrType", "ResponseType", ] diff --git a/src/fastcs_pandablocks/types/annotations.py b/src/fastcs_pandablocks/types/annotations.py index 1f2b5c2..35355e3 100644 --- a/src/fastcs_pandablocks/types/annotations.py +++ b/src/fastcs_pandablocks/types/annotations.py @@ -1,3 +1,4 @@ +from fastcs.attributes import AttrR, AttrRW, AttrW from pandablocks.responses import ( BitMuxFieldInfo, BitOutFieldInfo, @@ -29,3 +30,5 @@ | TimeFieldInfo | UintFieldInfo ) + +AttrType = AttrRW | AttrR | AttrW diff --git a/src/fastcs_pandablocks/types/string_types.py b/src/fastcs_pandablocks/types/string_types.py index 4d54ef7..e1fe0ac 100644 --- a/src/fastcs_pandablocks/types/string_types.py +++ b/src/fastcs_pandablocks/types/string_types.py @@ -37,7 +37,24 @@ def _format_with_seperator( def _to_python_attribute_name(string: str): - return string.replace("-", "_").lower() + return string.replace("-", "_") + + +def _choose_sub_pv(sub_pv_1: T, sub_pv_2: T) -> T: + if sub_pv_1 is not None and sub_pv_2 is not None: + if sub_pv_1 != sub_pv_2: + raise TypeError( + "Ambiguous pv elements on add " f"{sub_pv_1} and {sub_pv_2}" + ) + return sub_pv_2 or sub_pv_1 + + +def _check_eq(sub_pv_1: T, sub_pv_2: T) -> bool: + if sub_pv_1 is not None and sub_pv_2 is not None: + return sub_pv_1 == sub_pv_2 + elif sub_pv_1 and sub_pv_2 is None: + return False + return True @dataclass(frozen=True) @@ -77,6 +94,14 @@ def epics_name(self): sub_field=self.sub_field, ) + def __add__(self, other: PandaName) -> PandaName: + return PandaName( + 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), + ) + @cached_property def attribute_name(self) -> str: if self.sub_field: @@ -85,7 +110,7 @@ def attribute_name(self) -> str: return _to_python_attribute_name(self.field) if self.block: return _to_python_attribute_name(self.block) + ( - f"_{self.block_number}" if self.block_number is not None else "" + f"{self.block_number}" if self.block_number is not None else "" ) return "" @@ -150,15 +175,6 @@ def __add__(self, other: EpicsName) -> EpicsName: == EpicsName.from_string("PREFIX:BLOCK:FIELD") """ - def _choose_sub_pv(sub_pv_1: T, sub_pv_2: T) -> T: - if sub_pv_1 is not None and sub_pv_2 is not None: - if sub_pv_1 != sub_pv_2: - raise TypeError( - "Ambiguous pv elements on `EpicsName` add " - f"{sub_pv_1} and {sub_pv_2}" - ) - return sub_pv_2 or sub_pv_1 - return EpicsName( prefix=_choose_sub_pv(self.prefix, other.prefix), block=_choose_sub_pv(self.block, other.block), @@ -177,13 +193,6 @@ def __contains__(self, other: EpicsName) -> bool: (EpicsName(block="field1") in EpicsName("prefix:block1:field2")) == False """ - def _check_eq(sub_pv_1: T, sub_pv_2: T) -> bool: - if sub_pv_1 is not None and sub_pv_2 is not None: - return sub_pv_1 == sub_pv_2 - elif sub_pv_1 and sub_pv_2 is None: - return False - return True - return ( _check_eq(self.prefix, other.prefix) and _check_eq(self.block, other.block) diff --git a/tests/test_introspection.py b/tests/test_introspection.py new file mode 100644 index 0000000..e69de29