PyMoa-Remote Package

PyMoa-Remote

PyMoa-Remote is a library to remotely execute methods e.g. on a remote Raspberry Pi, in another process, or another local thread.

Exception

exception pymoa_remote.exception.RemoteException: Bases: Exception

pymoa_remote.exception.extract_frames(exception: BaseException) → list

pymoa_remote.exception.get_fake_traceback_obj(filename, lineno, name, line)

pymoa_remote.exception.get_traceback_from_frames(frames)

pymoa_remote.exception.raise_remote_exception_from_frames(frames=(), err_type=None, value='')

pymoa_remote.exception.serialize_exception(e)

Utilities

Module that provides helpful classes and functions.

class pymoa_remote.utils.MaxSizeErrorDeque(max_size=0, **kwargs)

Bases: object

Async queue that raises an error on the read side as soon as the queue is full and the queue will silently stop taking items.

It is thread safe.

add_item(item, size=1)

max_size: int = 0: Approximate maximum size (maybe over with multiple threads).

queue: deque = None

receive_channel: MemoryReceiveChannel = None

send_channel: MemorySendChannel = None

size: int = 0

size_lock: lock = None

stopped: bool = False

thread_id = None

token: TrioToken = None

class pymoa_remote.utils.MaxSizeSkipDeque(max_size=0, **kwargs)

Bases: object

Async queue that skips appends when full, but indicates to consumer that packets were skipped.

add_item(item, size=1, force=False)

max_size = 0

packet: int = 0

receive_channel: MemoryReceiveChannel = None

send_channel: MemorySendChannel = None

size: int = 0

exception pymoa_remote.utils.QueueFull: Bases: Exception

pymoa_remote.utils.asynccontextmanager(func)

@asynccontextmanager decorator.

Typical usage:

@asynccontextmanager async def some_async_generator(<arguments>):

<setup> try:

yield <value>

finally:
<cleanup>

This makes this:

async with some_async_generator(<arguments>) as <variable>:
<body>

equivalent to this:

<setup> try:

<variable> = <value> <body>

finally:
<cleanup>

for ... in pymoa_remote.utils.get_class_bases(cls)

Gets all the base-classes of the class.

Parameters:: cls
Returns:

Server Side Executor

class pymoa_remote.server.DataLogger

Bases: object

Data logger used to log all data updates and stream it to clients.

log_event_callback(name, hash_name, callback, obj, *args)

log_item(hash_name, props=None, trigger_name=None, trigger_value=None)

log_property_callback(name, hash_name, callback, obj, value)

log_trigger_event_callback(name, tracked_props, hash_name, callback, obj, *args)

log_trigger_property_callback(name, tracked_props, hash_name, callback, obj, value)

start_logging(callback: Callable, obj: Any, hash_name: str, trigger_names: Iterable[str] = (), triggered_logged_names: Iterable[str] = (), logged_names: Iterable[str] = (), initial_properties: Iterable[str] = ())

logged_names cannot have events if trigger is not empty.

Can’t have prop bound as trigger and as name without trigger (causes dups in SSELogger).

stop_logging(obj, binding)

class pymoa_remote.server.ExecutorServer(registry: RemoteRegistry = None, executor: ExecutorBase = None, **kwargs)

Bases: ExecutorServerBase

Concrete server side handler of remote object method execution.

await call_instance_method(obj: Any, method_name: str, args: tuple, kwargs: dict) → Any

call_instance_method_gen(obj: Any, method_name: str, args: tuple, kwargs: dict) → Any

decode(data)

encode(data)

executor: ExecutorBase | None = None

registry: RemoteRegistry = None

stream_data_logger: DataLogger = None

class pymoa_remote.server.ExecutorServerBase

Bases: object

Base class for the server side handling of remote object method execution.

allow_import_from_main = False

allow_remote_class_registration = False

decode(data)

await delete_instance(*args, **kwargs)

encode(data)

await ensure_instance(*args, **kwargs) → bool

await execute(*args, **kwargs)

await execute_generator(*args, **kwargs)

await get_echo_clock(*args, **kwargs)

await get_object_config(*args, **kwargs)

await get_object_data(*args, **kwargs)

await get_objects(*args, **kwargs)

post_stream_channel(data, channel, hash_name)

await register_remote_class(*args, **kwargs)

await remote_import(*args, **kwargs)

await sleep(*args, **kwargs) → int

await start_logging_object_data(*args, **kwargs)

await stop_logging_object_data(*args, **kwargs)

stream_changes = False

class pymoa_remote.server.RemoteRegistry(**kwargs)

Bases: InstanceRegistry

Server side object registry.

create_instance(cls_triple: Tuple[str, str, str], hash_name: str, args: tuple, kwargs: dict, config: dict) → Any

delete_instance(hash_name: str)

get_instance(hash_name: str)

class pymoa_remote.server.SimpleExecutorServer(registry: RemoteRegistry = None, executor: ExecutorBase = None, **kwargs)

Bases: ExecutorServer

await delete_instance(data: dict) → None

await ensure_instance(data: dict) → bool

await execute(data: dict)

async for ... in execute_generator(data: dict)

await get_echo_clock(data: dict)

await get_object_config(data: dict)

await get_object_data(data: dict)

await get_objects(data: dict)

await register_remote_class(data: dict)

await remote_import(data: dict)

await sleep(data: dict)

await start_logging_object_data(data: dict, log_callback)

await stop_logging_object_data(binding)

pymoa_remote.server.dispatch_stream_channel_to_queues(data: dict, channel: str, hash_name: str, stream_clients: Dict[Tuple[str, str], Dict[Any, MaxSizeErrorDeque]], encode: Callable)

REST SSE Client

class pymoa_remote.rest.sse.SSEStream

Bases: object

Generator that takes a asks streaming response object and decodes and yields individual sse events.

classmethod finish_sse_event(event: dict) -> (<class 'dict'>, <class 'tuple'>)

classmethod set_sse_field(event: dict, buffer: list) → list

classmethod async for ... in stream(response_body) → AsyncGenerator[Tuple[str | None, str | None, str | None, int | None], None]