diff --git a/mxcubecore/HardwareRepository.py b/mxcubecore/HardwareRepository.py index c7d6317a18..19beddde13 100644 --- a/mxcubecore/HardwareRepository.py +++ b/mxcubecore/HardwareRepository.py @@ -54,6 +54,7 @@ HardwareObjectFileParser, ) from mxcubecore.dispatcher import dispatcher +from mxcubecore.protocols_config import setup_commands_channels from mxcubecore.utils.conversion import ( make_table, string_types, @@ -184,6 +185,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..cc4d889890 --- /dev/null +++ b/mxcubecore/model/protocols/epics.py @@ -0,0 +1,54 @@ +"""Models the `epics` section of YAML hardware configuration file. + +Provides an API to read configured EPICS channels. +""" + +from typing import ( + Dict, + Iterable, + Optional, + Tuple, +) + +from pydantic import BaseModel + + +class Channel(BaseModel): + """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..004ed4cf50 --- /dev/null +++ b/mxcubecore/model/protocols/exporter.py @@ -0,0 +1,78 @@ +"""Models the `exporter` section of YAML hardware configuration file. + +Provides an API to read configured exporter channels and commands. +""" + +from typing import ( + Dict, + Iterable, + Optional, + Tuple, +) + +from pydantic import BaseModel + + +class Command(BaseModel): + """Exporter command configuration.""" + + # name of the exporter device command + name: Optional[str] + + +class Channel(BaseModel): + """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..1d758175aa --- /dev/null +++ b/mxcubecore/model/protocols/tango.py @@ -0,0 +1,79 @@ +"""Models the `tango` section of YAML hardware configuration file. + +Provides an API to read configured tango channels and commands. +""" + +from typing import ( + Dict, + Iterable, + Optional, + Tuple, +) + +from pydantic import BaseModel + + +class Command(BaseModel): + """Tango command configuration.""" + + # name of the tango device command + name: Optional[str] + + +class Channel(BaseModel): + """Tango channel configuration.""" + + attribute: Optional[str] + polling_period: Optional[int] + timeout: 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..3bcaa50d71 --- /dev/null +++ b/mxcubecore/protocols_config.py @@ -0,0 +1,147 @@ +""" +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 ( + Callable, + Iterable, +) + +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 ( + Device, + TangoConfig, + ) + + 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.polling_period: + attrs["polling"] = channel_config.polling_period + + if channel_config.timeout: + attrs["timeout"] = channel_config.timeout + + 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 ( + Address, + ExporterConfig, + ) + + 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)