From 1a495afe4c931ea4bddcde191f30a1c6d52073c1 Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:49:39 -0700 Subject: [PATCH 1/6] Add Inficon driver submodule and libby daemon deps --- .gitmodules | 3 +++ pyproject.toml | 4 ++++ src/driver/inficon | 1 + 3 files changed, 8 insertions(+) create mode 160000 src/driver/inficon diff --git a/.gitmodules b/.gitmodules index 7b4cb61..184f4d5 100644 --- a/.gitmodules +++ b/.gitmodules @@ -7,3 +7,6 @@ [submodule "src/driver/lris2_csu"] path = src/driver/lris2_csu url = git@github.com:CaltechOpticalObservatories/lris2-csu.git +[submodule "src/driver/inficon"] + path = src/driver/inficon + url = https://github.com/COO-Utilities/inficon diff --git a/pyproject.toml b/pyproject.toml index c9dd99e..ca4efbd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,10 @@ authors = [ ] license = {text = "MIT"} readme = "README.md" +dependencies = [ + "pyserial", + "pyyaml", +] [project.optional-dependencies] dev = [ diff --git a/src/driver/inficon b/src/driver/inficon new file mode 160000 index 0000000..d713d3c --- /dev/null +++ b/src/driver/inficon @@ -0,0 +1 @@ +Subproject commit d713d3cd81d103eee918f5e7c85049368e348c0a From efd85f82e20c2aa4f7f7cee5a0a80710b19c0bb2 Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:52:47 -0700 Subject: [PATCH 2/6] Add libby-based lris2-daemon base class and yaml config loader --- src/config.py | 202 +++++++++++++++++++++++++++++++ src/daemon.py | 324 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 526 insertions(+) create mode 100644 src/config.py create mode 100644 src/daemon.py diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..efb87f1 --- /dev/null +++ b/src/config.py @@ -0,0 +1,202 @@ +""" +Configuration ingestion daemons. +""" + +from __future__ import annotations # for Python 3.9 compatibility +from pathlib import Path +from typing import Any, Dict, List, Optional +import yaml + + +class ConfigError(Exception): + """Raised when configuration loading or validation fails.""" + pass + + +# Known HispecDaemon attributes that map directly from config +DAEMON_ATTRS = frozenset({ + "peer_id", + "bind", + "address_book", + "discovery_enabled", + "discovery_interval_s", + "transport", + "rabbitmq_url", + "group_id", +}) + + +def load_file(path: str | Path) -> Dict[str, Any]: + """ + Load a configuration file (YAML). + + Args: + path: Path to the config file + + Returns: + Parsed configuration dictionary + + Raises: + ConfigError: If the file cannot be loaded or parsed + """ + path = Path(path) + + if not path.exists(): + raise ConfigError(f"Config file not found: {path}") + + try: + with open(path, "r") as f: + return yaml.safe_load(f) or {} + except Exception as e: + raise ConfigError(f"Failed to load config from {path}: {e}") + + +def extract_daemon_config( + full_config: Dict[str, Any], + daemon_id: str, +) -> Dict[str, Any]: + """ + Extract configuration for a specific daemon from a subsystem config. + + Merges subsystem-level defaults with daemon-specific overrides. + + Args: + full_config: Full subsystem configuration + daemon_id: Identifier of the daemon to extract + + Returns: + Merged configuration for the specific daemon + + Raises: + ConfigError: If the daemon is not found in the config + """ + daemons = full_config.get("daemons", {}) + + if daemon_id not in daemons: + available = list(daemons.keys()) if daemons else [] + raise ConfigError( + f"Daemon '{daemon_id}' not found in config. " + f"Available daemons: {available}" + ) + + # Start with subsystem-level defaults (excluding 'daemons' key) + result = {k: v for k, v in full_config.items() if k != "daemons"} + + # Merge daemon-specific config (overrides subsystem defaults) + daemon_config = daemons[daemon_id] + if daemon_config: + _deep_merge(result, daemon_config) + + # Ensure peer_id is set + if "peer_id" not in result: + result["peer_id"] = daemon_id + + return result + + +def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> None: + """Deep merge override into base (modifies base in place).""" + for key, value in override.items(): + if ( + key in base + and isinstance(base[key], dict) + and isinstance(value, dict) + ): + _deep_merge(base[key], value) + else: + base[key] = value + + +def list_daemons(config: Dict[str, Any]) -> List[str]: + """ + List all daemon IDs defined in a subsystem config. + + Args: + config: Subsystem configuration dictionary + + Returns: + List of daemon identifiers + """ + return list(config.get("daemons", {}).keys()) + + +def is_subsystem_config(config: Dict[str, Any]) -> bool: + """Check if a config dict is a subsystem config (has 'daemons' key).""" + return "daemons" in config + + +class DaemonConfigLoader: + """ + Helper class for loading daemon configurations. + + Usage: + loader = DaemonConfigLoader("config/hsfei.yaml") + + # For subsystem configs with multiple daemons + for daemon_id in loader.daemon_ids: + config = loader.get_daemon_config(daemon_id) + daemon = MyDaemon.from_config(config) + + # Or load a single daemon directly + config = loader.get_daemon_config("pickoff1", env_prefix="HISPEC_") + """ + + def __init__(self, path: str | Path): + """ + Initialize loader with a config file path. + + Args: + path: Path to the config file (YAML or JSON) + """ + self.path = Path(path) + self._config: Optional[Dict[str, Any]] = None + + @property + def config(self) -> Dict[str, Any]: + """Lazily load and return the raw configuration.""" + if self._config is None: + self._config = load_file(self.path) + return self._config + + @property + def is_subsystem(self) -> bool: + """Check if this is a subsystem config with multiple daemons.""" + return is_subsystem_config(self.config) + + @property + def subsystem(self) -> Optional[str]: + """Get the subsystem name, if defined.""" + return self.config.get("subsystem") + + @property + def daemon_ids(self) -> List[str]: + """List all daemon IDs in this config.""" + if self.is_subsystem: + return list_daemons(self.config) + # Single daemon config - use peer_id or filename + peer_id = self.config.get("peer_id", self.path.stem) + return [peer_id] + + def get_daemon_config( + self, + daemon_id: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Get configuration for a specific daemon. + + Args: + daemon_id: Daemon identifier (required for subsystem configs, + optional for single-daemon configs) + + Returns: + Merged configuration dictionary + """ + if self.is_subsystem: + if daemon_id is None: + raise ConfigError( + "daemon_id required for subsystem configs. " + f"Available: {self.daemon_ids}" + ) + return extract_daemon_config(self.config, daemon_id) + else: + return self.config.copy() diff --git a/src/daemon.py b/src/daemon.py new file mode 100644 index 0000000..1627a5d --- /dev/null +++ b/src/daemon.py @@ -0,0 +1,324 @@ +""" LRIS2 daemon base class """ +from __future__ import annotations # for Python 3.9 compatibility +import json +import logging +from dataclasses import is_dataclass, asdict +import collections.abc as cabc +import signal, sys, threading, time +from typing import Any, Callable, Dict, Iterable, List, Optional, Type + +from libby import Keyword, Libby +from . import config as cfg + +Payload = Dict[str, Any] +RPCHandler = Callable[[Payload], Dict[str, Any]] +EvtHandler = Callable[[Payload], None] + +class Lris2Daemon: + """ + Base daemon class for Libby peers with support for multiple transports. + + ZMQ Usage: + class MyPeer(Lris2Daemon): + peer_id = "my-peer" + bind = "tcp://*:5555" + address_book = {"other-peer": "tcp://localhost:5556"} + + services = {"echo": lambda payload: {"echo": payload}} + topics = {"alerts": lambda payload: print(payload)} + + RabbitMQ Usage: + class MyPeer(Lris2Daemon): + transport = "rabbitmq" + peer_id = "my-peer" + rabbitmq_url = "amqp://localhost" # optional, defaults to this + + services = {"echo": lambda payload: {"echo": payload}} + topics = {"alerts": lambda payload: print(payload)} + + Note: RabbitMQ doesn't need bind or address_book since routing is + handled automatically by the broker. + """ + # simple attributes users set + peer_id: Optional[str] = None + bind: Optional[str] = None + address_book: Optional[Dict[str, str]] = None + discovery_enabled: bool = False + discovery_interval_s: float = 5.0 + + # transport selection: "zmq" or "rabbitmq (default)" + transport: str = "rabbitmq" + rabbitmq_url: Optional[str] = None + group_id: Optional[str] = None + # internal config + _config: Dict[str, Any] = {} + + # payload-only handlers + services: Dict[str, RPCHandler] = {} + topics: Dict[str, EvtHandler] = {} + + def __init__(self) -> None: + # Ensure per-instance handler tables + if type(self).services is self.services: + self.services = {} + if type(self).topics is self.topics: + self.topics = {} + + # Config ingestion + @classmethod + def from_config_file( + cls: Type["Lris2Daemon"], + path: str, + daemon_id: Optional[str] = None, + ) -> "Lris2Daemon": + """ + Build a daemon from a YAML config file. + + Args: + path: Path to yaml config file + daemon_id: For subsystem configs, which daemon to instantiate. + If None and config has multiple daemons, raises error. + + Returns: + Configured daemon instance + """ + loader = cfg.DaemonConfigLoader(path) + + # For subsystem configs, daemon_id is required unless only one daemon + if loader.is_subsystem: + if daemon_id is None: + if len(loader.daemon_ids) == 1: + daemon_id = loader.daemon_ids[0] + else: + raise cfg.ConfigError( + f"Subsystem config has multiple daemons: {loader.daemon_ids}. " + f"Specify daemon_id parameter." + ) + + config_dict = loader.get_daemon_config(daemon_id) + return cls.from_config(config_dict) + + @classmethod + def from_config(cls: Type["Lris2Daemon"], config: Dict[str, Any]) -> "Lris2Daemon": + """ + Build a daemon from a configuration dictionary. + + Known daemon attributes are mapped directly to instance attributes. + + Args: + config: Configuration dictionary + + Returns: + Configured daemon instance + """ + instance = cls() + + # Map known daemon attributes directly + for attr in cfg.DAEMON_ATTRS: + if attr in config: + setattr(instance, attr, config[attr]) + + # Store full config for subclass access + instance._config = config + + # Setup logging + instance._setup_logging() + + return instance + + def _setup_logging(self) -> None: + """Configure logging from config or defaults.""" + log_config = self._config.get("logging", {}) + level_str = log_config.get("level", "INFO").upper() + level = getattr(logging, level_str, logging.INFO) + log_file = log_config.get("file") + + logging.basicConfig( + level=level, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + filename=log_file, + ) + + self.logger = logging.getLogger(self.peer_id or self.__class__.__name__) + + def get_config(self, key: str, default: Any = None) -> Any: + """ + Get a value from the daemon's configuration. + + Args: + key: Config key (supports dot notation for nested keys, e.g., "hardware.ip_address") + default: Default value if key not found + + Returns: + Config value or default + """ + keys = key.split(".") + value = self._config + for k in keys: + if isinstance(value, dict) and k in value: + value = value[k] + else: + return default + return value + + + # optional hooks + def on_start(self, libby: Libby) -> None: ... + def on_stop(self, libby: Optional[Libby] = None) -> None: ... + def on_hello(self, libby: Libby) -> None: ... + def on_event(self, topic: str, msg) -> None: + print(f"[{self.__class__.__name__}] {topic}: {msg.env.payload}") + + # config getters + def config_peer_id(self) -> str: return self.peer_id or self._must("peer_id") + def config_bind(self) -> str: return self.bind or self._must("bind") + def config_rabbitmq_url(self) -> str: return self.rabbitmq_url or "amqp://localhost" + def config_group_id(self) -> Optional[str]: return self.group_id + def config_address_book(self) -> Dict[str, str]: return self.address_book if self.address_book is not None else {} + def config_discovery_enabled(self) -> bool: return bool(self.discovery_enabled) + def config_discovery_interval_s(self) -> float: return float(self.discovery_interval_s) + def config_rpc_keys(self) -> List[str]: return list(self.services.keys()) + def config_subscriptions(self) -> List[str]: return list(self.topics.keys()) + + # user-facing helpers + def add_service(self, key: str, fn: RPCHandler) -> None: + self.services[key] = fn + if hasattr(self, "libby"): self._register_services({key: fn}) + + def add_services(self, mapping: Dict[str, RPCHandler]) -> None: + self.services.update(mapping) + if hasattr(self, "libby"): self._register_services(mapping) + + def register_keyword(self, keyword: Keyword) -> None: + """Register a Keyword; delegates to the underlying Libby instance.""" + self.libby.register_keyword(keyword) + + def register_keywords(self, keywords: Iterable[Keyword]) -> None: + """Register many keywords in one libby call; call from on_start or later.""" + self.libby.register_keywords(keywords) + + @property + def keyword_registry(self): + """KeywordRegistry on the underlying Libby instance.""" + return self.libby.keyword_registry + + def add_topic(self, topic: str, fn: EvtHandler) -> None: + self.topics[topic] = fn + if hasattr(self, "libby"): + self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) + self.libby.subscribe(topic) + + def add_topics(self, mapping: Dict[str, EvtHandler]) -> None: + self.topics.update(mapping) + if hasattr(self, "libby"): + for topic, fn in mapping.items(): + self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) + self.libby.subscribe(*mapping.keys()) + + # internals + def _must(self, name: str): + raise NotImplementedError(f"Set `{name}` or override config_{name}()") + + def _service_adapter(self, fn): + def adapter(user_payload: dict, _ctx: dict) -> dict: + try: + result = fn(user_payload) # user returns ANYTHING + return self.payload(result) # we "shove it into payload" for them + except Exception as ex: + return {"ok": False, "error": str(ex)} + return adapter + + def _register_services(self, mapping: Dict[str, RPCHandler]) -> None: + for key, fn in mapping.items(): + self.libby.serve_keys([key], self._service_adapter(fn)) + + def build_libby(self) -> Libby: + """Build Libby instance with selected transport.""" + if self.transport == "rabbitmq": + return Libby.rabbitmq( + self_id=self.config_peer_id(), + rabbitmq_url=self.config_rabbitmq_url(), + keys=[], + callback=None, + group_id=self.config_group_id(), + ) + else: + # Default to ZMQ + return Libby.zmq( + self_id=self.config_peer_id(), + bind=self.config_bind(), + address_book=self.config_address_book(), + keys=[], callback=None, # register per-key + discover=self.config_discovery_enabled(), + discover_interval_s=self.config_discovery_interval_s(), + hello_on_start=True, + group_id=self.config_group_id(), + ) + + def serve(self) -> None: + stop_evt = threading.Event() + def _sig(_s, _f): stop_evt.set() + signal.signal(signal.SIGINT, _sig) + signal.signal(signal.SIGTERM, _sig) + + try: + self.libby = self.build_libby() + except Exception as ex: + print(f"[{self.__class__.__name__}] failed to start: {ex}", file=sys.stderr) + raise + + if self.services: + self._register_services(self.services) + if self.topics: + for topic, fn in self.topics.items(): + self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) + self.libby.subscribe(*self.topics.keys()) + + # discovery hello + hooks + try: + if self.config_discovery_enabled(): + self.libby.hello() + self.on_hello(self.libby) + except Exception: + pass + + try: + self.on_start(self.libby) + except Exception as ex: + print(f"[{self.__class__.__name__}] on_start error: {ex}", file=sys.stderr) + + keywords_to_register = self.libby.keyword_registry.drain() + if keywords_to_register: + self.libby.register_keywords(keywords_to_register) + + if self.transport == "rabbitmq": + print(f"[{self.__class__.__name__}] up: id={self.config_peer_id()} transport=rabbitmq url={self.rabbitmq_url}") + else: + print(f"[{self.__class__.__name__}] up: id={self.config_peer_id()} bind={self.config_bind()}") + try: + while not stop_evt.is_set(): time.sleep(0.5) + finally: + try: self.on_stop() + except Exception: pass + self.libby.stop() + print(f"[{self.__class__.__name__}] stopped") + + def payload(self, value=None, /, **extra) -> dict: + if value is None: + out = {} + elif is_dataclass(value): + out = asdict(value) + elif isinstance(value, cabc.Mapping): + out = dict(value) + else: + out = {"data": value} + + if extra: + out.update(extra) + + try: + json.dumps(out) + except TypeError as e: + raise ValueError(f"Payload not JSON-serializable: {e}") from e + + return out \ No newline at end of file From 3df9831eae53a3b99ac876b533fa51e66454c8a3 Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:53:15 -0700 Subject: [PATCH 3/6] Add Inficon vacuum gauge daemon --- config/lris2_vacuum.yaml | 21 ++++ daemons/vacuum-gauge | 228 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) create mode 100644 config/lris2_vacuum.yaml create mode 100755 daemons/vacuum-gauge diff --git a/config/lris2_vacuum.yaml b/config/lris2_vacuum.yaml new file mode 100644 index 0000000..105a361 --- /dev/null +++ b/config/lris2_vacuum.yaml @@ -0,0 +1,21 @@ +# +# Usage: +# daemons/vacuum-gauge -c config/lris2_vacuum.yaml + +peer_id: lris2_vacuum +group_id: lris2 + +# Local ZMQ transport for offline testing (no broker required). +# Switch to rabbitmq for deployment. +transport: zmq +bind: tcp://*:5610 +discovery_enabled: false + +hardware: + # Serial->TCP bridge endpoint for the VGC502 on lris2-137 (instrument LAN). + ip_address: 192.168.30.100 + tcp_port: 10001 + n_gauges: 2 # VGC502 has 2 channels + +logging: + level: INFO diff --git a/daemons/vacuum-gauge b/daemons/vacuum-gauge new file mode 100755 index 0000000..fe65894 --- /dev/null +++ b/daemons/vacuum-gauge @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +'''LRIS2 Inficon vacuum gauge daemon''' +import argparse +import sys +from typing import Any, Dict, Optional # pylint: disable=W0611 + +from lris2.daemon import Lris2Daemon +from lris2.driver.inficon.inficonvgc502 import InficonVGC502 + + +class VacuumGauge(Lris2Daemon): # pylint: disable=W0223 + '''Daemon for the LRIS2 Inficon VGC50x vacuum gauge controller''' + + # Defaults + group_id = "lris2" + + # pub/sub topics + topics = {} + + def __init__(self): + """Initialize the vacuum gauge daemon.""" + super().__init__() + + # Defaults + self.host = None + self.port = None + self.n_gauges = 1 + self.dev = InficonVGC502(log=True) + self.daemon_desc = None + + # Daemon state + self.state = { + 'error': '', + 'connected': False, + } + + def on_start(self, libby): + '''Start the daemon, register keywords, connect to the controller''' + self.host = self.get_config("hardware.ip_address") + self.port = self.get_config("hardware.tcp_port") + self.n_gauges = int(self.get_config("hardware.n_gauges", 1)) + self.daemon_desc = self.get_config("peer_id") + + if not (self.host and self.port): + self.logger.error("No IP address or port specified for vacuum gauge controller") + self.state['error'] = 'No IP address or port specified' + return + + # Keywords register regardless of hardware availability so the daemon + # is usable (and inspectable) even while the gauge is unreachable. + self._register_keywords() + + try: + connection = self.connect(True) + if not connection.get("is_connected"): + self.logger.warning("Gauge not reachable at %s:%s; daemon up, hardware offline", + self.host, self.port) + else: + self.initialize() + self.logger.info("Initialized %s", self.daemon_desc) + except (ConnectionRefusedError, OSError) as e: + self.logger.warning("Failed to connect to hardware: %s", e) + self.state['error'] = str(e) + + self.logger.info("Starting %s Daemon", self.daemon_desc) + + def _register_keywords(self): + """Register keywords for the daemon.""" + self.keyword_registry.bool("is_connected", + getter=self.dev.is_connected, + setter=self.keyword_wrapper(self.connect, key="is_connected"), + description="Check if daemon can talk to the vacuum gauge controller.") + for gauge in range(1, self.n_gauges + 1): + self.keyword_registry.float(f"pressure{gauge}", + getter=self.keyword_wrapper( + lambda g=gauge: self.get_pressure(g), key="pressure"), + units=self.dev.pressure_units or None, + description=f"Pressure reading from gauge {gauge}.") + self.keyword_registry.float("temperature", + getter=self.keyword_wrapper(self.get_temperature, key="temperature"), + units="C", + description="Controller temperature.") + self.keyword_registry.string("units", + getter=self.keyword_wrapper(self.get_units, key="units"), + setter=self.keyword_wrapper(self.set_units, key="units"), + validator=self._check_units, + description="Pressure units (mbar/Torr/Pascal/Micron/hPascal/Volt).") + self.keyword_registry.string("model", + getter=lambda: self.dev.model or "", + description="Controller model.") + self.keyword_registry.string("serial", + getter=lambda: str(self.dev.serial_number or ""), + description="Controller serial number.") + self.keyword_registry.string("firmware", + getter=lambda: self.dev.firmware_version or "", + description="Controller firmware version.") + + def initialize(self): + """Query controller identity, gauge count, and units.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + if not self.dev.initialize(): + return {"ok": False, "error": "Controller initialization failed"} + self.logger.debug("Initialized %s", self.daemon_desc) + except (OSError, ValueError) as e: + self.logger.error("error: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True} + + def on_stop(self, libby=None) -> None: + '''Stop the daemon and disconnect from the hardware device''' + try: + self.connect(False) + self.logger.info("Disconnected %s", self.daemon_desc) + except (OSError, ValueError) as e: + self.logger.error("Disconnect %s failed: %s", self.daemon_desc, e) + + def connect(self, connect): + """Connect to or disconnect from the controller.""" + try: + if connect: + self.dev.connect(host=self.host, port=self.port) + else: + self.dev.disconnect() + result = self.dev.is_connected() + self.state['connected'] = result + except (OSError, ValueError) as e: + self.logger.error("Failed to connect/disconnect with hardware: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "is_connected": result} + + def get_pressure(self, gauge: int): + """Read pressure from a gauge.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + pressure = float(self.dev.read_pressure(gauge=gauge)) + self.logger.debug("pressure%d: %s", gauge, pressure) + except (OSError, ValueError) as e: + self.logger.error("error: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "pressure": pressure} + + def get_temperature(self): + """Read controller temperature.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + temperature = float(self.dev.read_temperature()) + self.logger.debug("temperature: %s", temperature) + except (OSError, ValueError) as e: + self.logger.error("error: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "temperature": temperature} + + def get_units(self): + """Read the current pressure units.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + self.dev.get_pressure_unit() + units = self.dev.pressure_units + except (OSError, ValueError) as e: + self.logger.error("error: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "units": units} + + def set_units(self, units: str): + """Set the pressure units by name.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + code = InficonVGC502.UNIT_CODES.index(units) + if not self.dev.set_pressure_unit(code): + return {"ok": False, "error": f"Failed to set units to {units}"} + except (OSError, ValueError) as e: + self.logger.error("error: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "units": self.dev.pressure_units} + + def _check_units(self, units: str) -> Optional[str]: + """Validate a requested unit name against the controller's supported set.""" + if units not in InficonVGC502.UNIT_CODES: + return (f"unknown units '{units}'; " + f"available: {list(InficonVGC502.UNIT_CODES)}") + return None + + @staticmethod + def keyword_wrapper(func, key=None): + """Wrap a daemon method for use as a keyword getter/setter.""" + def wrapper(*args, **kwargs): + result = func(*args, **kwargs) + if not result.get("ok"): + raise RuntimeError(result.get("error", f"Unknown error in {func.__name__}")) + if key: + return result[key] + return {k: v for k, v in result.items() if k != "ok"} + return wrapper + + +def main(): + """Main entry point for the daemon.""" + parser = argparse.ArgumentParser(description='LRIS2 Inficon vacuum gauge daemon') + parser.add_argument('-c', '--config', type=str, + help='Path to config file (YAML)') + parser.add_argument('-d', '--daemon-id', type=str, + help='Daemon ID (required for subsystem configs with multiple daemons)') + + args = parser.parse_args() + + if not args.config: + print("--config is required", file=sys.stderr) + sys.exit(2) + + try: + daemon = VacuumGauge.from_config_file(args.config, daemon_id=args.daemon_id) + daemon.serve() + except KeyboardInterrupt: + print("\nDaemon interrupted by user") + sys.exit(0) + except Exception as e: # pylint: disable=W0718 + print(f"Error running daemon: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == '__main__': + main() From 862475fe84727ed26e5ed151ab003da73997dc45 Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:53:34 -0700 Subject: [PATCH 4/6] Add Sunpower cryocooler daemon --- config/lris2_cryocooler.yaml | 34 ++++ daemons/cryocooler | 290 +++++++++++++++++++++++++++++++++++ 2 files changed, 324 insertions(+) create mode 100644 config/lris2_cryocooler.yaml create mode 100755 daemons/cryocooler diff --git a/config/lris2_cryocooler.yaml b/config/lris2_cryocooler.yaml new file mode 100644 index 0000000..fee8e72 --- /dev/null +++ b/config/lris2_cryocooler.yaml @@ -0,0 +1,34 @@ +# +# Usage: +# daemons/cryocooler -c config/lris2_cryocooler.yaml + +peer_id: lris2_cryocooler +group_id: lris2 + +# Local ZMQ transport for offline testing (no broker required). +# Switch to rabbitmq for deployment. +transport: zmq +bind: tcp://*:5611 +discovery_enabled: false + +hardware: + # TODO: set to the real Sunpower cryocooler endpoint. + # Placeholder points at a closed local port so the daemon exercises its + # graceful-offline path (connection refused) during offline testing. + ip_address: 127.0.0.1 + tcp_port: 9 + con_type: tcp + units: K + +# Software limits enforced by the daemon on writes (the controller exposes no +# hard limits). TODO: set to the real safe operating envelope. +limits: + temp: + soft_min: 40.0 + soft_max: 330.0 + power: + soft_min: 0.0 + soft_max: 250.0 + +logging: + level: INFO diff --git a/daemons/cryocooler b/daemons/cryocooler new file mode 100755 index 0000000..164e791 --- /dev/null +++ b/daemons/cryocooler @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +'''LRIS2 Sunpower cryocooler daemon''' +import argparse +import sys +from typing import Any, Callable, Optional # pylint: disable=W0611 + +from lris2.daemon import Lris2Daemon +from lris2.driver.sunpower import SunpowerCryocooler + + +class Cryocooler(Lris2Daemon): # pylint: disable=W0223 + '''Daemon for the LRIS2 Sunpower cryocooler''' + + # Defaults + group_id = "lris2" + + # pub/sub topics + topics = {} + + def __init__(self): + """Initialize the cryocooler daemon.""" + super().__init__() + + # Defaults + self.host = None + self.port = None + self.con_type = "tcp" + self.units = "K" + self.dev = SunpowerCryocooler(log=True) + self.daemon_desc = None + self._temp_soft_min = None + self._temp_soft_max = None + self._power_soft_min = None + self._power_soft_max = None + + # Daemon state. The Sunpower controller has no query for control mode or + # cooler on/off, so both are tracked here from the commands we issue. + self.state = { + 'error': '', + 'connected': False, + 'cooler_on': False, + 'ctrlmode': 'temp', + } + + def on_start(self, libby): + '''Start the daemon, register keywords, connect to the controller''' + self.host = self.get_config("hardware.ip_address") + self.port = self.get_config("hardware.tcp_port") + self.con_type = self.get_config("hardware.con_type", "tcp") + self.units = self.get_config("hardware.units", "K") + self.daemon_desc = self.get_config("peer_id") + self._temp_soft_min = self.get_config("limits.temp.soft_min") + self._temp_soft_max = self.get_config("limits.temp.soft_max") + self._power_soft_min = self.get_config("limits.power.soft_min") + self._power_soft_max = self.get_config("limits.power.soft_max") + + if not (self.host and self.port): + self.logger.error("No IP address or port specified for cryocooler") + self.state['error'] = 'No IP address or port specified' + return + + self._register_keywords() + + try: + connection = self.connect(True) + if not connection.get("is_connected"): + self.logger.warning("Cryocooler not reachable at %s:%s; daemon up, " + "hardware offline", self.host, self.port) + except (ConnectionRefusedError, OSError) as e: + self.logger.warning("Failed to connect to hardware: %s", e) + self.state['error'] = str(e) + + self.logger.info("Starting %s Daemon", self.daemon_desc) + + def _register_keywords(self): + """Register keywords for the daemon.""" + self.keyword_registry.bool("is_connected", + getter=self.dev.is_connected, + setter=self.keyword_wrapper(self.connect, key="is_connected"), + description="Check if daemon can talk to the cryocooler.") + self.keyword_registry.float("targettemp", + getter=self._reader(self.dev.get_target_temp, "targettemp"), + setter=self.keyword_wrapper(self.set_targettemp, key="targettemp"), + validator=self._check_temp_limits, + units=self.units, + description="Cold-head temperature setpoint.") + self.keyword_registry.float("coldtemp", + getter=self._reader(self.dev.get_cold_head_temp, "coldtemp"), + units=self.units, + description="Cold-head temperature.") + self.keyword_registry.float("rejecttemp", + getter=self._reader(self.dev.get_reject_temp, "rejecttemp"), + units=self.units, + description="Reject (warm end) temperature.") + self.keyword_registry.float("measpower", + getter=self._reader(self.dev.get_measured_power, "measpower"), + units="W", + description="Measured cooler power.") + self.keyword_registry.float("cmdpower", + getter=self._reader(self.dev.get_commanded_power, "cmdpower"), + setter=self.keyword_wrapper(self.set_cmdpower, key="cmdpower"), + validator=self._check_power_limits, + units="W", + description="Commanded (steady) cooler power.") + self.keyword_registry.float("curcmdpower", + getter=self._reader(self.dev.get_current_commanded_power, "curcmdpower"), + units="W", + description="Current commanded power.") + self.keyword_registry.bool("cooler", + getter=lambda: self.state['cooler_on'], + setter=self.keyword_wrapper(self.set_cooler, key="cooler"), + description="Cooler on/off (commanded state).") + self.keyword_registry.string("ctrlmode", + getter=lambda: self.state['ctrlmode'], + setter=self.keyword_wrapper(self.set_ctrlmode, key="ctrlmode"), + validator=self._check_ctrlmode, + description="Control mode: 'temp' (setpoint) or 'power' (fixed power).") + self.keyword_registry.float("tempsoftmin", + getter=lambda: self._temp_soft_min, + setter=lambda v: setattr(self, "_temp_soft_min", float(v)), + units=self.units, + description="Software lower limit for the temperature setpoint.") + self.keyword_registry.float("tempsoftmax", + getter=lambda: self._temp_soft_max, + setter=lambda v: setattr(self, "_temp_soft_max", float(v)), + units=self.units, + description="Software upper limit for the temperature setpoint.") + self.keyword_registry.float("powersoftmin", + getter=lambda: self._power_soft_min, + setter=lambda v: setattr(self, "_power_soft_min", float(v)), + units="W", + description="Software lower limit for commanded power.") + self.keyword_registry.float("powersoftmax", + getter=lambda: self._power_soft_max, + setter=lambda v: setattr(self, "_power_soft_max", float(v)), + units="W", + description="Software upper limit for commanded power.") + self.keyword_registry.string("status", + getter=self._reader(lambda: " | ".join(self.dev.get_status()), "status"), + description="Raw controller status block.") + self.keyword_registry.string("error", + getter=self._reader(lambda: str(self.dev.get_error()), "error"), + description="Last controller error.") + self.keyword_registry.string("version", + getter=self._reader(lambda: str(self.dev.get_version()), "version"), + description="Controller firmware version.") + + def on_stop(self, libby=None) -> None: + '''Stop the daemon and disconnect from the hardware device''' + try: + self.connect(False) + self.logger.info("Disconnected %s", self.daemon_desc) + except (OSError, ValueError) as e: + self.logger.error("Disconnect %s failed: %s", self.daemon_desc, e) + + def connect(self, connect): + """Connect to or disconnect from the controller.""" + try: + if connect: + self.dev.connect(self.host, self.port, con_type=self.con_type) + else: + self.dev.disconnect() + result = self.dev.is_connected() + self.state['connected'] = result + except (OSError, ValueError) as e: + self.logger.error("Failed to connect/disconnect with hardware: %s", e) + return {"ok": False, "error": str(e)} + return {"ok": True, "is_connected": result} + + def set_targettemp(self, temp: float): + """Set the temperature setpoint and switch to temperature-control mode.""" + result = self._write(lambda: self.dev.set_target_temp(float(temp)), "targettemp", temp) + if result.get("ok"): + self.state['ctrlmode'] = 'temp' + return result + + def set_cmdpower(self, watts: float): + """Set the commanded power and switch to fixed-power mode.""" + result = self._write(lambda: self.dev.set_commanded_power(float(watts)), "cmdpower", watts) + if result.get("ok"): + self.state['ctrlmode'] = 'power' + return result + + def set_cooler(self, on: bool): + """Turn the cooler on or off.""" + action = self.dev.turn_on_cooler if on else self.dev.turn_off_cooler + result = self._write(action, "cooler", bool(on)) + if result.get("ok"): + self.state['cooler_on'] = bool(on) + return result + + def set_ctrlmode(self, mode: str): + """Record the intended control mode. + + The controller has no explicit mode command; the effective mode is + whichever setpoint (targettemp or cmdpower) was last written. This + keyword tracks that intent and is updated by those setters. + """ + self.state['ctrlmode'] = mode + return {"ok": True, "ctrlmode": mode} + + def _reader(self, fn: Callable[[], Any], key: str): + """Build a keyword getter that reads one value from the device.""" + def read(): + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + value = fn() + except (OSError, ValueError, IndexError) as e: + self.logger.error("error reading %s: %s", key, e) + return {"ok": False, "error": str(e)} + return {"ok": True, key: value} + return self.keyword_wrapper(read, key=key) + + def _write(self, fn: Callable[[], Any], key: str, value: Any): + """Run a device write, guarding on connection and mapping errors.""" + if not self.dev.is_connected(): + return {"ok": False, "error": "Not connected to hardware"} + try: + fn() + except (OSError, ValueError) as e: + self.logger.error("error writing %s: %s", key, e) + return {"ok": False, "error": str(e)} + return {"ok": True, key: value} + + def _check_temp_limits(self, value: float) -> Optional[str]: + """Validate a temperature setpoint against the software limits.""" + return self._check_limits(value, self._temp_soft_min, self._temp_soft_max, "temperature") + + def _check_power_limits(self, value: float) -> Optional[str]: + """Validate a commanded power against the software limits.""" + return self._check_limits(value, self._power_soft_min, self._power_soft_max, "power") + + @staticmethod + def _check_limits(value: float, low: Optional[float], high: Optional[float], + label: str) -> Optional[str]: + """Return an error string if value is outside [low, high], else None.""" + if low is not None and value < low: + return f"{label} {value} below soft min {low}" + if high is not None and value > high: + return f"{label} {value} above soft max {high}" + return None + + @staticmethod + def _check_ctrlmode(mode: str) -> Optional[str]: + """Validate a control-mode value.""" + if mode not in ("temp", "power"): + return f"ctrlmode must be 'temp' or 'power', got '{mode}'" + return None + + @staticmethod + def keyword_wrapper(func, key=None): + """Wrap a daemon method for use as a keyword getter/setter.""" + def wrapper(*args, **kwargs): + result = func(*args, **kwargs) + if not result.get("ok"): + raise RuntimeError(result.get("error", f"Unknown error in {func.__name__}")) + if key: + return result[key] + return {k: v for k, v in result.items() if k != "ok"} + return wrapper + + +def main(): + """Main entry point for the daemon.""" + parser = argparse.ArgumentParser(description='LRIS2 Sunpower cryocooler daemon') + parser.add_argument('-c', '--config', type=str, + help='Path to config file (YAML)') + parser.add_argument('-d', '--daemon-id', type=str, + help='Daemon ID (required for subsystem configs with multiple daemons)') + + args = parser.parse_args() + + if not args.config: + print("--config is required", file=sys.stderr) + sys.exit(2) + + try: + daemon = Cryocooler.from_config_file(args.config, daemon_id=args.daemon_id) + daemon.serve() + except KeyboardInterrupt: + print("\nDaemon interrupted by user") + sys.exit(0) + except Exception as e: # pylint: disable=W0718 + print(f"Error running daemon: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == '__main__': + main() From d484dea9a4fc38c8ccad888ce8f94f64ee69943c Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:12:45 -0700 Subject: [PATCH 5/6] replaced lris2 daemon template with a thin version as libby daemon has been upgraded --- src/daemon.py | 326 ++------------------------------------------------ 1 file changed, 7 insertions(+), 319 deletions(-) diff --git a/src/daemon.py b/src/daemon.py index 1627a5d..f530b3e 100644 --- a/src/daemon.py +++ b/src/daemon.py @@ -1,324 +1,12 @@ """ LRIS2 daemon base class """ -from __future__ import annotations # for Python 3.9 compatibility -import json -import logging -from dataclasses import is_dataclass, asdict -import collections.abc as cabc -import signal, sys, threading, time -from typing import Any, Callable, Dict, Iterable, List, Optional, Type +from libby.daemon import LibbyDaemon -from libby import Keyword, Libby -from . import config as cfg -Payload = Dict[str, Any] -RPCHandler = Callable[[Payload], Dict[str, Any]] -EvtHandler = Callable[[Payload], None] +class Lris2Daemon(LibbyDaemon): + """Instantiates the LRIS2 daemon base using LibbyDaemon. -class Lris2Daemon: + Transport is rabbitmq and discovery is false since rabbitmq includes + its own discovery. """ - Base daemon class for Libby peers with support for multiple transports. - - ZMQ Usage: - class MyPeer(Lris2Daemon): - peer_id = "my-peer" - bind = "tcp://*:5555" - address_book = {"other-peer": "tcp://localhost:5556"} - - services = {"echo": lambda payload: {"echo": payload}} - topics = {"alerts": lambda payload: print(payload)} - - RabbitMQ Usage: - class MyPeer(Lris2Daemon): - transport = "rabbitmq" - peer_id = "my-peer" - rabbitmq_url = "amqp://localhost" # optional, defaults to this - - services = {"echo": lambda payload: {"echo": payload}} - topics = {"alerts": lambda payload: print(payload)} - - Note: RabbitMQ doesn't need bind or address_book since routing is - handled automatically by the broker. - """ - # simple attributes users set - peer_id: Optional[str] = None - bind: Optional[str] = None - address_book: Optional[Dict[str, str]] = None - discovery_enabled: bool = False - discovery_interval_s: float = 5.0 - - # transport selection: "zmq" or "rabbitmq (default)" - transport: str = "rabbitmq" - rabbitmq_url: Optional[str] = None - group_id: Optional[str] = None - # internal config - _config: Dict[str, Any] = {} - - # payload-only handlers - services: Dict[str, RPCHandler] = {} - topics: Dict[str, EvtHandler] = {} - - def __init__(self) -> None: - # Ensure per-instance handler tables - if type(self).services is self.services: - self.services = {} - if type(self).topics is self.topics: - self.topics = {} - - # Config ingestion - @classmethod - def from_config_file( - cls: Type["Lris2Daemon"], - path: str, - daemon_id: Optional[str] = None, - ) -> "Lris2Daemon": - """ - Build a daemon from a YAML config file. - - Args: - path: Path to yaml config file - daemon_id: For subsystem configs, which daemon to instantiate. - If None and config has multiple daemons, raises error. - - Returns: - Configured daemon instance - """ - loader = cfg.DaemonConfigLoader(path) - - # For subsystem configs, daemon_id is required unless only one daemon - if loader.is_subsystem: - if daemon_id is None: - if len(loader.daemon_ids) == 1: - daemon_id = loader.daemon_ids[0] - else: - raise cfg.ConfigError( - f"Subsystem config has multiple daemons: {loader.daemon_ids}. " - f"Specify daemon_id parameter." - ) - - config_dict = loader.get_daemon_config(daemon_id) - return cls.from_config(config_dict) - - @classmethod - def from_config(cls: Type["Lris2Daemon"], config: Dict[str, Any]) -> "Lris2Daemon": - """ - Build a daemon from a configuration dictionary. - - Known daemon attributes are mapped directly to instance attributes. - - Args: - config: Configuration dictionary - - Returns: - Configured daemon instance - """ - instance = cls() - - # Map known daemon attributes directly - for attr in cfg.DAEMON_ATTRS: - if attr in config: - setattr(instance, attr, config[attr]) - - # Store full config for subclass access - instance._config = config - - # Setup logging - instance._setup_logging() - - return instance - - def _setup_logging(self) -> None: - """Configure logging from config or defaults.""" - log_config = self._config.get("logging", {}) - level_str = log_config.get("level", "INFO").upper() - level = getattr(logging, level_str, logging.INFO) - log_file = log_config.get("file") - - logging.basicConfig( - level=level, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - filename=log_file, - ) - - self.logger = logging.getLogger(self.peer_id or self.__class__.__name__) - - def get_config(self, key: str, default: Any = None) -> Any: - """ - Get a value from the daemon's configuration. - - Args: - key: Config key (supports dot notation for nested keys, e.g., "hardware.ip_address") - default: Default value if key not found - - Returns: - Config value or default - """ - keys = key.split(".") - value = self._config - for k in keys: - if isinstance(value, dict) and k in value: - value = value[k] - else: - return default - return value - - - # optional hooks - def on_start(self, libby: Libby) -> None: ... - def on_stop(self, libby: Optional[Libby] = None) -> None: ... - def on_hello(self, libby: Libby) -> None: ... - def on_event(self, topic: str, msg) -> None: - print(f"[{self.__class__.__name__}] {topic}: {msg.env.payload}") - - # config getters - def config_peer_id(self) -> str: return self.peer_id or self._must("peer_id") - def config_bind(self) -> str: return self.bind or self._must("bind") - def config_rabbitmq_url(self) -> str: return self.rabbitmq_url or "amqp://localhost" - def config_group_id(self) -> Optional[str]: return self.group_id - def config_address_book(self) -> Dict[str, str]: return self.address_book if self.address_book is not None else {} - def config_discovery_enabled(self) -> bool: return bool(self.discovery_enabled) - def config_discovery_interval_s(self) -> float: return float(self.discovery_interval_s) - def config_rpc_keys(self) -> List[str]: return list(self.services.keys()) - def config_subscriptions(self) -> List[str]: return list(self.topics.keys()) - - # user-facing helpers - def add_service(self, key: str, fn: RPCHandler) -> None: - self.services[key] = fn - if hasattr(self, "libby"): self._register_services({key: fn}) - - def add_services(self, mapping: Dict[str, RPCHandler]) -> None: - self.services.update(mapping) - if hasattr(self, "libby"): self._register_services(mapping) - - def register_keyword(self, keyword: Keyword) -> None: - """Register a Keyword; delegates to the underlying Libby instance.""" - self.libby.register_keyword(keyword) - - def register_keywords(self, keywords: Iterable[Keyword]) -> None: - """Register many keywords in one libby call; call from on_start or later.""" - self.libby.register_keywords(keywords) - - @property - def keyword_registry(self): - """KeywordRegistry on the underlying Libby instance.""" - return self.libby.keyword_registry - - def add_topic(self, topic: str, fn: EvtHandler) -> None: - self.topics[topic] = fn - if hasattr(self, "libby"): - self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) - self.libby.subscribe(topic) - - def add_topics(self, mapping: Dict[str, EvtHandler]) -> None: - self.topics.update(mapping) - if hasattr(self, "libby"): - for topic, fn in mapping.items(): - self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) - self.libby.subscribe(*mapping.keys()) - - # internals - def _must(self, name: str): - raise NotImplementedError(f"Set `{name}` or override config_{name}()") - - def _service_adapter(self, fn): - def adapter(user_payload: dict, _ctx: dict) -> dict: - try: - result = fn(user_payload) # user returns ANYTHING - return self.payload(result) # we "shove it into payload" for them - except Exception as ex: - return {"ok": False, "error": str(ex)} - return adapter - - def _register_services(self, mapping: Dict[str, RPCHandler]) -> None: - for key, fn in mapping.items(): - self.libby.serve_keys([key], self._service_adapter(fn)) - - def build_libby(self) -> Libby: - """Build Libby instance with selected transport.""" - if self.transport == "rabbitmq": - return Libby.rabbitmq( - self_id=self.config_peer_id(), - rabbitmq_url=self.config_rabbitmq_url(), - keys=[], - callback=None, - group_id=self.config_group_id(), - ) - else: - # Default to ZMQ - return Libby.zmq( - self_id=self.config_peer_id(), - bind=self.config_bind(), - address_book=self.config_address_book(), - keys=[], callback=None, # register per-key - discover=self.config_discovery_enabled(), - discover_interval_s=self.config_discovery_interval_s(), - hello_on_start=True, - group_id=self.config_group_id(), - ) - - def serve(self) -> None: - stop_evt = threading.Event() - def _sig(_s, _f): stop_evt.set() - signal.signal(signal.SIGINT, _sig) - signal.signal(signal.SIGTERM, _sig) - - try: - self.libby = self.build_libby() - except Exception as ex: - print(f"[{self.__class__.__name__}] failed to start: {ex}", file=sys.stderr) - raise - - if self.services: - self._register_services(self.services) - if self.topics: - for topic, fn in self.topics.items(): - self.libby.listen(topic, lambda msg, _h=fn: _h(msg.env.payload)) - self.libby.subscribe(*self.topics.keys()) - - # discovery hello + hooks - try: - if self.config_discovery_enabled(): - self.libby.hello() - self.on_hello(self.libby) - except Exception: - pass - - try: - self.on_start(self.libby) - except Exception as ex: - print(f"[{self.__class__.__name__}] on_start error: {ex}", file=sys.stderr) - - keywords_to_register = self.libby.keyword_registry.drain() - if keywords_to_register: - self.libby.register_keywords(keywords_to_register) - - if self.transport == "rabbitmq": - print(f"[{self.__class__.__name__}] up: id={self.config_peer_id()} transport=rabbitmq url={self.rabbitmq_url}") - else: - print(f"[{self.__class__.__name__}] up: id={self.config_peer_id()} bind={self.config_bind()}") - try: - while not stop_evt.is_set(): time.sleep(0.5) - finally: - try: self.on_stop() - except Exception: pass - self.libby.stop() - print(f"[{self.__class__.__name__}] stopped") - - def payload(self, value=None, /, **extra) -> dict: - if value is None: - out = {} - elif is_dataclass(value): - out = asdict(value) - elif isinstance(value, cabc.Mapping): - out = dict(value) - else: - out = {"data": value} - - if extra: - out.update(extra) - - try: - json.dumps(out) - except TypeError as e: - raise ValueError(f"Payload not JSON-serializable: {e}") from e - - return out \ No newline at end of file + transport = "rabbitmq" + discovery_enabled = False \ No newline at end of file From 3d6f3297105e923784164efeaccaa35da3ec2951 Mon Sep 17 00:00:00 2001 From: Mike Langmayr <1809691+mikelangmayr@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:18:02 -0700 Subject: [PATCH 6/6] Remove config.py which lives in Libby now --- daemons/vacuum-gauge | 2 +- src/config.py | 202 ------------------------------------------- 2 files changed, 1 insertion(+), 203 deletions(-) delete mode 100644 src/config.py diff --git a/daemons/vacuum-gauge b/daemons/vacuum-gauge index fe65894..6e2b685 100755 --- a/daemons/vacuum-gauge +++ b/daemons/vacuum-gauge @@ -2,7 +2,7 @@ '''LRIS2 Inficon vacuum gauge daemon''' import argparse import sys -from typing import Any, Dict, Optional # pylint: disable=W0611 +from typing import Optional # pylint: disable=W0611 from lris2.daemon import Lris2Daemon from lris2.driver.inficon.inficonvgc502 import InficonVGC502 diff --git a/src/config.py b/src/config.py deleted file mode 100644 index efb87f1..0000000 --- a/src/config.py +++ /dev/null @@ -1,202 +0,0 @@ -""" -Configuration ingestion daemons. -""" - -from __future__ import annotations # for Python 3.9 compatibility -from pathlib import Path -from typing import Any, Dict, List, Optional -import yaml - - -class ConfigError(Exception): - """Raised when configuration loading or validation fails.""" - pass - - -# Known HispecDaemon attributes that map directly from config -DAEMON_ATTRS = frozenset({ - "peer_id", - "bind", - "address_book", - "discovery_enabled", - "discovery_interval_s", - "transport", - "rabbitmq_url", - "group_id", -}) - - -def load_file(path: str | Path) -> Dict[str, Any]: - """ - Load a configuration file (YAML). - - Args: - path: Path to the config file - - Returns: - Parsed configuration dictionary - - Raises: - ConfigError: If the file cannot be loaded or parsed - """ - path = Path(path) - - if not path.exists(): - raise ConfigError(f"Config file not found: {path}") - - try: - with open(path, "r") as f: - return yaml.safe_load(f) or {} - except Exception as e: - raise ConfigError(f"Failed to load config from {path}: {e}") - - -def extract_daemon_config( - full_config: Dict[str, Any], - daemon_id: str, -) -> Dict[str, Any]: - """ - Extract configuration for a specific daemon from a subsystem config. - - Merges subsystem-level defaults with daemon-specific overrides. - - Args: - full_config: Full subsystem configuration - daemon_id: Identifier of the daemon to extract - - Returns: - Merged configuration for the specific daemon - - Raises: - ConfigError: If the daemon is not found in the config - """ - daemons = full_config.get("daemons", {}) - - if daemon_id not in daemons: - available = list(daemons.keys()) if daemons else [] - raise ConfigError( - f"Daemon '{daemon_id}' not found in config. " - f"Available daemons: {available}" - ) - - # Start with subsystem-level defaults (excluding 'daemons' key) - result = {k: v for k, v in full_config.items() if k != "daemons"} - - # Merge daemon-specific config (overrides subsystem defaults) - daemon_config = daemons[daemon_id] - if daemon_config: - _deep_merge(result, daemon_config) - - # Ensure peer_id is set - if "peer_id" not in result: - result["peer_id"] = daemon_id - - return result - - -def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> None: - """Deep merge override into base (modifies base in place).""" - for key, value in override.items(): - if ( - key in base - and isinstance(base[key], dict) - and isinstance(value, dict) - ): - _deep_merge(base[key], value) - else: - base[key] = value - - -def list_daemons(config: Dict[str, Any]) -> List[str]: - """ - List all daemon IDs defined in a subsystem config. - - Args: - config: Subsystem configuration dictionary - - Returns: - List of daemon identifiers - """ - return list(config.get("daemons", {}).keys()) - - -def is_subsystem_config(config: Dict[str, Any]) -> bool: - """Check if a config dict is a subsystem config (has 'daemons' key).""" - return "daemons" in config - - -class DaemonConfigLoader: - """ - Helper class for loading daemon configurations. - - Usage: - loader = DaemonConfigLoader("config/hsfei.yaml") - - # For subsystem configs with multiple daemons - for daemon_id in loader.daemon_ids: - config = loader.get_daemon_config(daemon_id) - daemon = MyDaemon.from_config(config) - - # Or load a single daemon directly - config = loader.get_daemon_config("pickoff1", env_prefix="HISPEC_") - """ - - def __init__(self, path: str | Path): - """ - Initialize loader with a config file path. - - Args: - path: Path to the config file (YAML or JSON) - """ - self.path = Path(path) - self._config: Optional[Dict[str, Any]] = None - - @property - def config(self) -> Dict[str, Any]: - """Lazily load and return the raw configuration.""" - if self._config is None: - self._config = load_file(self.path) - return self._config - - @property - def is_subsystem(self) -> bool: - """Check if this is a subsystem config with multiple daemons.""" - return is_subsystem_config(self.config) - - @property - def subsystem(self) -> Optional[str]: - """Get the subsystem name, if defined.""" - return self.config.get("subsystem") - - @property - def daemon_ids(self) -> List[str]: - """List all daemon IDs in this config.""" - if self.is_subsystem: - return list_daemons(self.config) - # Single daemon config - use peer_id or filename - peer_id = self.config.get("peer_id", self.path.stem) - return [peer_id] - - def get_daemon_config( - self, - daemon_id: Optional[str] = None, - ) -> Dict[str, Any]: - """ - Get configuration for a specific daemon. - - Args: - daemon_id: Daemon identifier (required for subsystem configs, - optional for single-daemon configs) - - Returns: - Merged configuration dictionary - """ - if self.is_subsystem: - if daemon_id is None: - raise ConfigError( - "daemon_id required for subsystem configs. " - f"Available: {self.daemon_ids}" - ) - return extract_daemon_config(self.config, daemon_id) - else: - return self.config.copy()