diff --git a/mxcubecore/HardwareRepository.py b/mxcubecore/HardwareRepository.py index 03b87049aa..b998f0994e 100644 --- a/mxcubecore/HardwareRepository.py +++ b/mxcubecore/HardwareRepository.py @@ -45,6 +45,7 @@ from mxcubecore.dispatcher import dispatcher from mxcubecore import BaseHardwareObjects from mxcubecore import HardwareObjectFileParser +from mxcubecore.protocols_config import setup_commands_channels from warnings import warn @@ -173,6 +174,8 @@ def load_from_yaml( # Set configuration with non-object properties. result._config = result.HOConfig(**config) + setup_commands_channels(result, configuration) + if _container is None: load_time = 1000 * (time.time() - start_time) msg1 = "Start loading contents:" diff --git a/mxcubecore/model/protocols/__init__.py b/mxcubecore/model/protocols/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/mxcubecore/model/protocols/epics.py b/mxcubecore/model/protocols/epics.py new file mode 100644 index 0000000000..c6c0e954be --- /dev/null +++ b/mxcubecore/model/protocols/epics.py @@ -0,0 +1,57 @@ +""" +Models the `epics` section of YAML hardware configuration file. + +Provides an API to read configured EPICS channels. +""" +from typing import Optional, Iterable, Tuple, Dict +from pydantic import BaseModel + + +class Channel(BaseModel): + """ + an EPICS channel configuration + """ + + suffix: Optional[str] + poll: Optional[int] + + +class Prefix(BaseModel): + """ + configuration of an EPICS prefix section + """ + + channels: Optional[Dict[str, Optional[Channel]]] + + def get_channels(self) -> Iterable[Tuple[str, Channel]]: + """ + get all channels configured for prefix + + This method will fill in optional configuration properties for a channel. + """ + + if self.channels is None: + return [] + + for channel_name, channel_config in self.channels.items(): + if channel_config is None: + channel_config = Channel() + + if channel_config.suffix is None: + channel_config.suffix = channel_name + + yield channel_name, channel_config + + +class EpicsConfig(BaseModel): + """ + the 'epics' section of the hardware object's YAML configuration file + """ + + __root__: Dict[str, Prefix] + + def get_prefixes(self) -> Iterable[Tuple[str, Prefix]]: + """ + get all prefixes specified in this section + """ + return list(self.__root__.items()) diff --git a/mxcubecore/model/protocols/exporter.py b/mxcubecore/model/protocols/exporter.py new file mode 100644 index 0000000000..a19139befa --- /dev/null +++ b/mxcubecore/model/protocols/exporter.py @@ -0,0 +1,84 @@ +""" +Models the `exporter` section of YAML hardware configuration file. + +Provides an API to read configured exporter channels and commands. +""" +from typing import Optional, Iterable, Tuple, Dict +from pydantic import BaseModel + + +class Command(BaseModel): + """ + an exporter command configuration + """ + + # name of the exporter device command + name: Optional[str] + + +class Channel(BaseModel): + """ + an exporter channel configuration + """ + + attribute: Optional[str] + + +class Address(BaseModel): + """ + configuration of an exporter end point + """ + + commands: Optional[Dict[str, Optional[Command]]] + channels: Optional[Dict[str, Optional[Channel]]] + + def get_commands(self) -> Iterable[tuple[str, Command]]: + """ + get all commands configured for this exporter address + + This method will fill in optional configuration properties the commands. + """ + + if self.commands is None: + return [] + + for command_name, command_config in self.commands.items(): + if command_config is None: + command_config = Command() + + if command_config.name is None: + command_config.name = command_name + + yield command_name, command_config + + def get_channels(self) -> Iterable[Tuple[str, Channel]]: + """ + get all channels configured for this exporter address + + This method will fill in optional configuration properties for channels. + """ + if self.channels is None: + return [] + + for channel_name, channel_config in self.channels.items(): + if channel_config is None: + channel_config = Channel() + + if channel_config.attribute is None: + channel_config.attribute = channel_name + + yield channel_name, channel_config + + +class ExporterConfig(BaseModel): + """ + the 'exporter' section of the hardware object's YAML configuration file + """ + + __root__: Dict[str, Address] + + def get_addresses(self) -> Iterable[Tuple[str, Address]]: + """ + get all exporter addresses specified in this section + """ + return list(self.__root__.items()) diff --git a/mxcubecore/model/protocols/tango.py b/mxcubecore/model/protocols/tango.py new file mode 100644 index 0000000000..8078b14b9f --- /dev/null +++ b/mxcubecore/model/protocols/tango.py @@ -0,0 +1,84 @@ +""" +Models the `tango` section of YAML hardware configuration file. + +Provides an API to read configured tango channels and commands. +""" +from typing import Optional, Iterable, Tuple, Dict +from pydantic import BaseModel + + +class Command(BaseModel): + """ + a tango command configuration + """ + + # name of the tango device command + name: Optional[str] + + +class Channel(BaseModel): + """ + a tango channel configuration + """ + + attribute: Optional[str] + poll: Optional[int] + + +class Device(BaseModel): + """ + configuration of a tango device + """ + + commands: Optional[Dict[str, Optional[Command]]] + channels: Optional[Dict[str, Optional[Channel]]] + + def get_commands(self) -> Iterable[Tuple[str, Command]]: + """ + get all commands configured for this device + + This method will fill in optional configuration properties for commands. + """ + if self.commands is None: + return [] + + for command_name, command_config in self.commands.items(): + if command_config is None: + command_config = Command() + + if command_config.name is None: + command_config.name = command_name + + yield command_name, command_config + + def get_channels(self) -> Iterable[Tuple[str, Channel]]: + """ + get all channels configured for this device + + This method will fill in optional configuration properties for a channel. + """ + if self.channels is None: + return [] + + for channel_name, channel_config in self.channels.items(): + if channel_config is None: + channel_config = Channel() + + if channel_config.attribute is None: + channel_config.attribute = channel_name + + yield channel_name, channel_config + + +class TangoConfig(BaseModel): + """ + the 'tango' section of the hardware object's YAML configuration file + """ + + __root__: Dict[str, Device] + + def get_tango_devices(self) -> Iterable[Tuple[str, Device]]: + """ + get all tango devices specified in this section + """ + return list(self.__root__.items()) diff --git a/mxcubecore/protocols_config.py b/mxcubecore/protocols_config.py new file mode 100644 index 0000000000..a48d7cd8c8 --- /dev/null +++ b/mxcubecore/protocols_config.py @@ -0,0 +1,135 @@ +""" +Provides an API to add Command and Channel objects to hardware objects, +as specified in it's YAML configuration file. + +See setup_commands_channels() function for details. +""" +from __future__ import annotations +from typing import Iterable, Callable +from mxcubecore.BaseHardwareObjects import HardwareObject + + +def _setup_tango_commands_channels(hwobj: HardwareObject, tango_config: dict): + """ + set up Tango Command and Channel objects + + parameters: + tango: the 'tango' section of the hardware object's configuration + """ + from mxcubecore.model.protocols.tango import TangoConfig, Device + + def setup_tango_device(device_name: str, device_config: Device): + # + # set-up commands + # + for command_name, command_config in device_config.get_commands(): + attrs = dict(type="tango", name=command_config.name, tangoname=device_name) + hwobj.add_command(attrs, command_name) + + # + # set-up channels + # + for channel_name, channel_config in device_config.get_channels(): + attrs = dict(type="tango", name=channel_name, tangoname=device_name) + if channel_config.poll: + attrs["polling"] = channel_config.poll + + hwobj.add_channel(attrs, channel_config.attribute) + + tango_cfg = TangoConfig.parse_obj(tango_config) + for device_name, device_config in tango_cfg.get_tango_devices(): + setup_tango_device(device_name, device_config) + + +def _setup_exporter_commands_channels(hwobj: HardwareObject, exporter_config: dict): + from mxcubecore.model.protocols.exporter import ExporterConfig, Address + + def setup_address(address: str, address_config: Address): + # + # set-up commands + # + for command_name, command_config in address_config.get_commands(): + attrs = dict( + type="exporter", exporter_address=address, name=command_config.name + ) + hwobj.add_command(attrs, command_name) + + # + # set-up channels + # + for channel_name, channel_config in address_config.get_channels(): + attrs = dict(type="exporter", exporter_address=address, name=channel_name) + hwobj.add_channel(attrs, channel_config.attribute) + + exp_cfg = ExporterConfig.parse_obj(exporter_config) + for address, address_config in exp_cfg.get_addresses(): + setup_address(address, address_config) + + +def _setup_epics_channels(hwobj: HardwareObject, epics_config: dict): + from mxcubecore.model.protocols.epics import EpicsConfig, Prefix + + def setup_prefix(prefix: str, prefix_config: Prefix): + # + # set-up channels + # + for channel_name, channel_config in prefix_config.get_channels(): + attrs = dict(type="epics", name=channel_name) + if channel_config.poll: + attrs["polling"] = channel_config.poll + + pv_name = f"{prefix}{channel_config.suffix}" + hwobj.add_channel(attrs, pv_name) + + epics_cfg = EpicsConfig.parse_obj(epics_config) + for prefix, prefix_config in epics_cfg.get_prefixes(): + setup_prefix(prefix, prefix_config) + + +def _protocol_handles(): + return { + "tango": _setup_tango_commands_channels, + "exporter": _setup_exporter_commands_channels, + "epics": _setup_epics_channels, + } + + +def _get_protocol_names() -> Iterable[str]: + """ + get names of all supported protocols + """ + return _protocol_handles().keys() + + +def _get_protocol_handler(protocol_name: str) -> Callable: + """ + get the callable that will set up commands and channels for a specific protocol + """ + return _protocol_handles()[protocol_name] + + +def _setup_protocol(hwobj: HardwareObject, config: dict, protocol: str): + """ + add the Command and Channel objects configured in the specified protocol section + + parameters: + protocol: name of the protocol to handle + """ + protocol_config = config.get(protocol) + if protocol_config is None: + # no configuration for this protocol + return + + _get_protocol_handler(protocol)(hwobj, protocol_config) + + +def setup_commands_channels(hwobj: HardwareObject, config: dict): + """ + add the Command and Channel objects to a hardware object, as specified in the config + + parameters: + hwobj: hardware object where to add Command and Channel objects + config: the complete hardware object configuration, i.e. parsed YAML file as dict + """ + for protocol in _get_protocol_names(): + _setup_protocol(hwobj, config, protocol)