Source code for doot.workflow.transformer

  1#!/usr/bin/env python3
  2"""
  3Utility classes for building tasks with a bit of structure
  4"""
  5# Imports:
  6from __future__ import annotations
  7
  8# ##-- stdlib imports
  9import datetime
 10import enum
 11import functools as ftz
 12import itertools as itz
 13import logging as logmod
 14import pathlib as pl
 15import re
 16import time
 17from uuid import UUID, uuid1
 18
 19# ##-- end stdlib imports
 20
 21# ##-- 3rd party imports
 22from jgdv import Proto
 23from jgdv.structs.strang import CodeReference
 24# ##-- end 3rd party imports
 25
 26# ##-- 1st party imports
 27import doot
 28import doot.errors
 29from ._interface import TaskMeta_e
 30from .structs.task_name import TaskName
 31from .task import DootTask
 32
 33# ##-- end 1st party imports
 34
 35# ##-- types
 36# isort: off
 37import abc
 38import collections.abc
 39from typing import TYPE_CHECKING, cast, assert_type, assert_never
 40from typing import Generic, NewType
 41# Protocols:
 42from typing import Protocol, runtime_checkable
 43# Typing Decorators:
 44from typing import no_type_check, final, override, overload
 45
 46if TYPE_CHECKING:
 47    from jgdv import Maybe
 48    from typing import Final
 49    from typing import ClassVar, Any, LiteralString
 50    from typing import Never, Self, Literal
 51    from typing import TypeGuard
 52    from collections.abc import Iterable, Iterator, Callable, Generator
 53    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 54    from .structs import TaskStub, TaskSpec
 55
 56##--|
 57# isort: on
 58# ##-- end types
 59
 60##-- logging
 61logging = logmod.getLogger(__name__)
 62##-- end logging
 63

[docs]
 64class DootTransformer(DootTask):
 65    """
 66      Transformers have an abstract artifact dependency and product,
 67      and will auto-add to the task graph to transform that artifact
 68    """
 69    _help : ClassVar[tuple[str]] = tuple(["A Basic Task Constructor"])
 70    _default_flags = TaskMeta_e.TRANSFORMER
 71
 72    def __init__(self, spec:TaskSpec):
 73        assert(spec is not None), "Spec is empty"
 74        super().__init__(spec)
 75

[docs]
 76    @classmethod
 77    def stub_class(cls, stub:TaskStub) -> TaskStub:
 78        # Come first
 79        stub['active_when'].priority    = -90
 80        stub['required_for'].priority   = -90
 81        stub['depends_on'].priority     = -100
 82
 83        return stub

 84

[docs]
 85    def stub_instance(self, stub:TaskStub) -> TaskStub:
 86        stub                      = self.__class__.stub_class(stub)
 87        stub['name'].default      = self.name.de_uniq()
 88        if bool(self.doc):
 89            stub['doc'].default   = [f"\"{x}\"" for x in self.doc]
 90        return stub

 91

[docs]
 92    @classmethod
 93    def class_help(cls) -> str:
 94        """ Job *class* help. """
 95        help_lines = [f"Job : {cls.__qualname__} v{cls._version}    ({cls.__module__}:{cls.__qualname__})", ""]
 96
 97        mro = " -> ".join(x.__name__ for x in cls.mro())
 98        help_lines.append(f"Job MRO: {mro}")
 99        help_lines.append("")
100        help_lines += cls._help
101
102        params = cls.param_specs()
103        if bool(params):
104            help_lines += ["", "Params:"]
105            help_lines += [y for x in cls.param_specs() if (y:=str(x))]
106
107        return "\n".join(help_lines)