nanover.core.nanover_client module

A module for setting up typical NanoVer clients, containing a client that sets up a command service.

class nanover.core.nanover_client.NanoverClient(*, channel: Channel, make_channel_owner: bool = False)

Bases: GrpcClient

A base gRPC client for NanoVer services. Automatically sets up a stub for the CommandServicer, enabling the running of arbitrary commands.

attempt_update_locks(lock_updates: Mapping[str, float | None]) → bool

Attempt to acquire and/or free a number of locks on the shared state.

Parameters:

lock_updates – A dictionary of keys to either a duration in seconds to attempt to acquire or renew a lock, or None to indicate the lock should be released if held.

Returns:

True if the desired locks were acquired, and False otherwise.

attempt_update_state(change: DictionaryChange) → bool

Attempt to make a single atomic change to the shared state, blocking until a response is received.

Parameters:

change – A single change to make to the shared state that will either be made in full, or ignored if some of the keys are locked by another user.

Returns:

True if the server accepted our change, and False otherwise.

property available_commands: Dict[str, CommandInfo]

Returns a copy of the dictionary of commands available on the server, as determined by previously calling update_available_commands().

Returns:

A dictionary of command information, keyed by the command names.

close()

Shutdown all threads and close the underlying channel if the client has been given that responsibility.

copy_state() → Dict[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any] | Iterable[None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]] | Mapping[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]]]

Return a deep copy of the current state.

lock_state() → ContextManager[Dict[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any] | Iterable[None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]] | Mapping[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]]]]

Context manager that locks and returns the state. Any attempted state updates are delayed until the context is exited.

run_command(name: str, **arguments) → Dict[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any] | Iterable[None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]] | Mapping[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]]]

Runs a command on the command server.

Parameters:

• name – Name of command to run.

• arguments – Arguments to provide to command.

Returns:

Dictionary of results, which may be empty.

subscribe_all_state_updates(interval=0.03333333333333333)

Subscribe, in the background, to any updates made to the shared state.

Parameters:

interval – Minimum time (in seconds) between receiving new updates for any and all values.

update_available_commands() → Dict[str, CommandInfo]

Gets all the commands on the command server, and updates this client’s known commands. Blocks until the dictionary of available commands is received.

Returns:

A dictionary of all the commands on the command server, keyed by name

class nanover.core.nanover_client.NanoverStubClient(*, channel: Channel, stub, make_channel_owner: bool = False)

Bases: NanoverClient

A base gRPC client for NanoVer services. Automatically sets up a stub for the CommandServicer, and attaches the provided stub to the underlying gRPC channel.

Parameters:

stub – gRPC stub to attach.