API Reference

Generated from docstrings in the libby package.

Core

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

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)

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)

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)

Return type:

Libby

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

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

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)

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.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

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]

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

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

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.build_parser()

Return type:

ArgumentParser

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

Parameters:

argv (List[str] | None)

Return type:

int