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.

static 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 arguments passed to the exception.

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: AbstractContextManager

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 (dict[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 (dict[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: AbstractContextManager

Abstract Base class to ServiceHost and ServiceClient.

abstractmethod close() ¶

Close the service and release all resources.

class bpod_core.ipc.ServiceClient ¶

Bases: ServiceBase, Generic[U]

A client for communicating with ServiceHost.

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, by default None.
event_handler (Callable[[dict], Any] | None, default: None) – A callback to handle PUB messages, by default None.
discovery_timeout (float, default: 10.0) – Timeout in seconds for service discovery.
txt_properties (dict | None, default: None) – Properties for service filtering during discovery, by default None.
default_reply_type (type[TypeVar(U)] | None, default: None) – The default data type for incoming replies (type[U]); the client is parameterized on this type.
remote (bool, default: True) – Whether to use Zeroconf for discovering remote services, by default True.

close() ¶

Close the client and clean up resources.

request(request_data, reply_type=None) ¶

Send a generic request to the server.

Parameters:

request_data (Any) – The request payload.
reply_type (type[TypeVar(T)] | None, default: None) – Override the expected reply type.

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.

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

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

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.
properties (dict[str, str | None] | None, default: None) – Additional properties for service advertisement.
uuid (UUID | None, default: None) – UUID for local IPC. Will be generated if not provided.
event_handler (Callable[[Any], Any] | None, default: None) – Function to handle incoming requests.
port_pub (int | None, default: None) – TCP port to bind the PUB socket. If None, a random available port is chosen.
port_rep (int | None, default: None) – TCP port to bind the REP socket. If None, a random available port is chosen.
serialization (Literal['json', 'msgpack'], default: 'msgpack') – Serialization format for message encoding. Can be either ‘msgpack’ or ‘json’.
default_request_type (type | Union | None, default: None) – The default data type for decoding incoming requests. Pass a tagged union here to dispatch requests on a msgspec tag.
remote (bool, default: True) – If True, binds TCP sockets to ‘0.0.0.0’. Otherwise, binds to ‘127.0.0.1’.

close() ¶

Close the host and clean up resources.

class bpod_core.ipc.ServiceIterator ¶

Bases: Iterator[ServiceEvent], AbstractContextManager

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 (dict[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 (dict[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:

str – The service address, e.g., ‘tcp://192.168.1.10:1234’.
dict – A dictionary of service properties.

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 (dict[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.T = ~T¶: Per-call reply type for ServiceClient.

bpod_core.ipc.U = ~U¶: Default reply type for ServiceClient, set at construction.