from abc import abstractmethod
from logging import Logger
from typing import TYPE_CHECKING, Any, ClassVar, Mapping, Optional, Protocol, runtime_checkable
from typing_extensions import Self
from viam.operations import Operation
from viam.proto.common import ResourceName
from .types import Subtype
if TYPE_CHECKING:
from viam.robot.client import RobotClient
from viam.utils import ValueTypes
[docs]@runtime_checkable
class ResourceBase(Protocol):
"""
The base requirements for a Resource.
"""
SUBTYPE: ClassVar["Subtype"]
"""The Subtype of the Resource"""
name: str
"""The name of the Resource"""
logger: Logger
"""A logger allowing for setting log levels on a per-resource basis"""
[docs] @classmethod
def get_resource_name(cls, name: str) -> ResourceName:
"""
Get the ResourceName for this Resource with the given name
::
# Can be used with any resource, using an arm as an example
my_arm_name = Arm.get_resource_name("my_arm")
Args:
name (str): The name of the Resource
Returns:
ResourceName: The ResourceName of this Resource
"""
return ResourceName(
namespace=cls.SUBTYPE.namespace,
type=cls.SUBTYPE.resource_type,
subtype=cls.SUBTYPE.resource_subtype,
name=name,
)
[docs] @classmethod
@abstractmethod
def from_robot(cls, robot: "RobotClient", name: str) -> Self:
"""Get the Resource named ``name`` from the provided robot.
::
# Can be used with any resource, using an arm as an example
my_arm = Arm.from_robot(robot, "my_arm")
Args:
robot (RobotClient): The robot
name (str): The name of the Resource
Returns:
Self: The Resource, if it exists on the robot
"""
...
[docs] @abstractmethod
async def do_command(
self, command: Mapping[str, "ValueTypes"], *, timeout: Optional[float] = None, **kwargs
) -> Mapping[str, "ValueTypes"]:
"""Send/Receive arbitrary commands to the Resource
::
command = {"cmd": "test", "data1": 500}
result = await component.do_command(command)
Args:
command (Mapping[str, ValueTypes]): The command to execute
Raises:
NotImplementedError: Raised if the Resource does not support arbitrary commands
Returns:
Mapping[str, ValueTypes]: Result of the executed command
"""
...
[docs] def get_operation(self, kwargs: Mapping[str, Any]) -> Operation:
"""Get the ``Operation`` associated with the currently running function.
When writing custom resources, you should get the ``Operation`` by calling this function and check to see if it's cancelled.
If the ``Operation`` is cancelled, then you can perform any necessary (terminating long running tasks, cleaning up connections, etc.
).
Args:
kwargs (Mapping[str, Any]): The kwargs object containing the operation
Returns:
viam.operations.Operation: The operation associated with this function
"""
return kwargs.get(Operation.ARG_NAME, Operation._noop())
[docs] async def close(self):
"""Safely shut down the resource and prevent further use.
Close must be idempotent. Later configuration may allow a resource to be "open" again.
If a resource does not want or need a close function, it is assumed that the resource does not need to return errors when future
non-Close methods are called.
::
await component.close()
"""
return