bpod_core.ipc#

Inter-process Communication, service discovery and related.

Exceptions#

exception bpod_core.ipc.RemoteError #

Bases: ServiceError

Exception representing an error on the remote side.

exception bpod_core.ipc.ServiceError #

Bases: Exception

Base exception for IPC service errors.

Classes#

class bpod_core.ipc.ErrorData #

Bases: Struct

A struct representing error data.

classmethod from_exception(exception=None) #

Serialize an exception to ErrorData.

Parameters:

exception (BaseException | None, default: None) – The exception to serialize.

Returns:

An ErrorData struct containing the serialized exception data.

Return type:

ErrorData

Raises:

ValueError – If no exception is provided and no active exception is available.

args: tuple#

The exception’s arguments.

message: str#

The error message.

name: str#

The name of the exception class.

traceback: str | None#

The formatted traceback of the exception.

class bpod_core.ipc.LocalServiceAdvertisement #

Bases: object

File-based local service advertisement for IPC discovery.

Advertises a service by writing a JSON file to the user’s runtime directory. This provides a lightweight alternative to Zeroconf for discovering services on the same machine. Stale advertisements (from dead processes) are automatically cleaned up during discovery.

The advertisement is automatically removed when the instance is garbage collected or when close() is called explicitly.

Parameters:

  • service_name (str) – The name of the service being advertised (e.g., ‘Bpod 3’).

  • service_type (str) – The type of service being advertised (e.g., ‘bpod’).

  • address (str) – The address where the service can be reached (e.g., ‘ipc:///tmp/foo.ipc’).

  • properties (Mapping[str, str | None] | None, default: None) – Additional key-value properties to advertise with the service.

  • pid (int | None, default: None) – Process ID of the service. Used to detect stale advertisements.

  • uuid (UUID | None, default: None) – Unique identifier for this service instance. Generated if not provided.

Examples

Advertise a service:

>>> ad = LocalServiceAdvertisement('Bpod1', 'bpod', 'tcp://127.0.0.1:5555')

Discover advertised services:

>>> services = list(LocalServiceAdvertisement.discover('bpod'))

Notes

Importing this class suppresses debug-level log messages from the filelock logger, as the per-file lock/unlock events it emits are too noisy for routine use. To re-enable them:

logging.getLogger('filelock').setLevel(logging.DEBUG)
static discover(service_type, properties=None) #

Discover locally advertised services.

Parameters:

  • service_type (str) – The service type to discover.

  • properties (Mapping[str, str | None] | None, default: None) – Properties to match against the service’s properties.

Yields:

LocalServiceInfo – Information structure describing the discovered services.

Return type:

Iterator[LocalServiceInfo]

close() #

Remove the service advertisement and clean up empty directories.

service_file: Path#

Path to the advertisement file.

class bpod_core.ipc.LocalServiceInfo #

Bases: Struct

Information about a locally advertised service.

Yielded by LocalServiceAdvertisement.discover().

address: str#

The address where the service can be reached.

pid: int#

Process ID of the service.

properties: dict[str, str | None]#

Additional key-value properties to advertise with the service.

service_name: str#

The name of the service being advertised.

service_type: str#

The type of service being advertised.

uuid: UUID#

Unique identifier for the service instance.

class bpod_core.ipc.ServiceBase #

Bases: ABC

Abstract Base class to ServiceHost and ServiceClient.

abstractmethod close() #

Close the service and release all resources.

class bpod_core.ipc.ServiceClient #

Bases: ServiceBase, Generic[Q, R, E]

A client for communicating with ServiceHost.

The client connects to the host’s REQ/REP channel - either directly via address or by discovering the service - and adopts the host’s serialization format during an initial handshake. If an event_handler is given, the client additionally subscribes to the host’s PUB/SUB channel and forwards each decoded event to the handler from a background thread.

The type parameters are the request payload type accepted by request() (Q, set via annotation), the default reply type (R, inferred from default_reply_type), and the event type decoded from PUB messages (E, inferred from event_type). All default to Any.

Parameters:

  • service_type (str) – The service type to discover or connect to.

  • address (str | None, default: None) – The direct connection address for the REQ channel. If None (default), the service is discovered via service_type and txt_properties.

  • event_handler (Callable[[TypeVar(E)], object] | None, default: None) – A callback to handle PUB messages, by default None. Called with the decoded message (E, see event_type); its return value is ignored. Runs on a background thread and should not block; exceptions it raises are logged but not propagated.

  • discovery_timeout (float, default: 10.0) – Timeout in seconds for service discovery.

  • txt_properties (Mapping[str, str | None] | None, default: None) – Properties for service filtering during discovery, by default None.

  • default_reply_type (TypeForm[TypeVar(R)], default: typing.Any) – The default data type for incoming replies (R); the client is parameterized on this type. May be a tagged union; Any by default.

  • event_type (TypeForm[TypeVar(E)], default: typing.Any) – The data type for decoding incoming PUB messages (E), independent of default_reply_type; Any by default. Pass a tagged union here to dispatch published messages on a msgspec tag. Requires an event_handler, whose parameter must accept this type.

  • remote (bool, default: True) – Whether to use Zeroconf for discovering remote services, by default True.

Raises:

  • ValueError – If event_type is given without an event_handler.

  • TimeoutError – If no matching service is discovered within discovery_timeout, or the host does not answer the handshake in time.

  • RemoteError – If the host reports an error during the handshake.

  • ServiceError – If communication with the host fails during the handshake.

close() #

Close the client and clean up resources.

request(request_data, reply_type=None, timeout=None) #

Send a generic request to the server.

Parameters:

  • request_data (TypeVar(Q)) – The request payload.

  • reply_type (type[TypeVar(T)] | None, default: None) – Override the expected reply type.

  • timeout (float | None, default: None) – Maximum time in seconds to await the reply. By default, the call blocks until a reply arrives or the client is closed.

Returns:

The deserialized reply from the server. If reply_type is given, returns an instance of reply_type; otherwise returns an instance of default_reply_type defined during instantiation of the class.

Return type:

Any

Raises:

  • RemoteError – If an error occurred on the host side.

  • ServiceError – If the host sent an unexpected reply kind or the client has been closed.

  • TimeoutError – If no reply arrived within timeout seconds.

property address_req: str #

The address for the REQ channel.

property address_sub: str #

The address for the SUB channel.

is_local: bool#

Whether the client is connected to a service on localhost.

class bpod_core.ipc.ServiceEvent #

Bases: NamedTuple

A service discovery event yielded by iter_services().

address: str#

The address of the service.

kind: Literal['added', 'removed']#

added when a service appears, removed when it disappears.

properties: dict[str, str | None]#

The properties of the service.

class bpod_core.ipc.ServiceHost #

Bases: ServiceBase, Generic[Q, E]

A ZeroMQ host providing REQ/REP and PUB/SUB sockets with service discovery.

Provides two communication channels: a REQ/REP channel for synchronous request-reply messaging and a PUB/SUB channel for broadcasting events to subscribers. Incoming requests are dispatched to a user-provided request_handler callback. Events are broadcast with publish(); subscribers decode them via the event_type of ServiceClient.

The type parameters are the request type decoded from incoming requests (Q, inferred from request_type) and the published event type (E, set via annotation); calls to publish() are checked against E. Both default to Any.

The service is automatically advertised for discovery by ServiceClient. Local advertisement uses the LocalServiceAdvertisement class. When remote=True, the service is additionally advertised via Zeroconf (mDNS) for network-wide discovery and remote-process communication.

Parameters:

  • service_name (str) – Service name to advertise.

  • service_type (str) – Service type.

  • request_handler (Callable[[TypeVar(Q)], Any]) – Function to handle incoming requests. Called with the decoded request payload (Q, see request_type); its return value is sent to the client as the reply. Runs on a background thread (calls are serialized); exceptions it raises are caught and sent to the client, where they raise RemoteError.

  • properties (Mapping[str, str | None] | None, default: None) – Additional properties for service advertisement.

  • uuid (UUID | None, default: None) – UUID identifying the service instance; determines the IPC addresses, the local advertisement file, and the TCP ports remembered across restarts. Will be generated if not provided.

  • port_pub (int | None, default: None) – Preferred TCP port for the PUB socket; takes precedence over the port remembered for uuid. If unavailable, a random port is chosen instead.

  • port_rep (int | None, default: None) – Preferred TCP port for the REP socket; takes precedence over the port remembered for uuid. If unavailable, a random port is chosen instead.

  • serialization (Literal['json', 'msgpack'], default: 'msgpack') – Serialization format for message encoding. Can be either ‘msgpack’ or ‘json’.

  • request_type (TypeForm[TypeVar(Q)], default: typing.Any) – The data type for decoding incoming requests (Q); the request_handler’s parameter must accept this type. May be a tagged union; Any by default.

  • remote (bool, default: True) – If True, binds TCP sockets to ‘0.0.0.0’. Otherwise, binds to ‘127.0.0.1’.

Raises:

ValueError – If serialization is not a supported format.

static get_metadata(option, default=None) #

Read a metadata value from the request currently being handled.

Intended to be called from within a request_handler to access per-request connection information, such as the client’s address ('Peer-Address') or application metadata (e.g. 'X-Hostname').

Parameters:

  • option (int | str) – An integer frame property (e.g. zmq.SRCFD) or a string metadata key (e.g. 'Peer-Address').

  • default (int | bytes | str | None, default: None) – The value to return if no request is currently being handled or the metadata option is unavailable. Defaults to None.

Returns:

The requested value, or default if no request is currently being handled or the option is unavailable.

Return type:

int | bytes | str | Any

close() #

Close the host and clean up resources.

publish(data) #

Broadcast a message to all subscribers over the PUB/SUB channel.

If no subscriber is connected (see has_subscribers), the message is dropped without being encoded - mirroring what the socket would do with it.

Parameters:

data (TypeVar(E)) – The payload to broadcast. Encoded with the host’s serialization protocol.

Raises:

ServiceError – If the message cannot be encoded or published.

wait_for_subscribers(timeout) #

Block until someone subscribes to the PUB/SUB channel.

Returns as soon as has_subscribers becomes True - i.e., once a subscription registers or a client handshake starts the grace period - or when the timeout expires.

Parameters:

timeout (float) – Maximum time to wait, in seconds.

Returns:

True if a subscriber registered within the timeout, False otherwise.

Return type:

bool

property has_subscribers: bool #

Whether anyone is subscribed to the PUB/SUB channel.

True while at least one subscriber is connected, or briefly after a client handshake (a subscription typically follows within the grace period). The flag is refreshed by the event loop and may lag actual subscription changes by up to one poll interval. It signals “someone subscribed to something” - there is no per-topic granularity.

pub_tcp_addr: str#

TCP address of the PUB socket, as handed out to clients.

pub_tcp_port: int#

TCP port of the PUB socket.

rep_tcp_addr: str#

TCP address of the REP socket, as handed out to clients.

rep_tcp_port: int#

TCP port of the REP socket.

class bpod_core.ipc.ServiceIterator #

Bases: Iterator[ServiceEvent]

Lazy iterator for service discovery events.

Monitors both local (IPC) and remote (Zeroconf/TCP) services, yielding ServiceEvent instances as services appear and disappear. Local services are always preferred — if a service is reachable both via IPC and TCP, only the IPC address is yielded.

Parameters:

  • service_type (str) – The service type to discover, e.g., 'bpod'.

  • properties (Mapping[str, str | None] | None, default: None) – Dictionary of expected service properties to match.

  • timeout (float | None, default: 10.0) – How many seconds to monitor. Pass None to monitor indefinitely until the iterator is closed.

  • poll_interval (float, default: 1.0) – How often to poll for local service changes, in seconds.

  • local (bool, default: True) – Whether to search for services on the local machine.

  • remote (bool, default: True) – Whether to also search for services on the network.

Notes

Prefer iter_services() over instantiating this class directly.

close() #

Stop monitoring and release all resources.

Functions#

bpod_core.ipc.discover(service_type, properties=None, *, timeout=10.0, poll_interval=1.0, local=True, remote=True) #

Discover a device/service on the local network matching given properties.

Parameters:

  • service_type (str) – The service type to discover, e.g., ‘bpod’

  • properties (Mapping[str, str | None] | None, default: None) – Dictionary of expected service properties to match.

  • timeout (float, default: 10.0) – How many seconds to wait for a matching service before timing out.

  • poll_interval (float, default: 1.0) – How often to poll for local service changes, in seconds.

  • local (bool, default: True) – Whether to search for a matching service on the local machine.

  • remote (bool, default: True) – Whether to search for a matching service on the network.

Return type:

tuple[str, dict[str, str | None]]

Returns:

Raises:

TimeoutError – If no matching device/service is found within the timeout period.

bpod_core.ipc.iter_services(service_type, properties=None, *, timeout=10.0, poll_interval=1.0, local=True, remote=True) #

Discover all services matching the given type and properties.

Continuously monitors both local (IPC) and remote (Zeroconf/TCP) services, yielding 'added' and 'removed' events as services appear and disappear.

Parameters:

  • service_type (str) – The service type to discover, e.g., ‘bpod’.

  • properties (Mapping[str, str | None] | None, default: None) – Dictionary of expected service properties to match.

  • timeout (float | None, default: 10.0) – How many seconds to monitor. Pass None to monitor indefinitely until the iterator is closed.

  • poll_interval (float, default: 1.0) – How often to poll for local service changes, in seconds.

  • local (bool, default: True) – Whether to search for services on the local machine.

  • remote (bool, default: True) – Whether to also search for services on the network.

Yields:

ServiceEvent – A named tuple with the following fields:

  • kind: str, either 'added' or 'removed'

  • address: str, the service address, e.g., 'tcp://192.168.1.10:1234'

  • properties: dict, the service properties, e.g., {'name': 'MyDevice'}

Return type:

ServiceIterator

Examples

Print events as services appear and disappear:

for event in iter_services('bpod', timeout=None):
    print(event.kind, event.address)

Attributes#

bpod_core.ipc.Serialization#

Serialization supported by ServiceHost and ServiceClient.

alias of Literal[‘json’, ‘msgpack’]

bpod_core.ipc.T = ~T#

Per-call reply type for ServiceClient.

bpod_core.ipc.R = ~R#

Default reply type for ServiceClient, set at construction.

bpod_core.ipc.Q = ~Q#

Request type; parameterizes ServiceClient (set via annotation) and ServiceHost (inferred from request_type, correlated with the request_handler).

bpod_core.ipc.E = ~E#

Event type; parameterizes ServiceHost (set via annotation) and ServiceClient (inferred from event_type, correlated with the event_handler).