Test Package Imports#

This temporary page exists to make sure the packages we painstakingly install, with a variety of conflict-resolution tricks, still import, even if we pull no docs from them elsewhere yet.

adapter_common#

Asset API Test#

class AssetTreeValue(
asset: artificial.adapter_common.apis.interfaces.Asset,
parent: artificial.adapter_common.apis.interfaces.Asset | NoneType,
depth: int,
)#
class AssetsImpl(
labmanChannel: Channel,
scope: Scope,
)#
async create(
*assets: Asset,
) List[str]#

Create all given Assets. Empty IDs will be randomly created, otherwise given IDs will be used. Fails if any given ID already exists or if definition ID is not set. Returns the IDs of all created Assets.

async delete(*ids: str) List[str]#

Delete all Assets for given IDs. IDs that don’t exist are ignored. Returns all Asset IDs that were actually deleted.

async get(
*ids: str,
) List[Asset]#

Returns all Assets for given IDs. Fails if any ID does not exist.

async query(
match: Match,
) List[Asset]#

Returns any Assets matching all given criteria. For properties, Assets must have all given query keys and exactly match given property values. A given property value of “None” matches all values.

async update(
*updates: Update,
) None#

Applies all given updates to Assets. Fails if any Asset does not exist.

async traverse_asset_tree(
aapi: AssetsApi,
root_id: str = '',
) Iterator[AssetTreeValue]#

Iterate through all assets of a scene graph, depth first. The iterator provides asset, its parent and nesting depth. If root_id is given then only the sub tree from given root is traversed.

Health Check API#

class HealthImpl(
labman_channel: Channel,
scope: Scope,
)#
class Monitor(
parent: 'HealthImpl',
id: 'str',
cb: 't.Callable[[],
t.Coroutine[None,
None,
Health.Message]]',
interval: 'float',
grace: 'float',
last_status: 'health_pb.HealthMessage' = <factory>,
)#
async add_monitor(
id: str,
callback: Callable[[], Coroutine[None, None, HealthMessage]],
interval_s: float = 10.0,
grace_s: float = 2.0,
) None#

Sets a health status monitor for a given id. The given callback will be regularly called and is expected to return within the configured grace period, otherwise the invocation is canceled. If the callback does not return in time or raises an error the state is considered to be in error.

async remove_monitor(id: str) None#

Removes the monitor with given id. Its callback will not be called anymore and its status does not contribute to the overall system status anymore. This can also be used to remove the remote health state, even if the monitor has not been added locally.

Interfaces and Data Types#

class Asset(*args, **kwargs)#
class Match(*args, **kwargs)#
class Update(*args, **kwargs)#
apply(update: Update) bool#

Applies an update to self, returning True if any field or property was changed.

copy() Asset#

Returns a clone of this Asset, copying all fields and properties into a different object.

delta(
to: Asset,
strict_properties: bool = True,
) Update#

Returns an Asset.Update that contains the changes that would need to be applied to this asset to make it to be equal to given other Asset. If strict_properties is False, then the Update leave any extra properties of self, otherwise, extra properties are removed.

exception AssetFetchError(asset_id: str)#

Used to indicate a failure to fetch the asset

class AssetsApi(*args, **kwargs)#
abstract async create(
*assets: Asset,
) List[str]#

Create all given Assets. Empty IDs will be randomly created, otherwise given IDs will be used. Fails if any given ID already exists or if definition ID is not set. Returns the IDs of all created Assets.

abstract async delete(*ids: str) List[str]#

Delete all Assets for given IDs. IDs that don’t exist are ignored. Returns all Asset IDs that were actually deleted.

abstract async get(
*ids: str,
) List[Asset]#

Returns all Assets for given IDs. Fails if any ID does not exist.

abstract async query(
match: Match,
) List[Asset]#

Returns any Assets matching all given criteria. For properties, Assets must have all given query keys and exactly match given property values. A given property value of “None” matches all values.

abstract async update(
*updates: Update,
) None#

Applies all given updates to Assets. Fails if any Asset does not exist.

class ExperimentalApi(*args, **kwargs)#
abstract async run_assistant(
job_id: str,
assistant: str,
args: Dict[str, Any] = {},
) Any#

Run an assistant, blocks until the assistant is finished. The assistant must have been already included in the job via its workflow.

class Health(*args, **kwargs)#
Message#

alias of HealthMessage

abstract async add_monitor(
id: str,
callback: Callable[[], Coroutine[None, None, HealthMessage]],
interval_s: float = 10.0,
grace_s: float = 2.0,
) None#

Sets a health status monitor for a given id. The given callback will be regularly called and is expected to return within the configured grace period, otherwise the invocation is canceled. If the callback does not return in time or raises an error the state is considered to be in error.

abstract async remove_monitor(id: str) None#

Removes the monitor with given id. Its callback will not be called anymore and its status does not contribute to the overall system status anymore. This can also be used to remove the remote health state, even if the monitor has not been added locally.

class Job(
display_name: 'str',
id: 'str',
parent_id: 'str',
action_id: 'str',
lab_id: 'str',
properties: 't.Dict[str,
t.Union[None,
str,
int,
float,
bool,
bytes,
Message]]' = <factory>,
)#
class Match(
display_name: 't.Optional[str]' = None,
parent_id: 't.Optional[str]' = None,
action_id: 't.Optional[str]' = None,
properties: 't.Dict[str,
t.Union[None,
str,
int,
float,
bool,
bytes,
Message]]' = <factory>,
)#
class JobsApi(*args, **kwargs)#
class LabApi(*args, **kwargs)#
abstract property assets: AssetsApi#

API for reading and modifying Assets of a Lab.

abstract property health: Health#

API for providing health status of Adapter components.

abstract property sockets: SocketsApi#

API for managing asset location and sockets.

exception PlugDimensionError(x: float | None, y: float | None, z: float | None)#

Used to indicate plug asset is missing vital dimension information

exception SocketChildrenFullError(num_children: int = 0, max_children: int = 0)#

Used to indicate socket has no more room for children plugs

exception SocketError#

Used to indicate socketing failed

exception SocketFormatError(plug_type: List[str] = [], socket_types: str = '')#

Used to indicate socket and plug are not compatible formats

exception SocketFullError(occupant_id: str = '', occupant_name: str = '')#

Used to indicate socket is full, plug doesn’t fit

exception SocketHasNoChildrenError#

Used to indicate socket has no valid children assets

exception SocketHeightError(asset_height: float = 0, socket_height: float = 0)#

Used to indicate socket is not tall enough for the plug

exception SocketIndexError#

Used to indicate definitional socket does not exist/is not valid

exception SocketStackFullError(stack_height: float = 0, total_stack_height: float = 0)#

Used to indicate stacking socket has no more room

class SocketsApi(*args, **kwargs)#
abstract async get_all_sbs_socket_addresses(*, socket_asset_id: str) List[int]#

Retrieves the indices of all sockets on an asset that conform with SBS standard

Parameters:

socket_asset_id – Instance ID of the asset

Returns:

List of integers that correspond to the socket indices of the SBS sockets on the asset

Raises:

AssetFetchError – If fails to fetch asset

abstract async socket_sbs_asset(
*,
plug_asset_id: str,
socket_asset_id: str,
index: int,
) None#

Sockets plug_asset_id into socket_asset_id, throws a subclass of SocketError if operation can’t be completed

Parameters:
  • plug_asset_id – Instance ID of asset being socketed into another asset

  • socket_asset_id – Instance ID of the asset receiving the plug

  • index – Socket index to plug asset into

Raises:
abstract async unsocket_all_sbs_assets(
*,
socket_asset_id: str,
index: int,
) List[str]#

Unplugs all assets from a socket, if there are multiple it unplugs from top down, similar to a pop.

Parameters:
  • socket_asset_id – Instance ID of the asset receiving the plug

  • index – Socket index to unplug assets from

Returns:

List of strings that correspond to asset ID’s of what was unplugged.

Raises:

AssetFetchError – If fails to fetch asset

abstract async unsocket_sbs_asset(
*,
socket_asset_id: str,
index: int,
num_items: int = 1,
) List[str]#

Unplugs assets from a socket, if there are multiple it unplugs from top down, similar to a pop.

Parameters:
  • socket_asset_id – Instance ID of the asset receiving the plug

  • index – Socket index to unplug assets from

  • num_items – How many things to unplug from socket, defaults to 1

Returns:

List of strings that correspond to asset ID’s of what was unplugged. Order is first asset unplugged (top of stack) to last asset unplugged

Raises:

AssetFetchError – If fails to fetch asset