dootle.control.fsm.fsm_tracker

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

Abstract

Classes

FSMTracker

Tracks tasks by their FSM state

Module Contents

dootle.control.fsm.fsm_tracker.Abstract: TypeAlias = T

class dootle.control.fsm.fsm_tracker.FSMTracker(**kwargs: Any)

Bases: doot.control.tracker.Tracker_abs

Tracks tasks by their FSM state

TODO modify default ctor’s of specs to be FSMTask on register

Parameters:

kwargs (Any)

machines: dict[doot.workflow.TaskName | doot.workflow._interface.TaskName_p, dootle.control.fsm.machines.TaskMachine]

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

Parameters:

target (jgdv.Maybe[str | doot.workflow.TaskName])

Return type:

jgdv.Maybe[doot.workflow._interface.Task_p | doot.workflow.TaskArtifact]

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]]

Parameters:

• name (str | doot.workflow._interface.TaskName_p | doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i)

• from_user (bool)

• status (jgdv.Maybe[doot.workflow._interface.TaskStatus_e])

• kwargs (Any)

Return type:

jgdv.Maybe[Concrete[doot.workflow._interface.TaskName_p | doot.workflow._interface.Artifact_i]]

_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

Parameters:

• target (doot.workflow._interface.TaskName_p | doot.workflow._interface.RelationSpec_i)

• args (Any)

• task (bool)

• kwargs (Any)

Return type:

jgdv.Maybe[doot.workflow._interface.TaskName_p]

get_status(

*,

target: jgdv.Maybe[doot.workflow._interface.TaskName_p] = None,

) → tuple[doot.workflow._interface.TaskStatus_e, Priority]

Parameters:

target (jgdv.Maybe[doot.workflow._interface.TaskName_p])

Return type:

tuple[doot.workflow._interface.TaskStatus_e, Priority]

set_status(*args: Any) → None

No-op as the FSM’s control status

Parameters:

args (Any)

Return type:

None

_dependency_states_of(

focus: doot.workflow._interface.TaskName_p,

) → list[tuple]

Parameters:

focus (doot.workflow._interface.TaskName_p)

Return type:

list[tuple]

_successor_states_of(

focus: doot.workflow._interface.TaskName_p,

) → list[tuple]

Parameters:

focus (doot.workflow._interface.TaskName_p)

Return type:

list[tuple]