1#!/usr/bin/env python3
2"""
3
4
5"""
6
7# Imports:
8from __future__ import annotations
9
10# ##-- stdlib imports
11# import abc
12import datetime
13import enum
14import functools as ftz
15import importlib
16import itertools as itz
17import logging as logmod
18import pathlib as pl
19import re
20import time
21import types
22import weakref
23# from copy import deepcopy
24from dataclasses import MISSING, InitVar, dataclass, field
25from typing import (TYPE_CHECKING, Any, Callable, ClassVar, Final, Generator,
26 Generic, GenericAlias, Iterable, Iterator, Mapping, Match,
27 MutableMapping, Protocol, Sequence, Tuple, TypeAlias,
28 TypeGuard, TypeVar, cast, final, overload,
29 runtime_checkable)
30from uuid import UUID, uuid1
31
32# ##-- end stdlib imports
33
34# ##-- 3rd party imports
35from pydantic import (BaseModel, Field, InstanceOf, field_validator,
36 model_validator)
37from jgdv import Maybe
38from jgdv.structs.chainguard import ChainGuard
39from jgdv.structs.strang import CodeReference
40from jgdv.structs.locator import Location
41# ##-- end 3rd party imports
42
43# ##-- 1st party imports
44import doot
45import doot.errors
46from jgdv._abstract.protocols.general import (Buildable_p, StubStruct_p)
47from jgdv._abstract.protocols.pydantic import ProtocolModelMeta
48from doot.workflow.structs.task_name import TaskName
49from doot.workflow.structs.task_spec import TaskSpec, TaskMeta_e
50from doot.workflow._interface import QueueMeta_e
51
52# ##-- end 1st party imports
53
54##-- logging
55logging = logmod.getLogger(__name__)
56##-- end logging
57
58TaskFlagNames : Final[list[str]] = [x.name for x in TaskMeta_e]
59
60DEFAULT_CTOR : Final[CodeReference] = CodeReference("cls::" + doot.aliases.task[doot.constants.entrypoints.DEFAULT_TASK_CTOR_ALIAS])
61
[docs]
62class TaskStub(BaseModel, StubStruct_p, Buildable_p, metaclass=ProtocolModelMeta, arbitrary_types_allowed=True):
63 """ Stub Task Spec for description in toml
64 Automatically Adds default keys from TaskSpec
65
66 This essentially wraps a dict, adding toml stubs parts as you access keys.
67 eg:
68 obj = TaskStub()
69 ob["blah"].type = "int"
70
71 # str(obj) -> will now generate toml, including a "blah" key
72
73 """
74 ctor : str|CodeReference|type = DEFAULT_CTOR
75 parts : dict[str, TaskStubPart] = {}
76
77 # Don't copy these from TaskSpec blindly
78 skip_parts : ClassVar[set[str]] = set(["name", "extra", "ctor", "source", "version", "queue_behaviour"])
79
[docs]
80 @classmethod
81 def build(cls, data:Maybe[dict]=None):
82 match data:
83 case None:
84 return cls()
85 case _:
86 return cls(**data)
87
[docs]
88 @model_validator(mode="after")
89 def initial_values(self):
90 self['name'].default = TaskName(doot.constants.names.DEFAULT_STUB_TASK_NAME)
91 self['version'].default = "0.1"
92 # Auto populate the stub with what fields are defined in a TaskSpec:
93 for dcfield, data in TaskSpec.model_fields.items():
94 if dcfield in TaskStub.skip_parts:
95 continue
96
97 self.parts[dcfield] = TaskStubPart(key=dcfield, type_=data.annotation)
98 if data.default_factory is not None:
99 self.parts[dcfield].default = data.default_factory()
100 else:
101 self.parts[dcfield].default= data.default
102
103 return self
104
105 def __getitem__(self, key):
106 """ If the key doesnt exist, a new stub part is created """
107 if key not in self.parts:
108 self.parts[key] = TaskStubPart(key=key)
109 return self.parts[key]
110
111 def __contains__(self, key):
112 return key in self.parts
113
114 def __iadd__(self, other):
115 match other:
116 case [head, val] if head in self.parts:
117 self.parts[head].default = val
118 case [head, val]:
119 self.parts[head] = TaskStubPart(head, default=val)
120 case { "name" : name, "type": type, "default": default, "doc": doc, }:
121 pass
122 case { "name" : name, "default": default }:
123 pass
124 case dict():
125 part = TaskStubPart(**other)
126 case ChainGuard():
127 pass
128 case TaskStubPart() if other.key not in self.parts:
129 self.parts[other.key] = other
130 case _:
131 raise TypeError("Unrecognized Toml Stub component")
132
[docs]
133 def to_toml(self) -> str:
134 parts = []
135 parts.append(self.parts['name'])
136 parts.append(self.parts['version'])
137 parts.append(self.parts['doc'])
138 if 'ctor' in self.parts:
139 parts.append(self.parts['ctor'])
140 elif isinstance(self.ctor, type):
141 parts.append(TaskStubPart(key="ctor", type_="type", default=f"\"{self.ctor.__module__}{doot.constants.patterns.IMPORT_SEP}{self.ctor.__name__}\""))
142 else:
143 parts.append(TaskStubPart(key="ctor", type_="type", default=f"\"{self.ctor}\""))
144
145 delayed_actions = []
146 for key, part in sorted(self.parts.items(), key=lambda x: x[1]):
147 if key in ["name", "version", "ctor", "doc"]:
148 continue
149 if 'actions' in key:
150 delayed_actions.append(part)
151 continue
152 parts.append(part)
153
154 # Actions always go at the end
155 for part in delayed_actions:
156 parts.append(part)
157
158 return "\n".join(map(str, parts))
159
[docs]
160class TaskStubPart(BaseModel, arbitrary_types_allowed=True):
161 """ Describes a single part of a stub task in toml """
162 key : str
163 type_ : str|InstanceOf[type]|Any = "str"
164 prefix : str = ""
165
166 default : Any = Field(default="Undefined")
167 comment : str = ""
168 priority : int = 0
169
170 def __lt__(self, other):
171 return self.priority < other.priority
172
173 def __str__(self) -> str:
174 """
175 the main conversion method of a stub part -> toml string
176 the match statement handles the logic of different types.
177 eg: lowercasing the python bool from False to false for toml
178 """
179 # shortcut on being the name:
180 if isinstance(self.default, TaskName) and self.key == "name":
181 return f"[[tasks.{self.default[0,:]}]]\n{'name':<20} = \"{self.default[1,:]}\""
182
183 key_str = self._key_str()
184 # type_str = self._type_str()
185 # comment_str = self._comment_str()
186 val_str = self._default_str()
187
188 return f"{self.prefix}{key_str} = {val_str}"
189
[docs]
190 def _key_str(self) -> str:
191 return f"{self.key:<20}"
192
[docs]
193 def _type_str(self) -> str:
194 match type(self.type_), self.type_:
195 case _, t if hasattr(t, "__name__"):
196 return f"<{self.type_.__name__}>"
197 case _, _:
198 return f"<{self.type_}>"
199
202
[docs]
203 def _default_str(self) -> str:
204 """ Formats the default toml representation of this stub part"""
205 match self.default:
206 case "" if isinstance(self.type_, enum.EnumMeta):
207 val_str = f'[ "{self.type_.default.name}" ]'
208 case enum.Flag(): # TaskMeta_e()
209 parts = [x.name for x in TaskMeta_e if x in self.default]
210 joined = ", ".join(map(lambda x: f"\"{x}\"", parts))
211 val_str = f"[ {joined} ]"
212 case QueueMeta_e():
213 val_str = '"{}"'.format(self.default.name)
214 case bool():
215 val_str = str(self.default).lower()
216 case str() if self.type_ == "type":
217 val_str = self.default
218 case int() | float():
219 val_str = f"{self.default}"
220 case str() if "\n" in self.default:
221 flat = self.default.replace("\n", "\\n")
222 val_str = f'"{flat}"'
223 case str():
224 val_str = f'"{self.default}"'
225 case list() if all(isinstance(x, int|float) for x in self.default):
226 def_str = ", ".join(str(x) for x in self.default)
227 val_str = f"[{def_str}]"
228 case set() | list() | tuple():
229 parts = ", ".join([f'"{x}"' for x in self.default])
230 val_str = f"[{parts}]"
231 case dict() | ChainGuard() if not bool(self.default):
232 val_str = "{}"
233 case _:
234 logging.debug("Unknown stub part reduction: %s : %s : %s", self.key, self.type_, self.default)
235 val_str = '"unknown"'
236
237 return val_str
238
[docs]
239 def set(self, **kwargs):
240 self.type_ = kwargs.pop('type_', self.type_)
241 self.prefix = kwargs.pop('prefix', self.prefix)
242 self.default = kwargs.pop('default', self.default)
243 self.comment = kwargs.pop('comment', self.comment)
244 self.priority = kwargs.pop('priority', self.priority)