import inspect
from typing import Any
from collections.abc import Callable
from ppc_robot_lib.steps import AbstractStep
from ppc_robot_lib.tasks import TaskContextInterface
[docs]
class ContextUpdateStep(AbstractStep):
"""
Updates the current :py:class:`ppc_robot_lib.work_set.WorkSet` context. Accepts either an dictionary
with keys to update, or a callable.
When callable is given, it is called with the current context as an argument. If the callable has an argument named
``task_ctx``, whole task context is passed as the second named argument. If the callable returns an dictionary
that is not the current context, it is applied to the current context.
**Example:**
>>> from ppc_robot_lib.steps.control import ContextUpdateStep
>>> ContextUpdateStep({'counter': 0})
>>> def context_counter_inc(context):
... context['counter'] += 1
... return {'new_key': True}
>>> ContextUpdateStep(context_counter_inc)
The final value of context will be equal to::
{'counter': 1, 'new_key': True}
"""
def __init__(self, data: dict[str, Any] | Callable[[dict[str, Any]], dict[str, Any] | None]):
"""
:param data: Dictionary with keys to update or a callable that will update the context.
"""
self.data = data
def execute(self, task_ctx: TaskContextInterface) -> None:
if callable(self.data):
signature = inspect.signature(self.data)
if 'task_ctx' in signature.parameters:
return_value = self.data(task_ctx.work_set.context, task_ctx=task_ctx)
else:
return_value = self.data(task_ctx.work_set.context)
if isinstance(return_value, dict) and return_value is not task_ctx.work_set.context:
task_ctx.work_set.context.update(return_value)
else:
task_ctx.work_set.context.update(self.data)