API Reference

Generated from docstrings in the libby package.

Core

class libby.libby.BroadcastReply(peer_id, payload)

Bases: object

One peer’s answer to Libby.broadcast_request().

Parameters:

• peer_id (str)

• payload (Dict[str, Any])

peer_id: str

payload: Dict[str, Any]

class libby.libby.Libby(self_id, transport, keys=None, callback=None, *, discover=False, discover_interval_s=5.0, hello_on_start=True, is_daemon=False)

Bases: object

Parameters:

• self_id (str)

• keys (Optional[List[str]])

• callback (Optional[Callable[[dict, dict], Optional[dict]]])

• discover (bool)

• discover_interval_s (float)

• hello_on_start (bool)

• is_daemon (bool)

classmethod zmq(self_id, bind, address_book, keys=None, callback=None, *, discover=False, discover_interval_s=5.0, hello_on_start=True, group_id=None, is_daemon=False)

Parameters:

• self_id (str)

• bind (str)

• address_book (Dict[str, str])

• keys (List[str] | None)

• callback (Callable[[dict, dict], dict | None] | None)

• discover (bool)

• discover_interval_s (float)

• hello_on_start (bool)

• group_id (str | None)

• is_daemon (bool)

Return type:

Libby

classmethod rabbitmq(self_id, rabbitmq_url='amqp://localhost', keys=None, callback=None, group_id=None, *, is_daemon=False)

Create a Libby instance using RabbitMQ transport.

Parameters:

• self_id (str) – Unique identifier for this peer

• rabbitmq_url (str) – RabbitMQ connection URL (default: “amqp://localhost”)

• keys (List[str] | None) – List of RPC keys this peer will serve

• callback (Callable[[dict, dict], dict | None] | None) – Default callback for RPC requests

• group_id (str | None) – Optional group identifier

• is_daemon (bool)

Returns:

Configured Libby instance

Return type:

Libby

Example

>>> libby = Libby.rabbitmq(
...     self_id="peer-A",
...     rabbitmq_url="amqp://user:pass@localhost:5672/",
...     keys=["echo"],
...     group_id="hsfei"
... )

start()

Return type:

None

stop()

Return type:

None

close()

Return type:

None

request(peer_id, key, payload, ttl_ms=8000)

Parameters:

• peer_id (str)

• key (str)

• payload (Dict[str, Any])

• ttl_ms (int)

rpc(peer_id, key, payload, ttl_ms=8000)

Parameters:

• peer_id (str)

• key (str)

• payload (Dict[str, Any])

• ttl_ms (int)

broadcast_request(key, payload, timeout_s=1.0)

Send key to every reachable peer and collect their replies.

Nothing on the wire says how many peers will answer, so this always waits the full timeout_s. Over ZMQ “every reachable peer” is the address book. Our own reply is dropped: on RabbitMQ our queue sits on the fanout exchange like everyone else’s.

Parameters:

• key (str)

• payload (Dict[str, Any])

• timeout_s (float)

Return type:

List[BroadcastReply]

serve_keys(keys, callback)

Parameters:

• keys (List[str])

• callback (Callable[[dict, dict], dict | None])

Return type:

None

register_keyword(keyword)

Parameters:

keyword (Keyword)

Return type:

None

register_keywords(keywords)

Parameters:

keywords (Iterable[Keyword])

Return type:

None

listen(topic, handler)

Parameters:

• topic (str)

• handler (Callable[[Any], None])

Return type:

None

listen_many(handlers)

Parameters:

handlers (Dict[str, Callable[[Any], None]])

Return type:

None

publish(topic, payload)

Parameters:

• topic (str)

• payload (Dict[str, Any])

Return type:

int

emit(topic, payload)

Parameters:

• topic (str)

• payload (Dict[str, Any])

Return type:

int

subscribe(*topics)

Parameters:

topics (str)

Return type:

None

unsubscribe(*topics)

Parameters:

topics (str)

Return type:

None

hello()

Return type:

None

peers_alive(within_s=30)

Parameters:

within_s (int)

Return type:

Dict[str, float]

knows_key(peer_id, key)

Parameters:

• peer_id (str)

• key (str)

Return type:

bool

wait_for_key(peer_id, key, timeout_s=3.0, poll_s=0.05)

Parameters:

• peer_id (str)

• key (str)

• timeout_s (float)

• poll_s (float)

Return type:

bool

wait_for_peer(peer_id, timeout_s=3.0, poll_s=0.05)

Parameters:

• peer_id (str)

• timeout_s (float)

• poll_s (float)

Return type:

bool

learn_peer_keys(peer_id, keys)

Parameters:

• peer_id (str)

• keys (List[str])

Return type:

None

run_forever()

Return type:

None

class libby.daemon.LibbyDaemon

Bases: object

Base class for configurable Libby daemons.

The class owns the following:

• Libby transport construction

• service and topic registration

• typed keyword registration

• YAML/JSON configuration loading

• subsystem configuration selection

• logging and lifecycle management

Instrument-specific subclasses should normally only implement on_start and on_stop, plus any hardware-facing methods.

CONFIG_ATTRIBUTES = frozenset({'address_book', 'bind', 'discovery_enabled', 'discovery_interval_s', 'fail_fast_on_start', 'group_id', 'peer_id', 'rabbitmq_url', 'transport'})

peer_id: str | None = None

bind: str | None = None

address_book: Dict[str, str] | None = None

discovery_enabled: bool = True

discovery_interval_s: float = 5.0

transport: str = 'zmq'

rabbitmq_url: str | None = None

group_id: str | None = None

fail_fast_on_start: bool = True

services: Dict[str, Callable[[Dict[str, Any]], Any]] = {}

topics: Dict[str, Callable[[Dict[str, Any]], None]] = {}

classmethod config_attributes()

Return config keys mapped directly onto daemon attributes.

Subclasses may extend the set:

CONFIG_ATTRIBUTES = (
    LibbyDaemon.CONFIG_ATTRIBUTES | {"device_name"}
)

Return type:

frozenset[str]

classmethod from_config(config)

Build a daemon from a configuration mapping.

Parameters:

config (Mapping[str, Any])

Return type:

DaemonT

classmethod from_config_file(path, daemon_id=None, *, env_prefix=None)

Build a daemon from a single-daemon or subsystem config file.

For a subsystem config containing one daemon, the daemon is selected automatically. For multiple daemons, daemon_id is required.

Parameters:

• path (str)

• daemon_id (str | None)

• env_prefix (str | None)

Return type:

DaemonT

get_config(key, default=None)

Read a configuration value using optional dot notation.

Parameters:

• key (str)

• default (Any)

Return type:

Any

on_start(libby)

Initialize hardware and register keywords.

Parameters:

libby (Libby)

Return type:

None

on_stop(libby=None)

Release hardware resources.

Parameters:

libby (Libby | None)

Return type:

None

on_hello(libby)

Run after the initial discovery hello.

Parameters:

libby (Libby)

Return type:

None

on_event(topic, msg)

Parameters:

• topic (str)

• msg (Any)

Return type:

None

config_peer_id()

Return type:

str

config_bind()

Return type:

str

config_address_book()

Return type:

Dict[str, str]

config_rabbitmq_url()

Return type:

str

config_group_id()

Return type:

str | None

config_discovery_enabled()

Return type:

bool

config_discovery_interval_s()

Return type:

float

config_rpc_keys()

Return type:

List[str]

config_subscriptions()

Return type:

List[str]

add_service(key, fn)

Parameters:

• key (str)

• fn (Callable[[Dict[str, Any]], Any])

Return type:

None

add_services(mapping)

Parameters:

mapping (Mapping[str, Callable[[Dict[str, Any]], Any]])

Return type:

None

add_topic(topic, fn)

Parameters:

• topic (str)

• fn (Callable[[Dict[str, Any]], None])

Return type:

None

add_topics(mapping)

Parameters:

mapping (Mapping[str, Callable[[Dict[str, Any]], None]])

Return type:

None

register_keyword(keyword)

Register now, or defer registration until the daemon starts.

Parameters:

keyword (Keyword)

Return type:

None

register_keywords(keywords)

Register many keywords now, or defer until startup.

Parameters:

keywords (Iterable[Keyword])

Return type:

None

property keyword_registry

Return Libby’s typed keyword builder.

This property is intended for use in on_start, after the Libby instance has been constructed.

build_libby()

Build a Libby instance for the configured transport.

Return type:

Libby

start()

Start Libby and initialize the daemon without blocking.

Return type:

None

request_stop()

Request termination of a blocking serve call.

Return type:

None

stop()

Stop the daemon; safe to call more than once.

Return type:

None

serve()

Start the daemon and block until a signal or stop request.

Return type:

None

payload(value=None, /, **extra)

Normalize a user result into a JSON-serializable dictionary.

Parameters:

• value (Any)

• extra (Any)

Return type:

dict

Programmatic client for getting and setting keywords on libby daemons.

Client CLI spins up a connection and holds it for its lifetime, so a script can touch many keywords cheaply. It resolves a qualified <group>.<daemon>.<keyword> to a peer and key, calls Libby.rpc, and turns the reply into a value or a LibbyError via libby.response.unwrap().

class libby.client.WaitResult(satisfied, address, value, elapsed_s, polls)

Bases: object

What a Client.wait_for_result() wait observed before it stopped.

Parameters:

• satisfied (bool)

• address (str)

• value (Any)

• elapsed_s (float)

• polls (int)

satisfied: bool

True if the expression became true; False if the timeout expired.

address: str

<group>.<daemon>.<keyword> address the expression polled.

value: Any

Last value read. None if no read ever succeeded.

elapsed_s: float

Wall-clock seconds spent waiting.

polls: int

Number of reads attempted.

class libby.client.KeyListing(names, writeonly, services)

Bases: object

One peer’s answer to a keys.list request.

writeonly names those of names that have no value to read, so a caller polling the peer can skip them. It is bare keyword names, matching matches on the wire rather than the qualified names above.

services lists the non-keyword keys the peer answers, such as keys.read. It is empty for a peer on a libby old enough not to report the field, which is how a caller decides whether bulk reads are available: an unknown key is dropped without an ACK, so probing for one looks exactly like a dead peer.

Parameters:

• names (Tuple[str, ...])

• writeonly (Tuple[str, ...])

• services (Tuple[str, ...])

names: Tuple[str, ...]

writeonly: Tuple[str, ...]

services: Tuple[str, ...]

class libby.client.Client(libby)

Bases: object

Long-lived, in-process handle for getting and setting libby keywords.

Parameters:

libby (Libby)

classmethod rabbitmq(*, self_id='libby-client', rabbitmq_url='amqp://localhost')

Connect over RabbitMQ.

Parameters:

• self_id (str)

• rabbitmq_url (str)

Return type:

Client

classmethod zmq(*, self_id='libby-client', bind='tcp://127.0.0.1:56001', address_book=None)

Connect over ZMQ, given an address book of peer endpoints.

Parameters:

• self_id (str)

• bind (str)

• address_book (Dict[str, str] | None)

Return type:

Client

classmethod from_config(path=None, *, self_id='libby-client')

Build a Client from cli_config.yaml, resolving transport as the CLI does.

Parameters:

• path (str | None)

• self_id (str)

Return type:

Client

show(name, *, timeout_s=3.0)

Read a keyword’s full response (value, units, flags).

Parameters:

• name (str)

• timeout_s (float)

Return type:

Dict[str, Any]

get(name, *, timeout_s=3.0)

Read a keyword’s value.

Parameters:

• name (str)

• timeout_s (float)

Return type:

Any

set(name, value, *, timeout_s=None)

Write a keyword and return the value the daemon applied.

Parameters:

• name (str)

• value (Any)

• timeout_s (float | None)

Return type:

Any

listing(pattern, *, timeout_s=3.0)

List a peer’s matching keywords and the services it serves, in one call.

Use this over list() when the caller also needs to know whether the peer supports bulk reads; both come from the same keys.list response, so asking costs no extra round trip.

Parameters:

• pattern (str)

• timeout_s (float)

Return type:

KeyListing

list(pattern, *, timeout_s=3.0)

List qualified keyword names matching <group>.<daemon>.<pattern>.

Returns fully qualified names, so the result feeds straight back into get(), show() or read(). A % in the group or daemon segment spans daemons by broadcast, so that form waits the full timeout_s (see peers()).

Parameters:

• pattern (str)

• timeout_s (float)

Return type:

List[str]

peers(pattern='%.%', *, timeout_s=1.0)

List the live daemons matching <group>.<daemon> as group.daemon ids.

Every daemon answers a broadcast keys.list, so this waits the full timeout_s rather than returning on the first reply. Over ZMQ only daemons in the address book are asked.

Parameters:

• pattern (str)

• timeout_s (float)

Return type:

List[str]

peer_listings(pattern='%.%', *, timeout_s=1.0)

Map each live daemon matching pattern to its keyword names.

A keyword segment in pattern narrows the names; without one every keyword is listed. Same timeout semantics as peers().

Clients answer the broadcast too, since every Libby serves keys.list, so a reply counts only if it reports itself as a daemon.

Parameters:

• pattern (str)

• timeout_s (float)

Return type:

Dict[str, List[str]]

describe(name, *, timeout_s=3.0)

Read one keyword’s metadata (type, access, units, timeout_s).

Parameters:

• name (str)

• timeout_s (float)

Return type:

Dict[str, Any]

read(names, *, timeout_s=3.0, chunk_size=100)

Read many keywords using one request per peer, per chunk.

Unlike get() and set(), this never raises for a failed read: every requested name maps to its own response dict, so one dead peer or one broken getter costs only its own entries. Names may span peers; each peer is requested separately, in the order given.

Parameters:

• names (Sequence[str])

• timeout_s (float)

• chunk_size (int)

Return type:

Dict[str, Dict[str, Any]]

wait_for(expression, timeout=None, *, daemon=None, case=False, poll_s=0.1, rpc_timeout_s=3.0)

Block until expression is true; return whether it became true.

expression is one comparison between a $-prefixed keyword and a literal — see libby.expression for the accepted syntax:

client.wait_for('$hsfei.pickoff.positionvalue > 15', timeout=5)
client.wait_for('$ismoving == false', 30, daemon='hsfei.pickoff')

Parameters:

• expression (str) – The condition to wait on.

• timeout (float | None) – Seconds to wait before giving up. None waits indefinitely; 0 evaluates once and returns.

• daemon (str | None) – Default <group>.<daemon>, so the expression can name a keyword bare.

• case (bool) – Compare strings case-sensitively.

• poll_s (float) – Seconds between reads.

• rpc_timeout_s (float) – Per-read RPC timeout.

Returns:

True if the expression became true, False if the timeout expired with it still false.

Raises:

• ExpressionError – the expression is malformed, or its two sides cannot be compared at all.

• KeywordError – the daemon rejected the read (e.g. unknown or write-only keyword) — a condition waiting cannot resolve.

Return type:

bool

wait_for_result(expression, timeout=None, *, daemon=None, case=False, poll_s=0.1, rpc_timeout_s=3.0)

Like wait_for(), but report what the wait observed.

Same arguments and same exceptions; returns a WaitResult instead of a bool, for callers that want to show the value the expression settled on (or timed out against).

Parameters:

• expression (str)

• timeout (float | None)

• daemon (str | None)

• case (bool)

• poll_s (float)

• rpc_timeout_s (float)

Return type:

WaitResult

close()

Disconnect the underlying transport.

Return type:

None

Keywords

Keyword classes for Libby.

class libby.keyword.Keyword(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Bases: object

Named value exposed as a libby service.

Parameters:

• name (str)

• getter (Optional[Getter])

• setter (Optional[Setter])

• units (Optional[str])

• description (str)

• nullable (bool)

• validator (Optional[Validator])

• timeout_s (Optional[float])

type_name: str = 'any'

property readonly: bool

property writeonly: bool

describe()

Return type:

dict

handle(payload)

Parameters:

payload (dict)

Return type:

dict

class libby.keyword.BoolKeyword(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Bases: Keyword

Keyword whose value is a Python bool.

Parameters:

• name (str)

• getter (Optional[Getter])

• setter (Optional[Setter])

• units (Optional[str])

• description (str)

• nullable (bool)

• validator (Optional[Validator])

• timeout_s (Optional[float])

type_name: str = 'bool'

class libby.keyword.IntKeyword(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Bases: Keyword

Keyword whose value is a Python int; bool is rejected.

Parameters:

• name (str)

• getter (Optional[Getter])

• setter (Optional[Setter])

• units (Optional[str])

• description (str)

• nullable (bool)

• validator (Optional[Validator])

• timeout_s (Optional[float])

type_name: str = 'int'

class libby.keyword.FloatKeyword(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Bases: Keyword

Keyword whose value is a Python float; accepts int, rejects bool.

Parameters:

• name (str)

• getter (Optional[Getter])

• setter (Optional[Setter])

• units (Optional[str])

• description (str)

• nullable (bool)

• validator (Optional[Validator])

• timeout_s (Optional[float])

type_name: str = 'float'

class libby.keyword.StringKeyword(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Bases: Keyword

Keyword whose value is a Python str.

Parameters:

• name (str)

• getter (Optional[Getter])

• setter (Optional[Setter])

• units (Optional[str])

• description (str)

• nullable (bool)

• validator (Optional[Validator])

• timeout_s (Optional[float])

type_name: str = 'string'

class libby.keyword.TriggerKeyword(name, *, action, description='', timeout_s=None)

Bases: Keyword

Write-only action keyword. Any modify fires action; show returns false.

Parameters:

• name (str)

• action (Action)

• description (str)

• timeout_s (Optional[float])

type_name: str = 'trigger'

property readonly: bool

property writeonly: bool

handle(payload)

Parameters:

payload (dict)

Return type:

dict

libby.keyword.match_pattern(pattern, names)

Return names matching pattern, sorted. % is a wildcard; it matches any run of characters excluding ..

Parameters:

• pattern (str)

• names (Iterable[str])

Return type:

List[str]

Per-Libby buffer for building Keyword instances with a typed-method API.

class libby.keyword_registry.KeywordRegistry

Bases: object

Collects Keyword instances built via typed methods.

Build keywords with bool / int / float / string / trigger; each call appends to an internal buffer. Call drain() to retrieve and clear the buffer; pass the result to Libby.register_keywords.

bool(name, *, getter=None, setter=None, description='', nullable=False, validator=None, timeout_s=None)

Parameters:

• name (str)

• getter (Callable[[], Any] | None)

• setter (Callable[[Any], None] | None)

• description (str)

• nullable (bool)

• validator (Callable[[Any], str | None] | None)

• timeout_s (float | None)

Return type:

BoolKeyword

int(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Parameters:

• name (str)

• getter (Callable[[], Any] | None)

• setter (Callable[[Any], None] | None)

• units (str | None)

• description (str)

• nullable (bool)

• validator (Callable[[Any], str | None] | None)

• timeout_s (float | None)

Return type:

IntKeyword

float(name, *, getter=None, setter=None, units=None, description='', nullable=False, validator=None, timeout_s=None)

Parameters:

• name (str)

• getter (Callable[[], Any] | None)

• setter (Callable[[Any], None] | None)

• units (str | None)

• description (str)

• nullable (bool)

• validator (Callable[[Any], str | None] | None)

• timeout_s (float | None)

Return type:

FloatKeyword

string(name, *, getter=None, setter=None, description='', nullable=False, validator=None, timeout_s=None)

Parameters:

• name (str)

• getter (Callable[[], Any] | None)

• setter (Callable[[Any], None] | None)

• description (str)

• nullable (bool)

• validator (Callable[[Any], str | None] | None)

• timeout_s (float | None)

Return type:

StringKeyword

trigger(name, *, action, description='', timeout_s=None)

Parameters:

• name (str)

• action (Callable[[], None])

• description (str)

• timeout_s (float | None)

Return type:

TriggerKeyword

add(keyword)

Buffer a Keyword built directly (e.g. via a stage helper).

Parameters:

keyword (Keyword)

Return type:

Keyword

add_all(keywords)

Buffer many keywords at once.

Parameters:

keywords (Iterable[Keyword])

Return type:

None

drain()

Return the buffered keywords and clear the buffer.

Return type:

List[Keyword]

Configuration and naming

exception libby.config.ConfigError

Bases: ConfigError, ValueError

Raised when daemon configuration cannot be selected or validated.

Subclasses the public libby.ConfigError too, so callers can catch one exception regardless of whether it came from daemon subsystem-config loading (here) or client cli_config.yaml loading (libby.config_resolve).

libby.config.load_config(path, *, optional=False)

Load a JSON or YAML configuration file.

A missing file raises FileNotFoundError unless optional, in which case {} is returned - used by the client’s optional cli_config.yaml.

Parameters:

• path (str | PathLike)

• optional (bool)

Return type:

Dict[str, Any]

libby.config.with_env_overrides(config, prefix='LIBBY_')

Apply top-level environment-variable overrides.

Parameters:

• config (Mapping[str, Any])

• prefix (str)

Return type:

Dict[str, Any]

libby.config.deep_merge(base, override)

Return a recursive merge without mutating either input.

Parameters:

• base (Mapping[str, Any])

• override (Mapping[str, Any])

Return type:

Dict[str, Any]

libby.config.is_subsystem_config(config, *, daemon_section='daemons')

Return whether a config contains a daemon collection.

Parameters:

• config (Mapping[str, Any])

• daemon_section (str)

Return type:

bool

libby.config.list_daemons(config, *, daemon_section='daemons')

List daemon IDs in a subsystem config.

Parameters:

• config (Mapping[str, Any])

• daemon_section (str)

Return type:

List[str]

libby.config.extract_daemon_config(full_config, daemon_id, *, daemon_section='daemons')

Merge subsystem defaults with one daemon’s overrides.

Parameters:

• full_config (Mapping[str, Any])

• daemon_id (str)

• daemon_section (str)

Return type:

Dict[str, Any]

class libby.config.DaemonConfigLoader(path, *, daemon_section='daemons')

Bases: object

Load a single-daemon or multi-daemon subsystem config.

Parameters:

• path (PathLike)

• daemon_section (str)

property config: Dict[str, Any]

property is_subsystem: bool

property subsystem: str | None

property daemon_ids: List[str]

get_daemon_config(daemon_id=None)

Parameters:

daemon_id (str | None)

Return type:

Dict[str, Any]

Resolve libby connection settings from cli_config.yaml plus overrides.

Shared by the libby CLI and the lib’s Client.from_config so the precedence rules (override → config file → built-in default) live in one place.

libby.config_resolve.load_cli_config(path=None)

Load the client’s cli_config.yaml. Missing file → empty dict.

The client config is optional, so it wraps libby.config.load_config with optional=True and re-raises parse/shape failures as ConfigError.

Parameters:

path (str | None)

Return type:

Dict[str, Any]

libby.config_resolve.resolve_transport(override, config)

Pick the transport: override → config → default; validated.

Parameters:

• override (str | None)

• config (Dict[str, Any])

Return type:

str

libby.config_resolve.resolve_rabbitmq_url(override, config)

Pick the RabbitMQ URL: override → config → default.

Parameters:

• override (str | None)

• config (Dict[str, Any])

Return type:

str

libby.config_resolve.parse_addr_kv(entry)

Parse a ‘peer_id=tcp://host:port’ address-book entry.

Parameters:

entry (str)

Return type:

Tuple[str, str]

libby.config_resolve.resolve_address_book(config, extra_addrs=None)

Merge the config ‘peers’ map with extra ‘peer=addr’ override strings.

Parameters:

• config (Dict[str, Any])

• extra_addrs (List[str] | None)

Return type:

Dict[str, str]

Keyword-name parsing, value coercion, and peer/group naming (transport-agnostic).

libby.naming.parse_keyword(arg, *, allow_pattern=False)

Parse ‘<group>.<daemon>.<keyword>’ into (group, daemon, keyword).

With allow_pattern=True, % is allowed in the keyword segment. Group and daemon must always be explicit.

Parameters:

• arg (str)

• allow_pattern (bool)

Return type:

Tuple[str, str, str]

class libby.naming.AddressPattern(group, daemon, keyword)

Bases: object

A <group>.<daemon>[.<keyword>] pattern with % allowed in every segment.

Parameters:

• group (str)

• daemon (str)

• keyword (str | None)

group: str

daemon: str

keyword: str | None

property spans_peers: bool

Return True when the group or daemon segment holds a wildcard.

property peer_pattern: str

Return the pattern to match wire ids against, lowercased like the ids.

libby.naming.parse_address_pattern(arg)

Parse <group>.<daemon>[.<keyword>] with % allowed anywhere.

Unlike parse_keyword() this admits patterns that span daemons, so it is only for callers that resolve them by broadcast (list, peers).

Parameters:

arg (str)

Return type:

AddressPattern

libby.naming.qualified_peer_id(peer_id, group_id=None)

Return the lowercased wire identity “<group_id>.<peer_id>”, or bare peer_id.

The single choke point both daemon and client build their identity from, so a short peer_id cannot collide across groups and case never matters.

Parameters:

• peer_id (str)

• group_id (str | None)

Return type:

str

libby.naming.peer_id(group, daemon)

Map an address’s group/daemon segments to the daemon peer id.

daemon is the peer_id, group is its group_id: the same pair qualified_peer_id joins on the daemon side, so client and daemon always agree on the wire identity by construction. See plans/peer_group_naming_design.md.

Parameters:

• group (str)

• daemon (str)

Return type:

str

libby.naming.coerce_value(value)

Coerce a modify value string.

Empty / ‘null’ → None; ‘true’/’false’ → bool; parseable as int → int; parseable as float → float; otherwise the original string.

Parameters:

value (str)

Return type:

Any

Expressions for wait_for: one keyword compared against one literal.

$hsfei.pickoff.positionvalue > 15
$hsfei.pickoff.ismoving == false
$hsfei.pickoff.status != Ready

Keyword references are prefixed with $, and are either fully qualified ($<group>.<daemon>.<keyword>) or, when a default daemon is supplied, a bare keyword ($<keyword>).

Values need quoting only when they contain whitespace, start with $, or look like an operator; one level of quoting is stripped. An unquoted value is coerced the way the CLI coerces a modify value, so false is a bool and 15 an int, while 'false' is the string.

Boolean operators, arithmetic, and multi-keyword expressions are not supported.

class libby.expression.Comparison(address, op, operand)

Bases: object

One parsed comparison: a keyword address, an operator, and a literal.

Always keyword-first, so a caller reads one keyword and compares one way; parse_comparison() flips the operator to make it so.

Parameters:

• address (str)

• op (str)

• operand (Any)

address: str

The keyword’s <group>.<daemon>.<keyword> address.

op: str

One of ==, !=, <, <=, >, >=.

operand: Any

The literal the keyword’s value is compared against.

evaluate(value, *, case=False)

Compare value (a keyword’s current value) against the operand.

String comparisons are case-insensitive unless case is true.

A value of None under an ordering operator is unsatisfied rather than an error, so a wait started before a daemon has populated a nullable keyword keeps waiting instead of failing.

Raises:

ExpressionError – the two types cannot be ordered at all (e.g. "ready" > 15), which no amount of waiting will fix.

Parameters:

• value (Any)

• case (bool)

Return type:

bool

libby.expression.parse_comparison(expression, *, daemon=None)

Parse expression into a Comparison.

Parameters:

• expression (str) – One comparison, e.g. '$hsfei.pickoff.softmax >= 120'.

• daemon (str | None) – Optional default <group>.<daemon>, letting the expression name a keyword bare ('$softmax >= 120').

Raises:

• ExpressionError – the expression is not a single keyword-to-literal comparison, or the default daemon is malformed.

• KeywordNameError – the resolved address is not a valid <group>.<daemon>.<keyword>.

Return type:

Comparison

Keygrabber

The InfluxDB sink is omitted here: it imports its client at module scope, and the docs build does not install the optional influxdb extra.

The sample the keygrabber collects, and the sink contract it writes to.

A Sample carries nothing backend-specific: turning one into measurements, fields, tags or columns is the sink’s job, so a second backend is a new sink rather than a change to the collector.

exception libby.keygrabber.sink.SinkError

Bases: LibbyError

A sink could not be reached, configured, or written to.

exception libby.keygrabber.sink.SinkWriteError

Bases: SinkError

A batch of samples could not be written and may be worth retrying.

class libby.keygrabber.sink.Sample(keyword, group, peer, value, units, timestamp)

Bases: object

One keyword’s value, read at one instant.

Parameters:

• keyword (str)

• group (str)

• peer (str)

• value (bool | int | float | str | None)

• units (str | None)

• timestamp (datetime)

keyword: str

group: str

peer: str

value: bool | int | float | str | None

units: str | None

timestamp: datetime

class libby.keygrabber.sink.RetryPolicy(max_batches=64, base_backoff_s=1.0, max_backoff_s=60.0)

Bases: object

Bounds on how a RetryingWriter queues and re-attempts batches.

Parameters:

• max_batches (int)

• base_backoff_s (float)

• max_backoff_s (float)

max_batches: int = 64

base_backoff_s: float = 1.0

max_backoff_s: float = 60.0

class libby.keygrabber.sink.Sink(*args, **kwargs)

Bases: Protocol

Destination for collected samples.

Implementations own their own schema. write returns how many samples it actually stored, which can be fewer than it was given when a backend cannot represent some of them, and raises SinkWriteError when the batch failed and should be retried.

connect()

Open the connection, or reopen it after a failure.

Return type:

None

is_connected()

Return whether the backend is currently reachable.

Return type:

bool

write(samples)

Store a batch and return how many samples were written.

Parameters:

samples (Sequence[Sample])

Return type:

int

close()

Release the connection.

Return type:

None

class libby.keygrabber.sink.RetryingWriter(sink, *, policy=None, clock=<built-in function monotonic>)

Bases: object

Wraps a sink, holding failed batches in a bounded queue for retry.

Retrying is backend-independent, so it lives here rather than inside any one sink. The queue is bounded and drops its oldest batch when full, so a database that stays down cannot grow the daemon’s memory without limit.

Nothing here starts a thread and nothing sleeps: the caller decides when to call flush_due(), and clock is injectable so backoff is testable without waiting for it.

Parameters:

• sink (Sink)

• policy (Optional[RetryPolicy])

• clock (Callable[[], float])

property queue_depth: int

Number of batches waiting to be retried.

property dropped_batches: int

Number of batches discarded because the queue was full.

property healthy: bool

Whether the most recent attempt on the sink succeeded.

Worth asking for, because write() deliberately does not raise when a batch fails: it queues it and returns 0, which is what makes retrying possible and what stops a caller learning about the failure from an exception.

property failed_attempts: int

Total attempts on the sink that failed, including retries.

write(samples)

Write a batch now, or queue it if the sink is in backoff.

Parameters:

samples (Sequence[Sample])

Return type:

int

flush_due()

Retry queued batches, oldest first, once backoff has elapsed.

Return type:

int

connect()

Reconnect the wrapped sink and clear its backoff.

Return type:

None

is_connected()

Return whether the wrapped sink is reachable.

Return type:

bool

close()

Release the wrapped sink.

Return type:

None

Parse and validate the keygrabber’s own configuration sections.

A keygrabber config file is an ordinary LibbyDaemon config (peer_id, group_id, transport, …) plus three sections this module owns: sink, defaults and collections.

class libby.keygrabber.config.CollectionConfig(name, group, daemon, keywords, exclude, interval_s, timeout_s, refresh_s)

Bases: object

One configured peer, keyword selection and cadence.

Parameters:

• name (str)

• group (str)

• daemon (str)

• keywords (Tuple[str, ...])

• exclude (Tuple[str, ...])

• interval_s (float)

• timeout_s (float)

• refresh_s (float)

name: str

group: str

daemon: str

keywords: Tuple[str, ...]

exclude: Tuple[str, ...]

interval_s: float

timeout_s: float

refresh_s: float

property peer: str

Return the qualified peer address this collection reads.

excluded()

Return every exclusion, including the defaults not named explicitly.

Return type:

Tuple[str, …]

class libby.keygrabber.config.KeygrabberConfig(collections, sink, workers, retry)

Bases: object

The keygrabber’s parsed configuration.

Parameters:

• collections (Tuple[CollectionConfig, ...])

• sink (Mapping[str, Any])

• workers (int)

• retry (RetryPolicy)

collections: Tuple[CollectionConfig, ...]

sink: Mapping[str, Any]

workers: int

retry: RetryPolicy

libby.keygrabber.config.parse_config(config)

Validate a daemon config’s keygrabber sections into typed objects.

Parameters:

config (Mapping[str, Any])

Return type:

KeygrabberConfig

libby.keygrabber.config.build_sink(sink_config)

Build the configured sink, importing its backend client on demand.

Parameters:

sink_config (Mapping[str, Any])

Return type:

Sink

libby.keygrabber.config.select_keywords(collection, available)

Return the bare keyword names this collection should read.

available is the peer’s full keyword list. Includes are applied first, then exclusions, so a default exclusion is dropped even when a wildcard would otherwise have matched it.

Parameters:

• collection (CollectionConfig)

• available (Sequence[str])

Return type:

Tuple[str, …]

One configured peer, resolved against the live peer and read on a cadence.

class libby.keygrabber.collection.TickResult(samples, read_errors, first_error=None)

Bases: object

What one read of a collection produced.

Parameters:

• samples (Tuple[Sample, ...])

• read_errors (int)

• first_error (str | None)

samples: Tuple[Sample, ...]

read_errors: int

first_error: str | None = None

class libby.keygrabber.collection.Collection(config, *, clock=<built-in function monotonic>)

Bases: object

Tracks what one peer exposes and turns a read of it into samples.

Resolution is refreshed periodically rather than once, so keywords added by a restarted daemon are picked up without restarting the keygrabber.

Parameters:

• config (CollectionConfig)

• clock (Callable[[], float])

last_sample: datetime | None

property name: str

Return the collection’s configured name.

property keyword_count: int

Return how many keywords the last resolve selected.

property bulk_read: bool

Return whether the peer advertised the bulk read service.

needs_resolve()

Return whether the keyword selection is due to be refreshed.

Return type:

bool

invalidate()

Force the next tick to resolve again, after a config change.

Return type:

None

note_failure()

Record a tick that read nothing.

Return type:

None

note_success()

Record a tick that read something, ending any backoff.

Return type:

None

backoff_interval_s()

Return the interval to use next, stretched while the peer fails.

A peer that is down would otherwise be retried, and logged about, on its configured cadence indefinitely. Backing off keeps a dead daemon from dominating both the logs and the read budget, while the cap keeps a recovered one from waiting long to be noticed.

Return type:

float

resolve(client)

Ask the peer what it serves and select the configured keywords.

One keys.list covers all of it: the peer’s keyword names, which of them cannot be read, and whether it serves keys.read. Selection is then local, so a collection with several patterns still costs one request.

Parameters:

client (Client)

Return type:

Tuple[str, …]

tick(client, timestamp)

Read every selected keyword once and return the samples.

Parameters:

• client (Client)

• timestamp (datetime)

Return type:

TickResult

Decides which collections are due to be read, and which to skip.

Deliberately free of threads, pools and sleeping: it answers “what should run now” and nothing else, so the due-time ordering and the skip rule can be tested against an injected clock rather than by racing real threads.

class libby.keygrabber.scheduler.Scheduler(*, clock=<built-in function monotonic>)

Bases: object

A due-time heap over collections, with at most one tick each in flight.

Every method takes the lock only for in-memory bookkeeping, never across a read or a write, because reload calls replace() from the transport’s receive thread and blocking that would time out every read already in flight.

Parameters:

clock (Callable[[], float])

replace(collections)

Swap the scheduled set, leaving in-flight ticks to finish.

Every new collection is due immediately, so a reload takes effect on the next pass rather than after a full interval.

Parameters:

collections (Iterable[Collection])

Return type:

None

next_delay()

Return seconds until the next tick, or None when nothing is scheduled.

Return type:

float | None

claim_due()

Claim every collection now due, and count those already running.

Returns the collections the caller should tick, and how many ticks were skipped because their predecessor had not finished. A skipped tick is not retried sooner: it waits for its next interval, so a slow peer settles at a lower rate instead of building a backlog.

Return type:

Tuple[Tuple[Collection, …], int]

update(collection)

Replace one collection and reschedule only it.

Used when a control keyword changes a cadence, so adjusting one collection does not re-seed every other collection’s due time and set the whole fleet reading at once.

Parameters:

collection (Collection)

Return type:

None

release(name)

Mark a collection’s tick finished, so its next one may run.

Parameters:

name (str)

Return type:

None

property in_flight: int

Number of ticks currently running.

property scheduled: Tuple[str, ...]

Names currently scheduled, in no particular order.

The keygrabber daemon: schedule reads, collect samples, write them out.

class libby.keygrabber.daemon.Counters

Bases: object

Tallies the daemon reports, guarded for cross-thread increments.

The lock is only ever held around an increment, never across a read or a sink write, because these are served on the transport’s receive thread and blocking it would time out every read already in flight.

add(**deltas)

Add to one or more counters.

Parameters:

deltas (int)

Return type:

None

set(**values)

Set one or more counters to an absolute value.

Parameters:

values (int)

Return type:

None

class libby.keygrabber.daemon.KeygrabberDaemon

Bases: LibbyDaemon

Polls keywords from other libby peers and writes them to a sink.

One process serves the whole fleet: cadence lives in this daemon’s config rather than in each hardware daemon, and the database credential lives in one place.

transport: str = 'rabbitmq'

discovery_enabled: bool = False

on_start(libby)

Parse config, open the sink, and start the reader and writer threads.

Parameters:

libby (Libby)

Return type:

None

on_stop(libby=None)

Stop the threads and close the sink once the writer has drained it.

The draining happens on the writer thread rather than here, so only one thread is ever inside the sink. This method just waits for it.

Parameters:

libby (Libby | None)

Return type:

None

classmethod from_config_file(path, daemon_id=None, *, env_prefix=None)

Build from a file, remembering where it came from for reload.

make_sink()

Build the configured sink.

Override to supply a sink the config cannot describe, or to inject one in a test without installing a database client.

Return type:

Sink

Responses and errors

Turn a bamboo RPC envelope into a keyword-response dict.

Libby.rpc returns a transport envelope, not the keyword response itself:

• {"status": "delivered", "resp": <payload|None>}

• {"status": "timeout", ...}

• {"status": "too_large", "mtu": ..., "size": ...}

The keyword <payload> is {"ok": True, "value": ..., "units"?: ...} or {"ok": False, "error": "..."}. unwrap raises on any failure; a caller that prefers a dict (e.g. the CLI’s table renderer) wraps it in its own non-raising adapter.

libby.response.unwrap(name, envelope)

Reduce an rpc envelope to its keyword-response dict, or raise.

Raises LibbyTimeout (not delivered / no response), KeywordError (daemon answered ok=False), or LibbyError (malformed / oversized).

Parameters:

• name (str)

• envelope (Any)

Return type:

Dict[str, Any]

Exceptions raised by libby core and lib code.

Core and library code raises these errors. Entry points (the CLI or a daemon) catch them and decide whether to exit the process or bubble up.

exception libby.errors.LibbyError

Bases: Exception

Base class for all libby errors.

exception libby.errors.ConfigError

Bases: LibbyError

A config file or connection setting is invalid.

exception libby.errors.KeywordNameError

Bases: LibbyError

A qualified keyword name is malformed.

exception libby.errors.ExpressionError

Bases: LibbyError

A wait_for expression is malformed, or cannot be evaluated.

exception libby.errors.LibbyTimeout

Bases: LibbyError

An RPC request was not delivered, or got no response, within its TTL.

exception libby.errors.KeywordError(name, error)

Bases: LibbyError

A daemon rejected a keyword get/set (responded with ok=False).

Parameters:

• name (str)

• error (str)

Return type:

None

Transports

class libby.zmq_transport.ZmqTransport(bind_router, address_book, my_id, group_id=None)

Bases: Transport

Simple ROUTER (bind) + per-peer DEALER (connect) transport.

• This peer binds a ROUTER at bind_router.

• For each remote peer_id in address_book, we lazily create a DEALER and set its ZMQ.IDENTITY to peer_id so the remote can see who sent.

• Incoming frames arrive on ROUTER as [IDENT, PAYLOAD] or [IDENT, b"", PAYLOAD]. We pass IDENT as “peer:<peer_id>” to the Protocol callback.

• Replies to a request we initiated over one of our own DEALER sockets arrive on that DEALER socket, not on our ROUTER (that’s how ZMQ’s ROUTER/DEALER pattern works: a reply travels back over whichever connection carried the request). Every DEALER we create is registered with the poller and mapped back to its peer_id so those replies are actually read instead of sitting unpolled forever.

Parameters:

• bind_router (str)

• address_book (Dict[str, str])

• my_id (str)

• group_id (str | None)

property group_id: str | None

property mtu: int

Best-effort max frame size in bytes.

start()

Return type:

None

stop()

Return type:

None

on_receive(cb)

Parameters:

cb (Callable[[str, bytes], None])

Return type:

None

reply_to(peer_id, frame)

Try to send directly back to the peer using the ROUTER routing-id we observed on the incoming request. Returns True if used, False otherwise.

Parameters:

• peer_id (str)

• frame (bytes)

Return type:

bool

send(dest, frame)

Send a frame to a destination.

• "peer:<peer_id>" or "<peer_id>" -> direct to that peer

• "broadcast:*" -> to all known peers

Dropped once stopped: sending lazily creates a DEALER per peer, and after stop() there is nothing left to close it. bamboo’s discovery announces from its own thread, which its stop() sets a flag for rather than joining, so a broadcast can arrive a whole announce interval after this transport was torn down.

Parameters:

• dest (str)

• frame (bytes)

Return type:

None

add_peer(peer_id, endpoint)

Dynamically add or update an endpoint for a peer.

Parameters:

• peer_id (str)

• endpoint (str)

Return type:

None

class libby.rabbitmq_transport.RabbitMQTransport(peer_id, rabbitmq_url='amqp://localhost', group_id=None)

Bases: object

RabbitMQ-based transport for Bamboo Protocol.

Architecture: - Each peer gets a unique queue: “libby.peer.<peer_id>” - Direct exchange “libby.direct” for peer-to-peer messages - Fanout exchange “libby.fanout” for broadcast messages - Each peer binds its queue to both exchanges

No address book needed - RabbitMQ broker handles all routing.

Parameters:

• peer_id (str)

• rabbitmq_url (str)

• group_id (str | None)

__init__(peer_id, rabbitmq_url='amqp://localhost', group_id=None)

Initialize RabbitMQ transport.

Parameters:

• peer_id (str) – Unique identifier for this peer

• rabbitmq_url (str) – RabbitMQ connection URL (e.g., “amqp://user:pass@host:5672/”)

• group_id (str | None) – Optional group identifier. If provided, included in queue name for future group functionality.

property group_id: str | None

property mtu: int

Maximum transmission unit - RabbitMQ can handle large messages.

start(ready_timeout_s=5.0)

Start consuming messages from RabbitMQ.

Blocks (up to ready_timeout_s) until the receive queue is declared, bound, and consuming, so callers can publish a request immediately after this returns without racing their own reply queue.

Parameters:

ready_timeout_s (float)

Return type:

None

stop()

Stop consuming and close RabbitMQ connection.

Return type:

None

on_receive(cb)

Register callback for incoming messages.

Parameters:

cb (Callable[[str, bytes], None]) – Callback function that receives (source, frame) where source is “peer:<peer_id>” and frame is raw bytes

Return type:

None

send(dest, frame)

Send a frame to a destination.

Parameters:

• dest (str) – Destination string, either: - "peer:<peer_id>" for direct peer-to-peer - "broadcast:*" for fanout to all peers

• frame (bytes) – Raw bytes to send (already serialized by Bamboo)

Return type:

None

Shared helper for building transport setup-error messages.

libby._transport_errors.describe_exception(exc)

Return a human-readable message for exc.

Some exceptions (e.g. pika’s AMQPConnectionError on a refused connection) have an empty str() and bury the real cause in repr()/args instead, which turns “transport setup failed” into a message with nothing after the colon. Fall back to repr() whenever str() has nothing useful to say.

Parameters:

exc (BaseException)

Return type:

str

CLI

Libby CLI — show/modify keywords on libby peers, plus raw req/sub.

libby.cli.libby_cli.cmd_show(namespace)

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_list(namespace)

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_describe(namespace)

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_modify(namespace)

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_waitfor(namespace)

Block until a keyword comparison holds, or the timeout expires.

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_req(namespace)

Raw RPC for debugging — works on either transport.

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_sub(namespace)

Subscribe to topics. ZMQ-only.

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.cmd_completion(namespace)

Print the shell code that wires TAB completion to the libby executable.

Parameters:

namespace (Namespace)

Return type:

int

libby.cli.libby_cli.build_parser()

Return type:

ArgumentParser

libby.cli.libby_cli.main(argv=None)

Parameters:

argv (List[str] | None)

Return type:

int