"""HERO driver for Menlo Systems frequency combs exposed via QWebChannel."""
import re
import threading
from collections.abc import Callable
from typing import Any
from herosdevices.helper import log, mark_driver, render_tree
from .qwebchannel import QWEBCHANNEL_DEFAULT_PORT, QWebChannelConnection
try:
from pywebchannel.qwebchannel import QObject as WebChannelObject
except ModuleNotFoundError:
WebChannelObject = object
log.exception(
"Could not import the 'pywebchannel' module, required for the Menlo OFC driver. "
"Install it with `pip install git+https://github.com/MenloSystems/pywebchannel`"
)
_TOKEN_RE = re.compile(r'\["(?P<key>[^"]+)"\]|(?P<attr>[^.\[\]]+)')
[docs]
def poll_observables(
observables: dict[str, dict[str, str]],
ensure_connected: Callable[[], None],
get_value: Callable[[str], Any],
context: str,
) -> dict[str, tuple[Any, str]]:
"""Read a dict of `{name: {"path", "unit"}}` observables, tolerating an unreachable device.
Shared by :py:class:`OFC` and :py:class:`~herosdevices.hardware.menlo.functional_layer.FunctionalLayerModule`,
whose `_observable_data` implementations both poll the device on demand rather than react to pushed
updates: the device is contacted once per poll (not once per observable), and a single unreachable device
or a single failing path does not stop the rest of the observables from being read.
Args:
observables: Maps an observable's name to a dict with keys "path" (passed to `get_value`) and "unit".
ensure_connected: Called once before reading any observable; may raise if the device is unreachable.
get_value: Reads the current value for one observable's "path".
context: Identifies the device/module in the log message if it is unreachable.
Returns:
A dict mapping each observable's name to its `(value, unit)` pair. Empty if `ensure_connected` raised.
"""
try:
ensure_connected()
except Exception as e: # noqa: BLE001
log.warning("Device at %s is not reachable, skipping this poll (%s)", context, e)
return {}
data = {}
for name, desc in observables.items():
try:
data[name] = (get_value(desc["path"]), desc.get("unit", ""))
except Exception: # noqa: BLE001
log.exception("Failed to read observable '%s' (%s)", name, desc["path"])
return data
# framework-level members every QWebChannel remote object carries, not useful for exploring the device tree
_EXPLORE_IGNORE = {"destroyed"}
def _list_members(node: WebChannelObject) -> list[str]:
"""List the property/method/signal names of a remote object relevant for tree exploration.
Filters out the `destroyed` signal every QWebChannel object carries and the auto-generated
`<property>Changed` notify signals, which just duplicate the property names already listed.
"""
return sorted(name for name in dir(node) if name not in _EXPLORE_IGNORE and not name.endswith("Changed"))
def _tokenize_path(path: str) -> list[tuple[str, str]]:
"""Split a node path into a list of (kind, key) tokens.
Supports plain attribute access (`a.b.c`) and dict-style indexing with string keys
(`modules["SYNCRO3U"].channel03`), matching the addressing syntax returned by :py:meth:`OFC.explore`.
"item" tokens are resolved with `[key]`, "attr" tokens with `getattr`.
"""
tokens = []
for match in _TOKEN_RE.finditer(path):
if match.group("key") is not None:
tokens.append(("item", match.group("key")))
else:
tokens.append(("attr", match.group("attr")))
return tokens
[docs]
@mark_driver(
name="Menlo OFC",
info="Menlo Systems frequency comb, connected via QWebChannel over websocket",
state="alpha",
requires={
"websocket": "websocket-client",
"pywebchannel": "git+https://github.com/MenloSystems/pywebchannel",
},
additional_docs=["/tutorials/menlo_ofc.rst"],
product_page="https://www.menlosystems.com/",
)
class OFC:
"""Driver for a Menlo Systems optical frequency comb (OFC) exposed via the QWebChannel websocket interface.
The OFC exposes its full control/status tree (functional layer, modules, settings, ...) via Qt's
WebChannel protocol. Because that tree is deep and firmware-dependent, individual nodes are addressed by
dotted path strings (see :py:meth:`get_node`) instead of being declared as fixed class attributes. Use
:py:meth:`explore` interactively to discover which paths are available on a given device, then list the
ones you want polled in `observables`.
Values are pushed by the device and kept in a local cache as soon as the connection is established, so
:py:meth:`get_node` and :py:meth:`_observable_data` never trigger network traffic themselves.
Args:
host: Hostname or IP address of the OFC's QWebChannel websocket endpoint.
port: Port of the websocket endpoint.
user: Username used for authentication.
password: Password used for authentication.
timeout: Seconds to wait for a single connection attempt, see :py:class:`QWebChannelConnection`.
reconnect_cooldown: Minimum seconds between two connection attempts, see
:py:class:`QWebChannelConnection`. Keeps a prolonged outage (e.g. the OFC being powered off for
half an hour) from causing a reconnect attempt on every single poll tick; the OFC is picked back
up automatically the next time it is read after coming back online, no restart needed.
observables: Additional observables to poll, merged on top of :py:attr:`DEFAULT_OBSERVABLES` (an
entry here with the same name overrides the default) rather than replacing it. Each entry maps
the name under which a value is emitted with the `observable_data` event to a dict with keys
"path" (dotted node path, see :py:meth:`get_node`) and "unit". The repetition-rate, CEO, and
oscillator modules have their own default observables instead, see
:py:class:`~herosdevices.hardware.menlo.RepetitionRate`,
:py:class:`~herosdevices.hardware.menlo.CEO`, and
:py:class:`~herosdevices.hardware.menlo.Oscillator`.
"""
DEFAULT_OBSERVABLES: dict[str, dict[str, str]] = {
"motor_position": {"path": 'modules["lac"].motor1_positionActual', "unit": "Step"},
"motor_setpoint": {"path": 'modules["lac"].motor1_positionSetpoint', "unit": "Step"},
}
def __init__(
self,
host: str,
port: int = QWEBCHANNEL_DEFAULT_PORT,
user: str = "guest",
password: str = "",
timeout: float = 5.0,
reconnect_cooldown: float = 30.0,
observables: dict[str, dict[str, str]] | None = None,
) -> None:
self.host = host
self.observables = self.DEFAULT_OBSERVABLES | (observables or {})
self.connection = QWebChannelConnection(
host, port, user=user, password=password, timeout=timeout, reconnect_cooldown=reconnect_cooldown
)
def _ensure_connected(self) -> None:
"""Connect if not already connected.
`FunctionalLayerModule` calls this on `self.ofc`, which can be a local Python reference or a `RemoteHERO`.
A boss deployment must mark this method `force_remote` via boss's `extra_decorators` config:
.. code-block:: json
"extra_decorators": [["_ensure_connected", "heros.inspect.force_remote"]]
"""
if not self.connection.is_ready():
self.connection.connect()
def _resolve(self, tokens: list[tuple[str, str]]) -> Any:
node = self.connection.root
for kind, key in tokens:
node = node[key] if kind == "item" else getattr(node, key)
return node
[docs]
def get_node(self, path: str) -> Any:
"""Read the current (cached) value of a node in the OFC's control tree.
Args:
path: Dotted path to the node, e.g. `functionalLayer.rrSettings.repetitionRate.rrCounterRepRate`
or, for dict-valued nodes such as `modules`, `modules["SYNCRO3U"].functionalLayer...`.
Returns:
The current value of the node.
"""
self._ensure_connected()
return self._resolve(_tokenize_path(path))
[docs]
def set_node(self, path: str, value: Any) -> None:
"""Set the value of a node in the OFC's control tree.
Args:
path: Dotted path to the node, see :py:meth:`get_node`.
value: Value to set.
"""
self._ensure_connected()
tokens = _tokenize_path(path)
parent = self._resolve(tokens[:-1])
kind, key = tokens[-1]
if kind == "item":
parent[key] = value
else:
setattr(parent, key, value)
[docs]
def call_method(self, path: str, *args: Any, timeout: float | None = None) -> Any:
"""Call a method on a node in the OFC's control tree and wait for its result.
QWebChannel method calls are inherently asynchronous (the remote object's generated method wrapper
never returns a value, it only accepts a callback for the result). This wraps that callback in a
blocking wait so `call_method` behaves like a normal synchronous function call.
Args:
path: Dotted path to the method, see :py:meth:`get_node`. E.g.
`functionalLayer.rrSettings.mainControls.unlockHere`.
*args: Positional arguments to pass to the method.
timeout: Seconds to wait for the method's response. Defaults to the connection's own `timeout`.
Returns:
The method's return value.
Raises:
TimeoutError: No response was received within `timeout` seconds.
"""
self._ensure_connected()
tokens = _tokenize_path(path)
parent = self._resolve(tokens[:-1])
kind, key = tokens[-1]
method = parent[key] if kind == "item" else getattr(parent, key)
done = threading.Event()
result_box: list[Any] = []
def _on_result(result: Any) -> None:
result_box.append(result)
done.set()
method(*args, _on_result)
if not done.wait(timeout if timeout is not None else self.connection.timeout):
msg = f"Timed out waiting for response calling '{path}'"
raise TimeoutError(msg)
return result_box[0]
[docs]
def explore(self, path: str = "", depth: int = 1) -> Any:
"""Explore the OFC's control/status node tree starting at `path`.
Use this interactively to find the dotted paths to put into `observables`, e.g.
`ofc.explore("functionalLayer.rrSettings", depth=2)`.
Args:
path: Dotted path to start exploring from (same syntax as :py:meth:`get_node`). Defaults to the
root object.
depth: How many levels of children to expand. Children beyond this depth are listed by name only,
without expanding further.
Returns:
A nested dictionary mapping child names to either their value (leaf), a list of child names
(unexpanded branch), or another such dictionary (expanded branch).
"""
self._ensure_connected()
node = self.get_node(path) if path else self.connection.root
return self._explore_node(node, depth)
def _explore_node(self, node: Any, depth: int) -> Any:
if isinstance(node, dict):
return {key: self._explore_child(value, depth) for key, value in node.items()}
if isinstance(node, WebChannelObject):
return {name: self._explore_child(self._safe_getattr(node, name), depth) for name in _list_members(node)}
return node
def _explore_child(self, value: Any, depth: int) -> Any:
if depth <= 0:
if isinstance(value, dict):
return sorted(value.keys())
if isinstance(value, WebChannelObject):
return _list_members(value)
return value
return self._explore_node(value, depth - 1)
@staticmethod
def _safe_getattr(node: WebChannelObject, name: str) -> Any:
try:
return getattr(node, name)
except Exception: # noqa: BLE001
return None
def _observable_data(self) -> dict[str, tuple[Any, str]]:
return poll_observables(self.observables, self._ensure_connected, self.get_node, self.host)