doot.control.tracker._base

Abstract Specs: A[n] Concrete Specs: C[n] Task: T[n]

Expansion: ∀x ∈ C[n].depends_on => A[x] -> C[x] Head: C[1].depends_on[A[n].$head$] => A[n] -> C[n], A[n].head -> C[n].head, connect

Type Aliases

Abstract

Classes

Tracker_abs

A public base implementation of most of a tracker

Module Contents

doot.control.tracker._base.Abstract: TypeAlias = T

class doot.control.tracker._base.Tracker_abs(**kwargs: Any)[source]

A public base implementation of most of a tracker Has three components: _registry : db for specs and tasks _network : the links between specs in the registry _queue : the logic for determining what task to run next

Parameters:: kwargs (Any)

_factory: doot.workflow._interface.TaskFactory_p

_subfactory: doot.workflow._interface.SubTaskFactory_p

_registry: doot.control.tracker._interface.Registry_p

_network: doot.control.tracker._interface.Network_p

_queue: doot.control.tracker._interface.Queue_p

_declare_priority: int

_min_priority: int

_root_node

property specs: dict[doot.workflow._interface.TaskName_p, doot.control.tracker._interface.SpecMeta_d][source]

Return type:: dict[doot.workflow._interface.TaskName_p, doot.control.tracker._interface.SpecMeta_d]

property artifacts: dict[doot.workflow._interface.Artifact_i, doot.control.tracker._interface.ArtifactMeta_d][source]

Return type:: dict[doot.workflow._interface.Artifact_i, doot.control.tracker._interface.ArtifactMeta_d]

property concrete: set[source]

Return type:: set

property abstract: set[source]

Return type:: set

property network: collections.abc.Mapping[source]

Return type:: collections.abc.Mapping

property active: set[source]

Return type:: set

property is_valid: bool[source]

Return type:: bool

register( *specs: doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec, ) → None[source]

Parameters:: specs (doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec)
Return type:: None

queue(

name: str | doot.workflow._interface.TaskName_p | doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec,

*,

from_user: int | 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]][source]

Parameters:

name (str | doot.workflow._interface.TaskName_p | doot.workflow._interface.TaskSpec_i | doot.workflow._interface.Artifact_i | doot.workflow._interface.DelayedSpec)
from_user (int | 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]]

build( *, sources: jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]] = None, ) → None[source]

Parameters:: sources (jgdv.Maybe[Literal[True] | list[Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i]])
Return type:: None

validate() → None[source]

Return type:: None

abstractmethod plan(*args: Any) → list[source]

Parameters:: args (Any)
Return type:: list

clear() → None[source]

Return type:: None

report( target: doot.workflow._interface.TaskName_p, ) → dict[source]

Parameters:: target (doot.workflow._interface.TaskName_p)
Return type:: dict

_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][source]

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]

_connect(

left: Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i,

right: jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i] = None,

**kwargs: Any,

) → None[source]

Parameters:

left (Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i)
right (jgdv.Maybe[Literal[False] | Concrete[doot.workflow._interface.TaskName_p] | doot.workflow._interface.Artifact_i])
kwargs (Any)

Return type:

None

_upgrade_delayed_to_actual( spec: doot.workflow._interface.DelayedSpec, ) → doot.workflow._interface.TaskSpec_i[source]

can’t be in taskfactory, as it requires the registered specs

Parameters:: spec (doot.workflow._interface.DelayedSpec)
Return type:: doot.workflow._interface.TaskSpec_i

_reify_partial_spec( spec: doot.workflow._interface.TaskSpec_i, ) → doot.workflow._interface.TaskSpec_i[source]

converts spec(name=group::task.a.b..$partial$, sources[*_, base], data) into spec(name=group::a.b, data) using base can’t be in the taskfactory, as it requires registered specs

Parameters:: spec (doot.workflow._interface.TaskSpec_i)
Return type:: doot.workflow._interface.TaskSpec_i

_generate_implicit_tasks( spec: doot.workflow._interface.TaskSpec_i, ) → list[doot.workflow._interface.TaskSpec_i][source]

Generate implicit subtasks for a concrete spec

Parameters:: spec (doot.workflow._interface.TaskSpec_i)
Return type:: list[doot.workflow._interface.TaskSpec_i]