Skip to content

Bases

hololinked.core.actions.Action

Object that models an action. These actions are unbound and return a bound action when accessed using the owning object.

Source code in hololinked/hololinked/core/actions.py
class Action:
    """
    Object that models an action.
    These actions are unbound and return a bound action when accessed using the owning object.
    """

    __slots__ = ["obj", "owner", "_execution_info"]

    def __init__(self, obj: MethodType) -> None:
        """
        Parameters
        ----------
        obj: MethodType
            the method that is being wrapped as an action
        """
        self.obj = obj

    def __set_name__(self, owner, name):
        self.owner = owner

    def __str__(self) -> str:
        return f"<Action({self.owner.__name__}.{self.obj.__name__})>"

    def __eq__(self, other) -> bool:
        if not isinstance(other, Action):
            return False
        return self.obj == other.obj

    def __hash__(self) -> int:
        return hash(self.obj)

    def __get__(self, instance, owner):
        if instance is None and not self._execution_info.isclassmethod:
            return self
        if self._execution_info.iscoroutine:
            return BoundAsyncAction(self.obj, self, instance, owner)
        return BoundSyncAction(self.obj, self, instance, owner)

    def __call__(self, *args, **kwargs):
        raise NotImplementedError(
            f"Cannot invoke unbound action {self.name} of {self.owner.__name__}."
            + " Bound methods must be called, not the action itself. Use the appropriate instance to call the method."
        )

    @property
    def name(self) -> str:
        """name of the action"""
        return self.obj.__name__

    @property
    def execution_info(self) -> ActionInfoValidator:
        """
        internal dataclass that holds all information about the action

        TODO: this can be refactored
        """
        return self._execution_info

    @execution_info.setter
    def execution_info(self, value: ActionInfoValidator) -> None:
        if not isinstance(value, ActionInfoValidator):
            raise TypeError("execution_info must be of type ActionInfoValidator")
        self._execution_info = value  # type: ActionInfoValidator

    def to_affordance(self, owner_inst=None):
        """
        Generates a `ActionAffordance` TD fragment for this Action.

        Parameters
        ----------
        owner_inst: Thing, optional
            The instance of the owning `Thing` object. If not supplied, the class is used.

        Returns
        -------
        ActionAffordance
            the affordance TD fragment for this action
        """
        from ..td import ActionAffordance

        return ActionAffordance.generate(self, owner_inst or self.owner)

Functions

__init__

__init__(obj: MethodType) -> None

Parameters:

Name Type Description Default
obj
MethodType

the method that is being wrapped as an action

required
Source code in hololinked/hololinked/core/actions.py
def __init__(self, obj: MethodType) -> None:
    """
    Parameters
    ----------
    obj: MethodType
        the method that is being wrapped as an action
    """
    self.obj = obj

to_affordance

to_affordance(owner_inst=None)

Generates a ActionAffordance TD fragment for this Action.

Parameters:

Name Type Description Default
owner_inst

The instance of the owning Thing object. If not supplied, the class is used.

None

Returns:

Type Description
ActionAffordance

the affordance TD fragment for this action

Source code in hololinked/hololinked/core/actions.py
def to_affordance(self, owner_inst=None):
    """
    Generates a `ActionAffordance` TD fragment for this Action.

    Parameters
    ----------
    owner_inst: Thing, optional
        The instance of the owning `Thing` object. If not supplied, the class is used.

    Returns
    -------
    ActionAffordance
        the affordance TD fragment for this action
    """
    from ..td import ActionAffordance

    return ActionAffordance.generate(self, owner_inst or self.owner)

hololinked.core.actions.BoundAction

A bound action - base class for both sync and async methods.

Source code in hololinked/hololinked/core/actions.py
class BoundAction:
    """
    A bound action - base class for both sync and async methods.
    """

    __slots__ = [
        "obj",
        "execution_info",
        "descriptor",
        "owner_inst",
        "owner",
        "bound_obj",
    ]

    def __init__(self, obj: FunctionType, descriptor: Action, owner_inst, owner) -> None:
        self.obj = obj
        self.descriptor = descriptor
        self.execution_info = descriptor._execution_info
        self.owner = owner
        self.owner_inst = owner_inst
        self.bound_obj = owner if self.execution_info.isclassmethod else owner_inst

    def __post_init__(self):
        # never called, neither possible to call, only type hinting
        from .thing import Thing, ThingMeta

        # owner class and instance
        self.owner: ThingMeta
        self.owner_inst: Thing
        self.obj: FunctionType
        # the validator that was used to accept user inputs to this action.
        # stored only for reference, hardly used.
        self._execution_info: ActionInfoValidator

    def validate_call(self, args, kwargs: dict[str, Any]) -> None:
        """
        Validate the call to the action, like payload, state machine state etc.
        Errors are raised as exceptions.

        Parameters
        ----------
        args: tuple
            positional arguments to the action
        kwargs: dict
            keyword arguments to the action
        """
        if self.execution_info.isparameterized and len(args) > 0:
            raise RuntimeError("parameterized functions cannot have positional arguments")
        if self.owner_inst is None:
            return
        if self.execution_info.state is None or (
            hasattr(self.owner_inst, "state_machine")
            and self.owner_inst.state_machine.current_state in self.execution_info.state
        ):
            if self.execution_info.schema_validator is not None:
                self.execution_info.schema_validator.validate_method_call(args, kwargs)
        else:
            raise StateMachineError(
                "Thing '{}' is in '{}' state, however action can be executed only in '{}' state".format(
                    self.owner_inst,
                    self.owner_inst.state,
                    self.execution_info.state,
                )
            )

    @property
    def name(self) -> str:
        """name of the action"""
        return self.obj.__name__

    def __call__(self, *args, **kwargs):
        raise NotImplementedError("call must be implemented by subclass")

    def external_call(self, *args, **kwargs):
        """validated call to the action with state machine and payload checks"""
        raise NotImplementedError("external_call must be implemented by subclass")

    def __str__(self):
        return f"<BoundAction({self.owner.__name__}.{self.obj.__name__} of {self.owner_inst.id})>"

    def __eq__(self, value):
        if not isinstance(value, BoundAction):
            return False
        return self.obj == value.obj

    def __hash__(self):
        return hash(str(self))

    def __getattribute__(self, name):
        # https://docs.python.org/3/howto/descriptor.html#functions-and-methods
        if name == "__doc__":
            return self.obj.__doc__
        return super().__getattribute__(name)

    def to_affordance(self):
        """
        Generates a `ActionAffordance` TD fragment for this Action.

        Parameters
        ----------
        owner_inst: Thing, optional
            The instance of the owning `Thing` object. If not supplied, the class is used.

        Returns
        -------
        ActionAffordance
            the affordance TD fragment for this action
        """
        return Action.to_affordance(self.descriptor, self.owner_inst or self.owner)

Functions

__init__

__init__(obj: FunctionType, descriptor: Action, owner_inst, owner) -> None
Source code in hololinked/hololinked/core/actions.py
def __init__(self, obj: FunctionType, descriptor: Action, owner_inst, owner) -> None:
    self.obj = obj
    self.descriptor = descriptor
    self.execution_info = descriptor._execution_info
    self.owner = owner
    self.owner_inst = owner_inst
    self.bound_obj = owner if self.execution_info.isclassmethod else owner_inst

validate_call

validate_call(args, kwargs: dict[str, Any]) -> None

Validate the call to the action, like payload, state machine state etc. Errors are raised as exceptions.

Parameters:

Name Type Description Default
args

positional arguments to the action

required
kwargs
dict[str, Any]

keyword arguments to the action

required
Source code in hololinked/hololinked/core/actions.py
def validate_call(self, args, kwargs: dict[str, Any]) -> None:
    """
    Validate the call to the action, like payload, state machine state etc.
    Errors are raised as exceptions.

    Parameters
    ----------
    args: tuple
        positional arguments to the action
    kwargs: dict
        keyword arguments to the action
    """
    if self.execution_info.isparameterized and len(args) > 0:
        raise RuntimeError("parameterized functions cannot have positional arguments")
    if self.owner_inst is None:
        return
    if self.execution_info.state is None or (
        hasattr(self.owner_inst, "state_machine")
        and self.owner_inst.state_machine.current_state in self.execution_info.state
    ):
        if self.execution_info.schema_validator is not None:
            self.execution_info.schema_validator.validate_method_call(args, kwargs)
    else:
        raise StateMachineError(
            "Thing '{}' is in '{}' state, however action can be executed only in '{}' state".format(
                self.owner_inst,
                self.owner_inst.state,
                self.execution_info.state,
            )
        )

external_call

external_call(*args, **kwargs)

validated call to the action with state machine and payload checks

Source code in hololinked/hololinked/core/actions.py
def external_call(self, *args, **kwargs):
    """validated call to the action with state machine and payload checks"""
    raise NotImplementedError("external_call must be implemented by subclass")

hololinked.core.actions.BoundSyncAction

Bases: BoundAction

non async(io) action call. The call is passed to the method as-it-is to allow local invocation without state machine checks. Use external_call to have validation.

Source code in hololinked/hololinked/core/actions.py
class BoundSyncAction(BoundAction):
    """
    non async(io) action call. The call is passed to the method as-it-is to allow local
    invocation without state machine checks. Use `external_call` to have validation.
    """

    def external_call(self, *args, **kwargs):
        """validated call to the action with state machine and payload checks"""
        self.validate_call(args, kwargs)
        return self.__call__(*args, **kwargs)

    def __call__(self, *args, **kwargs):
        if self.execution_info.isclassmethod:
            return self.obj(*args, **kwargs)
        return self.obj(self.bound_obj, *args, **kwargs)

Functions

external_call

external_call(*args, **kwargs)

validated call to the action with state machine and payload checks

Source code in hololinked/hololinked/core/actions.py
def external_call(self, *args, **kwargs):
    """validated call to the action with state machine and payload checks"""
    self.validate_call(args, kwargs)
    return self.__call__(*args, **kwargs)

hololinked.core.actions.BoundAsyncAction

Bases: BoundAction

async(io) action call. The call is passed to the method as-it-is to allow local invocation without state machine checks. Use external_call to have validation.

Source code in hololinked/hololinked/core/actions.py
class BoundAsyncAction(BoundAction):
    """
    async(io) action call. The call is passed to the method as-it-is to allow local
    invocation without state machine checks. Use `external_call` to have validation.
    """

    async def external_call(self, *args, **kwargs):
        """validated call to the action with state machine and payload checks"""
        self.validate_call(args, kwargs)
        return await self.__call__(*args, **kwargs)

    async def __call__(self, *args, **kwargs):
        if self.execution_info.isclassmethod:
            return await self.obj(*args, **kwargs)
        return await self.obj(self.bound_obj, *args, **kwargs)

Functions

external_call async

external_call(*args, **kwargs)

validated call to the action with state machine and payload checks

Source code in hololinked/hololinked/core/actions.py
async def external_call(self, *args, **kwargs):
    """validated call to the action with state machine and payload checks"""
    self.validate_call(args, kwargs)
    return await self.__call__(*args, **kwargs)