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( health_stub: HealthServiceStub, 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 AssetSyncApiV1(*args, **kwargs)#

abstract async add( local_id: Alias, asset: Asset, ) → None#: Adds a new asset to the syncer. The asset will be created in alab when the syncer is synchronized.

abstract alias_of( remote_id: Id, ) → Alias | None#: Returns the alias of a remote asset ID. This can be used to refer to the asset in the syncer.

abstract batch_sync( refresh_cache: bool = False, ) → AbstractAsyncContextManager[Context]#: Synchronize assets. With the returned context, multiple changes can be batched together. All changes are only executed to alab when the context returns normally. You can abort the synchronization by raising an exception. If refresh_cache is True, then the state of all synchronized remote assets is fetched first. For performance, it is not recommended to do this often, use False instead for frequent, small changes.

abstract id_of( local_id: Alias, ) → Id | None#: Returns the remote asset ID of an alias. This can be used to refer to the asset in alab.

abstract async link_assets( device_map: Mapping[str, DeviceConfig], strict: bool = True, ) → None#: Links existing remote assets to this syncer. These assets are considered fixed and won’t be removed but they will be part of the local cache and can be updated. This is typically used for devices of a pre-existing scene. If strict is True then any other asset links that are already in the scene but not in the given device_map will be removed.

abstract async remove( asset_id: Id | Alias, ) → None#: Removes an asset from the syncer. The asset will be deleted in alab when the syncer is synchronized.

abstract async reset(clear_fixed: bool = False) → None#: Removes any assets created from this syncer, if clear_fixed is true, also unlinks any fixed assets.

abstract async seize(from_sync_id: str) → None#: Take control of assets managed by another synchronizer ID. Use this if you changed the synchronizer ID.

abstract async update( asset_id: Id | Alias, update: Update, ) → None#: Updates an asset in the syncer. The asset will be updated in alab when the syncer is synchronized.

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:

SocketFormatError – If socket and plug types are not compatible
SocketIndexError – If socket does not exist at index
PlugDimensionError – If plug does not have the correct dimensions
SocketFullError – If socket has no room for plug
SocketHasNoChildrenError – If socket has no valid children assets
AssetFetchError – If fails to fetch asset

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