instrumentserver.base

Functions

decode(data)
encode(data)
recv(socket)
recvMultipart(socket)	Recieves the broadcast from a broadcast message. It should consist of 2 parts:
recv_router(socket)
send(socket, data[, use_string])
sendBroadcast(socket, name, message)	broadcasts the message. It will send 2 messages: First the name with the send more flag,
send_router(socket, identity, message)

Classes

Broadcaster(*args, **kwargs)

Mixin implementing the Broadcaster contract: an instrument that emits its own Broadcasts.

class instrumentserver.base.Broadcaster(*args: Any, **kwargs: Any)

Bases: object

Mixin implementing the Broadcaster contract: an instrument that emits its own Broadcasts.

The instrument calls broadcast() with a ParameterBroadcastBluePrint; the Server registers itself as a sink with add_broadcast_sink() when the instrument joins the Station. Used standalone (no sinks registered), broadcast() is a no-op.

Sinks are stored in a plain list: adding the same sink twice makes it receive every Broadcast twice. An exception raised inside one sink is logged and does not stop the remaining sinks.

The annotations of the public methods are strings on purpose: the client builds proxy methods by exec-ing the blueprint’s call signature string, and a rendered class annotation would reference a name the exec’d source cannot resolve. Keep new blueprint-carrying annotations quoted like these.

add_broadcast_sink(fn: Callable[[ParameterBroadcastBluePrint], None]) → None

Register fn as a sink that receives every Broadcast.

Parameters:

fn – callable taking a ParameterBroadcastBluePrint.

broadcast(bp: ParameterBroadcastBluePrint) → None

Send a Broadcast to every registered sink.

With no sinks registered this is a no-op. An exception raised inside one sink is logged and the remaining sinks still receive the Broadcast.

remove_broadcast_sink(fn: Callable[[ParameterBroadcastBluePrint], None]) → None

Remove a sink that was registered with add_broadcast_sink().

Removing a sink that is not registered does nothing.

instrumentserver.base.decode(data: str | bytes) → Any

instrumentserver.base.encode(data: Any) → str

instrumentserver.base.recv(socket: Socket) → Any

instrumentserver.base.recvMultipart(socket: Socket) → Tuple[str, Any]

Recieves the broadcast from a broadcast message. It should consist of 2 parts:

The first item is the name of the object sending it. Second part the message

instrumentserver.base.recv_router(socket: Socket) → Tuple[bytes, Any]

instrumentserver.base.send(socket: Socket, data: Any, use_string: bool = True) → Any

instrumentserver.base.sendBroadcast(socket: Socket, name: str, message: Any) → None

broadcasts the message. It will send 2 messages: First the name with the send more flag,

followed by the message.

Parameters:

• socket – The socket sending it.

• name – The name of the object, it will be the first part.

• messages – The data to send.

instrumentserver.base.send_router(socket: Socket, identity: bytes, message: Any) → None