Client API

`one_liner.client`

client for controlling/monitoring one or more remote python objects.

`RouterClient`

`version` `property`

Return client version.

`server_version` `property`

Return the server version.

`init(protocol='tcp', interface='localhost', rpc_port='5555', broadcast_port='5556', context=None, name=None)`

Create and return a RouterClient instance and connect it to an existing RouterServer.

Parameters:

Name	Type	Description	Default
`protocol`	`Protocol`	a valid zmq protocol (`"tcp"`, `"inproc"`, etc.)	`'tcp'`
`interface`	`str`	protocol interface. For `"tcp"`, this is the ip address of the PC running the `RouterServer`.	`'localhost'`
`rpc_port`	`str`	port to issue remote function calls to python object instances on the corresponding `RouterServer`.	`'5555'`
`broadcast_port`	`str`	port to receive streamed messages from the corresponding `RouterServer`'s stream client.	`'5556'`
`context`	`Context \| None`	The zmq Context. If unspecified, this class instance will take the default global zmq context for this process.	`None`

Warnings

rpc_port and broadcast_port values must match those of the RouterServer that this RouterClient is connecting to.

Notes

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

`is_connected()`

True if the client is connected to the server.

`call_by_name(call_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

Lookup a configured call by its call_name, call it with the specified args/kwargs, and return the result.

Parameters:

Name	Type	Description	Default
`call_name`	`str`	The name of the call specified in [`add_named_call`][one_liner.server.ZMQRPCServer.add_named_call].	required
`args`	`list \| None`	Args to pass to the underlying function. Note that any arg passed in will overwrite any existing pre-configured arg setup in [`add_named_call`][one_liner.server.ZMQRPCServer.add_named_call].	`None`
`kwargs`	`dict \| None`	Kwargs to pass to the underlying function. Note that these kwargs will overwrite existing pre-configured args setup with [`add_named_call`][one_liner.server.ZMQRPCServer.add_named_call] via a standard dict update.	`None`
`deserializer`	`Encoding \| Callable`	Callable function to deserialize the data or string-representation of one of the built-in options.	`'pickle'`
`timeout_s`	`float \| None`	max time in seconds to wait for a reply before raising a TimeoutError.	`None`

Raises:

Type	Description
`RPCException`	if the remotely-executed function raised an error.
`TimeoutError`	if the connected server does not issue a response within the specified timeout.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

Returns:

Type	Description
`The result of the call with a timestamp.`

`call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

Call a function/method within the scope of the connected RouterServer and return the result.

Parameters:

Name	Type	Description	Default
`obj_name`	`str`	object name. (Class instance or module)	required
`attr_name`	`str`	a callable attribute	required
`args`	`list \| None`	list of positional arguments for function call	`None`
`kwargs`	`dict \| None`	dict of keyword arguments for function call	`None`
`deserializer`	`Encoding \| Callable`	callable function to deserialize the data or string-representation of one of the built-in options.	`'pickle'`
`timeout_s`	`float \| None`	max time in seconds to wait for a reply before raising a TimeoutError.	`None`

Raises:

Type	Description
`RPCException`	if the remotely-executed function raised an error.
`TimeoutError`	if the connected server does not issue a response within the specified timeout.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

Notes

This is a blocking call that returns after the response has been returned.

`lock_write_token(timeout_s=None)`

Context Manager to get access to the write token.

Parameters:

Name	Type	Description	Default
`timeout_s`	`float \| None`	the timeout (in seconds) to wait to acquire the the token before raising an error.	`None`

Raises:

Type	Description
`PermissionError:`	if the write token never becomes available before the specified timeout.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

`configure_stream(name, storage_type='queue', deserializer='pickle')`

Configure data received from a stream to either hold one the latest data ("cache") or to hold onto all data in a buffer ("queue") of size 1000.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name	required
`storage_type`	`Literal['queue', 'cache']`	`"cache"` or `"queue"`. If `"cache"`, calling `get_stream` will return the most recently received stream data and not buffer any incoming data. The `"queue"` option will buffer up to 1000 messages such that calling `get_stream` will return data in a first-in-first-out (FIFO) manner.	`'queue'`

`get_stream(name, block=False)`

Receive the results of a configured stream as 2-tuple where the first value is a RouterServer-specified timestamp and the second value is the stream data..

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name	required
`block`	`bool`	if true, block until new data arrives.	`False`

Raises:

Type	Description
`Again`	if block is False (default) and no data is present.
`StreamException`	if the connected `RouterServer`'s underlying function call raised an exception.

Warnings

This stream must first be configured on the RouterClient-side with configure_stream.

`enable_stream(name)`

Enable broadcasting of a stream by name. The connected RouterServer will start periodically calling the underlying stream function, and calls to get_stream(name) will return new data.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name	required

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. The connected RouterServer will stop periodically calling the underlying stream function, and calls to get_stream(name) will return no new data.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name.	required

Notes

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

`get_stream_configurations(as_dict=False)`

Get the configuration for all streams.

Parameters:

Name	Type	Description	Default
`as_dict`	`bool`	if `True`, get the schema representation as a dict. Otherwise, return a [`Streams`][one_liner.stream_schema.Streams] model.	`False`

`get_rpc_configurations(as_dict=False)`

Get the configuration for all RPCs.

`close()`

Close the connection to the RouterServer

`ZMQRPCClient`

`_monitor_request_socket()`

Monitor thread to update the connection state.

`is_connected()`

True if the client is connected to the server.

`call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

Call a remote function available to the connected RouterServer and return the result.

Raises:

Type	Description
`RPCException`	if the remotely-executed function raised an error.
`TimeoutError`	if the connected server does not issue a response within the specified timeout.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

`get_write_token(force=False, timeout_s=None)`

Request a write token or acquire one by force. Idempotent if you already have the valid write token.

Parameters:

Name	Type	Description	Default
`force`	`bool`	whether to force take the write token	`False`
`timeout_s`	`float \| None`	if not forcing, how long to wait for the write token before giving up (default is forever). A timeout of 0 will try exactly once.	`None`

Raises:

Type	Description
`PermissionError`	if the write token was not acquired.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

`lock_write_token(timeout_s=None)`

Context Manager to get access to the write token.

Parameters:

Name	Type	Description	Default
`timeout_s`	`float \| None`	the timeout (in seconds) to wait to acquire the the token before raising an error.	`None`

Raises:

Type	Description
`PermissionError:`	if the write token never becomes available before the specified timeout.
`RouterServerDisconnectedError`	if the associated RouterServer is disconnected

`ZMQStreamClient`

Connect to an instrument server (likely running on an actual instrument) and receive periodically broadcasted function call results.

`init(protocol='tcp', interface='localhost', port='5556', context=None)`

`configure_stream(name, storage_type='queue', deserializer='pickle')`

Create a subscriber socket to receive a specific topic and setup how to buffer data.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name.	required
`storage_type`	`Literal['queue', 'cache']`	`"queue"` -> FIFO. `"cache"` -> only the latest data is received.	`'queue'`

`get(stream_name, block=False)`

Return the timestamped data.

Raises:

Type	Description
`zmq.Again`	if block is `False` (default) and no data is present.
`StreamException`	if the underlying function raised an exception while being executed.

Client API

one_liner.client

RouterClient

version property

server_version property

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

is_connected()

call_by_name(call_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)

call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)

lock_write_token(timeout_s=None)

configure_stream(name, storage_type='queue', deserializer='pickle')

get_stream(name, block=False)

enable_stream(name)

disable_stream(name)

get_stream_configurations(as_dict=False)

get_rpc_configurations(as_dict=False)

close()

ZMQRPCClient

_monitor_request_socket()

is_connected()

call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)

get_write_token(force=False, timeout_s=None)

lock_write_token(timeout_s=None)

ZMQStreamClient

__init__(protocol='tcp', interface='localhost', port='5556', context=None)

configure_stream(name, storage_type='queue', deserializer='pickle')

get(stream_name, block=False)

`one_liner.client`

`RouterClient`

`version` `property`

`server_version` `property`

`init(protocol='tcp', interface='localhost', rpc_port='5555', broadcast_port='5556', context=None, name=None)`

`is_connected()`

`call_by_name(call_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

`call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

`lock_write_token(timeout_s=None)`

`configure_stream(name, storage_type='queue', deserializer='pickle')`

`get_stream(name, block=False)`

`enable_stream(name)`

`disable_stream(name)`

`get_stream_configurations(as_dict=False)`

`get_rpc_configurations(as_dict=False)`

`close()`

`ZMQRPCClient`

`_monitor_request_socket()`

`is_connected()`

`call(obj_name, attr_name, args=None, kwargs=None, deserializer='pickle', timeout_s=None)`

`get_write_token(force=False, timeout_s=None)`

`lock_write_token(timeout_s=None)`

`ZMQStreamClient`

`init(protocol='tcp', interface='localhost', port='5556', context=None)`

`configure_stream(name, storage_type='queue', deserializer='pickle')`

`get(stream_name, block=False)`