Server API

one_liner.server

Server for enabling remote control and broadcasting results of periodic function calls.

RouterServer

version property

Get the server version.

__init__(protocol='tcp', interface='*', rpc_port='5555', broadcast_port='5556', context=None, instances=None, config=None, name=None)

Constructor.

Parameters:

Name	Type	Description	Default
protocol	str	A zmq supported protocol (tcp, inproc, etc).	'tcp'
interface	str	The interface to bind or connect to.	'*'
rpc_port	int	Port to issue remote procedure calls to python objects.	'5555'
broadcast_port	int	Port from which to stream periodic data.	'5556'
context	Context	The zmq context. Will be created automatically if unspecified.	None
instances	dict	Dict of object instances, keyed by name.	None
config	RouterServerConfig | dict[str, dict] | None	optional configuration dict of named calls and streams	None

Warnings

For sharing data within the same process using the inproc protocol, the zmq Context must be shared between RouterServer and RouterClient (or RouterServer and RouterServer if forwarding).

Notes

For the protocol setting, some options are system-dependent (i.e: ipc is for unix-like OSes only).

run(block=False)

Setup rpc listener and broadcaster.

Parameters:

Name	Type	Description	Default
block	bool	return immediately. Otherwise, run the streamer in the current thread and block (i.e: do not return).	False

add_named_call(call_name, obj_name, attr_name, args=None, kwargs=None, access_type=None)

Setup a call to be called with call_by_name on the RouterClient.

Parameters:

Name	Type	Description	Default
call_name	str	String to save the function call signature under.	required
obj_name	str	Underlying object instance name. Must be present in the objects dict passed into the __init__.	required
attr_name	str	Name of the callable attribute (method).	required
args	tuple	Default args to save with the function call.	None
kwargs	dict	Default kwargs to save with the function call.	None
access_type	AccessType | None	user annotation related to the effect that calling the function has on the remote object's state. This value is optional and appears as part of the metadata available from :py:meth:~one_line.client.RouterClient.get_rpc_configurations	None

Notes

args and kwargs can be overwritten by index or name respectively when actually calling the named function call with call_by_name.

add_stream(stream_name, frequency_hz, obj_name, attr_name, args=None, kwargs=None, enabled=True, serializer='pickle')

Create a stream from a callable object attribute in the instance dict.

Parameters:

Name	Type	Description	Default
stream_name	str		required
frequency_hz	float	frequency at which to call the underlying function	required
obj_name	str	name of instance in the instances dict	required
attr_name	str	name of callable instance attribute (a method)	required
args	list | None	any function arguments	None
kwargs	dict | None	any function keyword arguments	None
enabled	bool	if true, start with the stream enabled.	True
serializer	Encoding | Callable	callable function to serialize the data or string representation of built-in serializer (or None if the data is already serialized)	'pickle'

add_stream_from_callable(stream_name, frequency_hz, func, args=None, kwargs=None, enabled=True, serializer='pickle')

Create a stream.

i.e: Setup a function to be called with specific arguments at a set frequency. If the function is already being broadcasted, update the broadcast parameters.

Parameters:

Name	Type	Description	Default
name	str	Stream name.	required
frequency_hz	float or int	Frequency at which to call the underlying function.	required
func	callable	Function to call.	required
args	tuple	Any function arguments.	None
kwargs	dict	Any function keyword arguments.	None
enabled	bool	If true, start with the stream enabled.	True
serializer	callable or str	Callable function to serialize the data or string representation of built-in serializer (or None if the data is already serialized).	'pickle'

Examples:

>>> import cv2
>>> video = cv2.VideoCapture(0)  # Get the first available camera.
>>> def get_frame():
...     return video.read()[1]  # just get the frame.
>>> server = RouterServer()
>>> server.add_stream(
...     "live_video",  # name of the stream
...     30,            # How fast to call this function.
...     get_frame,     # func to call.
... )
>>> server.run()

add_zmq_stream(name, address, enabled=True, log_chatter=False)

Add a stream from an existing zmq PUB socket source (including another existing RouterServer.

Parameters:

Name	Type	Description	Default
name	str	stream name	required
address	str	zmq socket address: {protocol}://{interface}:{port}	required
enabled	bool	if True, start enabled.	True
log_chatter	bool	if True, intercept messages in the connected PUB socket and add them to the logs (if the data length is short).	False

get_rpc(as_dict=False)

Get a breakdown of every RPC with its corresponding function signature.

get_stream_fn(name, set_timestamp=False, serializer='pickle')

Get a function to broadcast the specified stream name.

Useful if the application is creating data at its own rate and needs a callback function to call upon producing new data. This implicitly adds a manual stream to the configuration.

Parameters:

Name	Type	Description	Default
name	str	Stream name.	required
serializer	callable or str	Callable function to serialize the data or string representation of a built-in serializer.	'pickle'
set_timestamp	bool	If true, return a function who's first argument is the timestamp to be set for the packet.	False

Returns:

Type	Description
callable	A broadcast function that accepts data to be streamed.

Examples:

>>> send_func = server.get_stream_fn("live_video_feed")
>>> video = cv2.VideoCapture(0)  # Connect to first available camera.
>>> # Send images as soon as we can get them off the camera.
>>> while True:
...     new_frame = video.read()[1]  # Get new video frame
...     send_func(new_frame)

enable_stream(name)

Enable broadcasting of a stream by name.

Any connected RouterClient will start receiving data from this stream after they have configured how to buffer the stream data with configure_stream.

Parameters:

Name	Type	Description	Default
name	str	Stream name.	required

Raises:

Type	Description
KeyError	If the stream name does not exist.
ValueError	If the stream exists but cannot be enabled/disabled.

Notes

Enabling streams only works for periodically-added streams added with add_stream and add_zmq_stream but not get_stream_fn.

disable_stream(name)

disable broadcasting of a stream by name.

remove_stream(name)

Remove an existing stream.

The stream must be re-added if needed later. Consider using enable_stream and disable_stream instead if you need to conditionally throttle whether a stream is sending data.

get_version()

Get the server version.

close()

Close the RPC and Stream clients.