.. _dootle.control.fsm.fsm_tracker: ============================== dootle.control.fsm.fsm_tracker ============================== .. py:module:: dootle.control.fsm.fsm_tracker .. autoapi-nested-parse:: An Alternative task tracker for doot. Uses FSM's to simplify StateTracker.next_for. The FSM wraps a DootTask/Job/Artifact, and manages its progression Type Aliases ------------ .. autoapisummary:: dootle.control.fsm.fsm_tracker.Abstract Classes ------- .. autoapisummary:: dootle.control.fsm.fsm_tracker.FSMTracker Module Contents =============== .. py:data:: Abstract :type: TypeAlias :value: T .. _dootle.control.fsm.fsm_tracker.FSMTracker: .. py:class:: FSMTracker(**kwargs: Any) Bases: :py:obj:`doot.control.tracker.Tracker_abs` Tracks tasks by their FSM state TODO modify default ctor's of specs to be FSMTask on register .. py:attribute:: machines :type: dict[doot.workflow.TaskName | doot.workflow._interface.TaskName_p, dootle.control.fsm.machines.TaskMachine] .. py:method:: next_for(target: jgdv.Maybe[str | doot.workflow.TaskName] = None) -> jgdv.Maybe[doot.workflow._interface.Task_p | doot.workflow.TaskArtifact] ask for the next task that can be performed Returns a Task or Artifact that needs to be executed or created Returns None if it loops too many times trying to find a target, or if theres nothing left in the queue .. py:method:: queue(name: str | doot.workflow._interface.TaskName_p | doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i, *, from_user: bool = False, status: jgdv.Maybe[doot.workflow._interface.TaskStatus_e] = None, **kwargs: Any) -> jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]] .. py:method:: _instantiate(target: doot.workflow._interface.TaskName_p | doot.workflow._interface.RelationSpec_i, *args: Any, task: bool = False, **kwargs: Any) -> jgdv.Maybe[doot.workflow._interface.TaskName_p] when a task is created, create a state machine for it as well .. py:method:: get_status(*, target: jgdv.Maybe[doot.workflow._interface.TaskName_p] = None) -> tuple[doot.workflow._interface.TaskStatus_e, Priority] .. py:method:: set_status(*args: Any) -> None No-op as the FSM's control status .. py:method:: _dependency_states_of(focus: doot.workflow._interface.TaskName_p) -> list[tuple] .. py:method:: _successor_states_of(focus: doot.workflow._interface.TaskName_p) -> list[tuple]