Skip to content

Update hispec daemon base to use LibbyDaemon #152

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
prkrtg wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Update hispec daemon base to use LibbyDaemon #152

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
326 changes: 6 additions & 320 deletions src/hispec/daemon.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,324 +1,10 @@
""" Hispec Daemon template """
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 HispecDaemon:
"""
Base daemon class for Libby peers with support for multiple transports.

ZMQ Usage:
class MyPeer(HispecDaemon):
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(HispecDaemon):
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.
class HispecDaemon(LibbyDaemon):
"""Instantiates the HispecDaemon base using LibbyDaemon.
Transport is with rabbitmq and discovery is false since rabbitmq includes it's own discovery.
"""
# 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["HispecDaemon"],
path: str,
daemon_id: Optional[str] = None,
) -> "HispecDaemon":
"""
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["HispecDaemon"], config: Dict[str, Any]) -> "HispecDaemon":
"""
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
transport = "rabbitmq"
discovery_enabled = False
Loading