Source code for herosdevices.hardware.menlo.functional_layer

"""Base class for drivers scoped to one `functionalLayer` sub-tree of a Menlo Systems optical frequency comb."""

from typing import Any

from .ofc import OFC, poll_observables




class FunctionalLayerModule:
    """Base for drivers that address one sub-tree of an OFC's `functionalLayer`, relative to a fixed base path.

    Does not open its own connection: `ofc` is meant to be an already-running
    :py:class:`~herosdevices.hardware.menlo.ofc.OFC` HERO, injected as a dependency (e.g. via boss's
    `"ofc": "$device_menlo_ofc"` reference in the config), so multiple modules share the one physical
    websocket connection to the OFC instead of each opening a duplicate connection to the same device.

    Args:
        ofc: The OFC HERO this module belongs to.
        base_path: Dotted path to this module's settings object, e.g. `functionalLayer.cw1112_1Settings` or
            `functionalLayer.rrSettings`.
        observables: Additional observables to poll for `observable_data`, merged on top of
            :py:attr:`DEFAULT_OBSERVABLES` (an entry here with the same name overrides the default). See
            :py:meth:`~herosdevices.hardware.menlo.ofc.OFC.get_node` for the "path"/"unit" dict shape.
    """

    DEFAULT_OBSERVABLES: dict[str, dict[str, str]] = {}

    def __init__(self, ofc: OFC, base_path: str, observables: dict[str, dict[str, str]] | None = None) -> None:
        self.ofc = ofc
        self.base_path = base_path
        self.observables = self.DEFAULT_OBSERVABLES | (observables or {})

    def _path(self, relative_path: str) -> str:
        return f"{self.base_path}.{relative_path}"



    def get(self, relative_path: str) -> Any:
        """Read the current value of a node relative to this module's settings object.

        Use this for module properties not already exposed as a named attribute. See
        :py:meth:`~herosdevices.hardware.menlo.ofc.OFC.explore` to find the available relative paths,
        e.g. `ofc.explore(module.base_path, depth=2)`.

        Args:
            relative_path: Dotted path relative to `base_path`.

        Returns:
            The current value of the node.
        """
        return self.ofc.get_node(self._path(relative_path))




    def set(self, relative_path: str, value: Any) -> None:
        """Set the value of a node relative to this module's settings object.

        Args:
            relative_path: Dotted path relative to `base_path`, see :py:meth:`get`.
            value: Value to set.
        """
        self.ofc.set_node(self._path(relative_path), value)


    def _observable_data(self) -> dict[str, tuple[Any, str]]:
        """Read every path in `observables` (`DEFAULT_OBSERVABLES` merged with any instance overrides).

        Tolerates an unreachable OFC. Defined here (ahead of `RFSource` in every subclass's MRO) so it
        supersedes `RFSource`'s cached-value `_observable_data` for every functional-layer module, regardless
        of a subclass's inheritance order.

        Returns:
            A dict mapping each observable's name to its `(value, unit)` pair. Empty if the OFC is not
            currently reachable.
        """
        return poll_observables(self.observables, self.ofc._ensure_connected, self.get, self.base_path)