pycram.language =============== .. py:module:: pycram.language Classes ------- .. autoapisummary:: pycram.language.LanguageMixin pycram.language.LanguagePlan pycram.language.SequentialPlan pycram.language.ParallelPlan pycram.language.TryInOrderPlan pycram.language.TryAllPLan pycram.language.RepeatPlan pycram.language.MonitorPlan pycram.language.CodePlan pycram.language.LanguageNode pycram.language.SequentialNode pycram.language.ParallelNode pycram.language.RepeatNode pycram.language.MonitorNode pycram.language.TryInOrderNode pycram.language.TryAllNode pycram.language.CodeNode Module Contents --------------- .. py:class:: LanguageMixin Parent class for language expressions. Implements the operators as well as methods to reduce the resulting language tree. .. py:method:: add_language_plan(child_plans: typing_extensions.Iterable[pycram.plan.Plan], root_node: LanguageNode) -> pycram.plan.Plan .. py:method:: __add__(other: pycram.plan.Plan) -> pycram.plan.Plan Language expression for sequential execution. :param other: Another Language expression, either a designator_description or language expression :return: A :func:`~Sequential` object which is the new root node of the language tree .. py:method:: __sub__(other: pycram.plan.Plan) -> pycram.plan.Plan Language expression for try in order. :param other: Another Language expression, either a designator_description or language expression :return: A :func:`~TryInOrder` object which is the new root node of the language tree .. py:method:: __or__(other: pycram.plan.Plan) -> pycram.plan.Plan Language expression for parallel execution. :param other: Another Language expression, either a designator_description or language expression :return: A :func:`~Parallel` object which is the new root node of the language tree .. py:method:: __xor__(other: pycram.plan.Plan) -> pycram.plan.Plan Language expression for try all execution. :param other: Another Language expression, either a designator_description or language expression :return: A :func:`~TryAll` object which is the new root node of the language tree .. py:method:: __rshift__(other: typing_extensions.Callable) Operator for Monitors, this always makes the Monitor the parent of the other expression. :param other: Another Language expression :return: The Monitor which is now the new root node. .. py:method:: __mul__(other: int) Language expression for Repeated execution. The other attribute of this operator has to be an integer. :param other: An integer which states how often the Language expression should be repeated :return: A :func:`~Repeat` object which is the new root node of the language tree .. py:method:: __rmul__(other: int) Language expression for Repeated execution. The other attribute of this operator has to be an integer. This is the reversed operator of __mul__ which allows to write: .. code-block:: python 2 * ParkAction() :param other: An integer which states how often the Language expression should be repeated :return: A :func:`~Repeat` object which is the new root node of the language tree .. py:method:: interrupt() -> None :abstractmethod: Base method for interrupting the execution of Language expression. To be overwritten in a sub-class. .. py:class:: LanguagePlan(root: LanguageNode, *children: pycram.plan.Plan) Bases: :py:obj:`pycram.plan.Plan` Base class for language plans .. py:method:: simplify_language_nodes() Traverses the plan and merges LanguageNodes of the same type .. py:class:: SequentialPlan(*children: pycram.plan.Plan) Bases: :py:obj:`LanguagePlan` Creates a plan which executes its children in sequential order .. py:class:: ParallelPlan(*children: pycram.plan.Plan, root: LanguageNode = None) Bases: :py:obj:`LanguagePlan` Creates a plan which executes all children in parallel in seperate threads .. py:attribute:: parallel_blocklist :value: ['PickUpAction', 'PlaceAction', 'OpenAction', 'CloseAction', 'TransportAction', 'GraspingAction'] A list of Actions which can't be part of a Parallel plan .. py:class:: TryInOrderPlan(*children: pycram.plan.Plan) Bases: :py:obj:`LanguagePlan` Creates a plan that executes all children in sequential order but does not stop if one of them throws an error .. py:class:: TryAllPLan(*children: pycram.plan.Plan) Bases: :py:obj:`ParallelPlan` Creates a plan which executes all children in parallel but does not abort if one throws an error .. py:class:: RepeatPlan(repeat=1, *children: pycram.plan.Plan) Bases: :py:obj:`LanguagePlan` A plan which repeats all children a number of times .. py:class:: MonitorPlan(condition, *children: pycram.plan.Plan) Bases: :py:obj:`LanguagePlan` A plan which monitors a condition and upon the condition becoming true interrupts all children .. py:class:: CodePlan(func: typing_extensions.Callable, kwargs: typing_extensions.Dict[str, typing_extensions.Any] = None) Bases: :py:obj:`pycram.plan.Plan` A Plan that contains a function to be executed. Mainly intended for debugging purposes .. py:class:: LanguageNode Bases: :py:obj:`pycram.plan.PlanNode` .. py:attribute:: action :type: typing_extensions.Type[LanguageNode] Superclass for language nodes in a plan. Used to distinguish language nodes from other types of nodes. .. py:class:: SequentialNode Bases: :py:obj:`LanguageNode` .. py:attribute:: action :type: typing_extensions.Type[SequentialNode] Executes all children sequentially, an exception while executing a child does not terminate the whole process. Instead, the exception is saved to a list of all exceptions thrown during execution and returned. Behaviour: Returns a tuple containing the final state of execution (SUCCEEDED, FAILED) and a list of results from each child's perform() method. The state is :py:attr:`~TaskStatus.SUCCEEDED` *iff* all children are executed without exception. In any other case the State :py:attr:`~TaskStatus.FAILED` will be returned. .. py:method:: perform() Behaviour of Sequential, calls perform() on each child sequentially :return: The state and list of results according to the behaviour described in :func:`Sequential` .. py:method:: perform_sequential(nodes: typing_extensions.List[pycram.plan.PlanNode], raise_exceptions=True) -> typing_extensions.Any Behavior of the sequential node, performs all children in sequence and raises error if they occur. :param nodes: A list of nodes which should be performed in sequence :param raise_exceptions: If True (default) errors will be raised .. py:method:: __hash__() .. py:method:: __repr__() .. py:class:: ParallelNode Bases: :py:obj:`LanguageNode` Executes all children in parallel by creating a thread per children and executing them in the respective thread. All exceptions during execution will be caught, saved to a list and returned upon end. Behaviour: Returns a tuple containing the final state of execution (SUCCEEDED, FAILED) and a list of results from each child's perform() method. The state is :py:attr:`~TaskStatus.SUCCEEDED` *iff* all children could be executed without an exception. In any other case the State :py:attr:`~TaskStatus.FAILED` will be returned. .. py:method:: perform() Behaviour of Parallel, creates a new thread for each child and calls perform() of the child in the respective thread. :return: The state and list of results according to the behaviour described in :func:`Parallel` .. py:method:: perform_parallel(nodes: typing_extensions.List[pycram.plan.PlanNode]) Behaviour of the parallel node performs the given nodes in parallel in different threads. :param nodes: A list of nodes which should be performed in parallel .. py:method:: _lang_call(node: pycram.plan.PlanNode) Wrapper method which is executed in the thread. Wraps the given node in a try catch and performs it :param node: The node which is to be performed .. py:method:: __hash__() .. py:method:: __repr__() .. py:class:: RepeatNode Bases: :py:obj:`SequentialNode` .. py:attribute:: repeat :type: int :value: 1 Executes all children a given number of times in sequential order. .. py:method:: perform() Behaviour of repeat, executes all children in a loop as often as stated on initialization. :return: .. py:method:: __hash__() .. py:class:: MonitorNode(condition: typing_extensions.Union[typing_extensions.Callable, pycram.fluent.Fluent] = None) Bases: :py:obj:`SequentialNode` Monitors a Language Expression and interrupts it when the given condition is evaluated to True. Behaviour: This Monitor is attached to a language expression, when perform on this Monitor is called it will start a new thread which continuously checks if the condition is True. When the condition is True the interrupt function of the child will be called. .. py:attribute:: kill_event .. py:attribute:: exception_queue .. py:method:: perform() Behavior of the Monitor, starts a new Thread which checks the condition and then performs the attached language expression :return: The state of the attached language expression, as well as a list of the results of the children .. py:method:: monitor() .. py:method:: __hash__() .. py:class:: TryInOrderNode Bases: :py:obj:`SequentialNode` Executes all children sequentially, an exception while executing a child does not terminate the whole process. Instead, the exception is saved to a list of all exceptions thrown during execution and returned. Behaviour: Returns a tuple containing the final state of execution (SUCCEEDED, FAILED) and a list of results from each child's perform() method. The state is :py:attr:`~TaskStatus.SUCCEEDED` if one or more children are executed without exception. In the case that all children could not be executed the State :py:attr:`~TaskStatus.FAILED` will be returned. .. py:method:: perform() Behaviour of TryInOrder, calls perform() on each child sequentially and catches raised exceptions. :return: The state and list of results according to the behaviour described in :func:`TryInOrder` .. py:method:: __hash__() .. py:class:: TryAllNode Bases: :py:obj:`ParallelNode` Executes all children in parallel by creating a thread per children and executing them in the respective thread. All exceptions during execution will be caught, saved to a list and returned upon end. Behaviour: Returns a tuple containing the final state of execution (SUCCEEDED, FAILED) and a list of results from each child's perform() method. The state is :py:attr:`~TaskStatus.SUCCEEDED` if one or more children could be executed without raising an exception. If all children fail the State :py:attr:`~TaskStatus.FAILED` will be returned. .. py:method:: perform() Behaviour of TryAll, creates a new thread for each child and executes all children in their respective threads. :return: The state and list of results according to the behaviour described in :func:`TryAll` .. py:method:: __hash__() .. py:class:: CodeNode(function: typing_extensions.Optional[typing_extensions.Callable] = None, kwargs: typing_extensions.Optional[typing_extensions.Dict] = None) Bases: :py:obj:`LanguageNode` Executable code block in a plan. :ivar function: The function (plan) that was called :ivar kwargs: Dictionary holding the keyword arguments of the function .. py:attribute:: action :type: LanguageNode Superclass for language nodes in a plan. Used to distinguish language nodes from other types of nodes. .. py:attribute:: function :type: typing_extensions.Callable :value: None .. py:attribute:: kwargs :type: typing_extensions.Dict[str, typing_extensions.Any] :value: None .. py:attribute:: perform .. py:attribute:: performable .. py:method:: execute() -> typing_extensions.Any Execute the code with its arguments :returns: Anything that the function associated with this object will return. .. py:method:: resolve() -> typing_extensions.Self .. py:method:: __hash__()