viam.utils

Attributes

ValueTypes

Types that can be encoded into a protobuf Value

SensorReading

Types that can be returned from a sensor

Classes

PointerCounter

Functions

primitive_to_value(→ google.protobuf.struct_pb2.Value)

Create a new google.protobuf.struct_pb2.Value

value_to_primitive(→ ValueTypes)

resource_names_for_resource(...)

message_to_struct(→ google.protobuf.struct_pb2.Struct)

struct_to_message(→ _T)

dict_to_struct(→ google.protobuf.struct_pb2.Struct)

struct_to_dict(→ Dict[str, ValueTypes])

datetime_to_timestamp(...)

get_geometries() → List[viam.proto.common.Geometry])

sensor_readings_native_to_value(→ Mapping[str, ...)

sensor_readings_value_to_native(→ Mapping[str, ...)

to_thread(→ _R)

Asynchronously run a function in a separate thread.

from_dm_from_extra(→ bool)

Used in modular filter components to get the 'fromDataManagement' value from an extra map.

create_filter(→ viam.proto.app.data.Filter)

Create a Filter.

Module Contents

viam.utils.ValueTypes

Types that can be encoded into a protobuf Value

viam.utils.SensorReading

Types that can be returned from a sensor

viam.utils.primitive_to_value(v: ValueTypes) → google.protobuf.struct_pb2.Value[source]

Create a new google.protobuf.struct_pb2.Value Supports primitive types of - None - Number - Bool - String - Dict - List - Bytes

Parameters:

v (ValueTypes) – object to convert to a Value

Raises:

TypeError – If the object cannot be converted

Returns:

a protobuf Value instance of the object

Return type:

Value

viam.utils.value_to_primitive(value: google.protobuf.struct_pb2.Value) → ValueTypes[source]
viam.utils.resource_names_for_resource(resource: viam.resource.base.ResourceBase) → List[viam.proto.common.ResourceName][source]
viam.utils.message_to_struct(message: google.protobuf.message.Message) → google.protobuf.struct_pb2.Struct[source]
viam.utils.struct_to_message(struct: google.protobuf.struct_pb2.Struct, message_type: Type[_T]) → _T[source]
viam.utils.dict_to_struct(obj: Mapping[str, ValueTypes] | None) → google.protobuf.struct_pb2.Struct[source]
viam.utils.struct_to_dict(struct: google.protobuf.struct_pb2.Struct) → Dict[str, ValueTypes][source]
viam.utils.datetime_to_timestamp(dt: datetime.datetime | None) → google.protobuf.timestamp_pb2.Timestamp | None[source]
async viam.utils.get_geometries(client: viam.resource.types.SupportsGetGeometries, name: str, extra: Dict[str, Any] | None = None, timeout: float | None = None, metadata: viam.resource.rpc_client_base.ResourceRPCClientBase.Metadata = ResourceRPCClientBase.Metadata()) → List[viam.proto.common.Geometry][source]
viam.utils.sensor_readings_native_to_value(readings: Mapping[str, Any]) → Mapping[str, google.protobuf.struct_pb2.Value][source]
viam.utils.sensor_readings_value_to_native(readings: Mapping[str, google.protobuf.struct_pb2.Value]) → Mapping[str, SensorReading][source]
class viam.utils.PointerCounter[source]
increment() → int[source]
decrement() → int[source]
async wait() → None[source]
property count: int
async viam.utils.to_thread(func: collections.abc.Callable[_P, _R], *args: _P, **kwargs: _P) → _R[source]

Asynchronously run a function in a separate thread.

This is a copy of the function defined in the python source, which is only available in python >= 3.9.

See: https://github.com/python/cpython/blob/main/Lib/asyncio/threads.py

viam.utils.from_dm_from_extra(extra: Dict[str, Any] | None) → bool[source]

Used in modular filter components to get the ‘fromDataManagement’ value from an extra map.

viam.utils.create_filter(component_name: str | None = None, component_type: str | None = None, method: str | None = None, robot_name: str | None = None, robot_id: str | None = None, part_name: str | None = None, part_id: str | None = None, location_ids: List[str] | None = None, organization_ids: List[str] | None = None, mime_type: List[str] | None = None, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, tags: List[str] | None = None, bbox_labels: List[str] | None = None, dataset_id: str | None = None) → viam.proto.app.data.Filter[source]

Create a Filter.

Parameters:

  • component_name (Optional[str]) – Optional name of the component that captured the data being filtered (for example, “left_motor”).

  • component_type (Optional[str]) – Optional type of the componenet that captured the data being filtered (for example, “motor”).

  • method (Optional[str]) – Optional name of the method used to capture the data being filtered (for example, “IsPowered”).

  • robot_name (Optional[str]) – Optional name of the robot associated with the data being filtered (for example, “viam_rover_1”).

  • robot_id (Optional[str]) – Optional ID of the robot associated with the data being filtered.

  • part_name (Optional[str]) – Optional name of the system part associated with the data being filtered (for example, “viam_rover_1-main”).

  • part_id (Optional[str]) – Optional ID of the system part associated with the data being filtered.

  • location_ids (Optional[List[str]]) – Optional list of location IDs associated with the data being filtered.

  • organization_ids (Optional[List[str]]) – Optional list of organization IDs associated with the data being filtered.

  • mime_type (Optional[List[str]]) – Optional mime type of data being filtered (for example, “image/png”).

  • start_time (Optional[datetime.datetime]) – Optional start time of an interval to filter data by.

  • end_time (Optional[datetime.datetime]) – Optional end time of an interval to filter data by.

  • tags (Optional[List[str]]) – Optional list of tags attached to the data being filtered (for example, [“test”]).

  • bbox_labels (Optional[List[str]]) – Optional list of bounding box labels attached to the data being filtered (for example, [“square”, “circle”]).

  • dataset_id (Optional[str]) – Optional ID of dataset associated with data being filtered

Returns:

The Filter object.

Return type:

viam.proto.app.data.Filter