adds support for configuring commands and channels using YAML

Adds support for 'tango', 'exporter' and 'epics' sections to YAML
configuration files.

These section are used to configure Command and Channel objects for
hardware objects.

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:"

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())

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())

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())

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)