1#!/usr/bin/env python3
2"""
3
4"""
5# mypy: disable-error-code="attr-defined"
6# ruff: noqa: B009
7# Imports:
8from __future__ import annotations
9
10# ##-- stdlib imports
11import datetime
12import enum
13import functools as ftz
14import importlib
15import itertools as itz
16import logging as logmod
17import pathlib as pl
18import re
19import time
20import types
21from collections import defaultdict
22from copy import deepcopy
23from importlib.resources import files
24from uuid import UUID, uuid1
25from weakref import ref
26
27# ##-- end stdlib imports
28
29# ##-- 3rd party imports
30from jgdv import Mixin, Proto
31from jgdv.structs.strang import CodeReference
32
33# ##-- end 3rd party imports
34
35# ##-- 1st party imports
36import doot
37import doot.errors
38from doot.util.dkey import DKey
39from doot.util.dkey import DKeyed
40from doot.workflow.job import DootJob
41from doot.workflow.structs.task_name import TaskName
42from doot.workflow.task import DootTask
43
44# ##-- end 1st party imports
45
46# ##-| Local
47from ._base import BaseCommand
48from .structs import TaskStub
49
50# # End of Imports.
51
52# ##-- types
53# isort: off
54import abc
55import collections.abc
56from typing import TYPE_CHECKING, cast, assert_type, assert_never
57from typing import Generic, NewType
58# Protocols:
59from typing import Protocol, runtime_checkable
60# Typing Decorators:
61from typing import no_type_check, final, override, overload
62
63if TYPE_CHECKING:
64 from jgdv import Maybe, Lambda
65 from jgdv.structs.chainguard import ChainGuard
66 from typing import Final
67 from typing import ClassVar, Any, LiteralString
68 from typing import Never, Self, Literal
69 from typing import TypeGuard
70 from collections.abc import Iterable, Iterator, Callable, Generator
71 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
72 type ListVal = str|Lambda|tuple[str,dict]
73
74##--|
75from doot.control.loaders._interface import PluginLoader_p
76from doot.workflow._interface import Task_p
77from ._interface import Command_p
78# isort: on
79# ##-- end types
80
81##-- logging
82logging = logmod.getLogger(__name__)
83##-- end logging
84
85##-- data
86data_path = files(doot.constants.paths.TEMPLATE_PATH).joinpath(doot.constants.paths.TOML_TEMPLATE) # type: ignore[attr-defined]
87##-- end data
88
89PRINT_LOCATIONS : Final[list[str]] = doot.constants.printer.PRINT_LOCATIONS # type: ignore[attr-defined]
90NL = None
91##--|
92
[docs]
93class _StubDoot_m:
94 """ Mixin for stubbing the doot.toml file """
95
[docs]
96 def param_specs(self:Command_p) -> list:
97 return [
98 *super().param_specs(), # type: ignore[safe-super]
99 self.build_param(name="--config", type=bool, default=False, desc="Stub a doot.toml"), # type: ignore[attr-defined]
100 ]
101
[docs]
102 def _stub_doot_toml(self) -> list[str]:
103 logging.info("---- Stubbing Doot Toml")
104 doot_toml = pl.Path("doot.toml")
105 data_text = data_path.read_text()
106 if doot_toml.exists():
107 return [
108 data_text,
109 "# doot.toml it already exists, printed to stdout instead",
110 ]
111
112 with doot_toml.open("a") as f:
113 f.write(data_text)
114
115 doot.report.gen.user("doot.toml stub")
116 return []
117
[docs]
118class _StubParam_m:
119 """ Mixin for stubbing a cli parameter """
120
[docs]
121 def param_specs(self) -> list:
122 return [
123 *super().param_specs(), # type: ignore[misc]
124 self.build_param(name="--param", type=bool, default=False, desc="Generate a stub cli arg dict"),
125 ]
126
[docs]
127 def _stub_cli_param(self) -> list[str]:
128 logging.info("---- Printing CLI Arg info")
129 result = [
130 "# - CLI Arg Form. Add to task spec: cli=[]",
131 '{',
132 'name="{}",'.format(doot.args.on_fail("default").cmds.args.name()),
133 'prefix="-", ',
134 'type="str", ',
135 'default="",',
136 'desc="", ',
137 "}",
138 ]
139 return result
140
[docs]
141class _StubAction_m:
142 """ Mixin for stubbing an action """
143
[docs]
144 def param_specs(self) -> list:
145 return [
146 *super().param_specs(), # type: ignore[misc]
147 self.build_param(name="--action", type=bool, default=False, desc="Help Stub Actions"),
148 ]
149
[docs]
150 def _stub_action(self, idx:int, plugins:ChainGuard) -> list[Maybe[str]]:
151 logging.info("---- Stubbing Actions")
152 result : list[Maybe[str]] = []
153 target_name = doot.args.cmds[self.name][idx].args.name
154 unaliased = doot.aliases.on_fail(target_name).action[target_name]()
155 matched = [x for x in plugins.action
156 if x.name == target_name
157 or x.value == unaliased]
158 if bool(matched):
159 loaded = matched[0].load()
160 result.append(f"- {matched[0].name} (Action, {matched[0].value})")
161 match getattr(loaded, "_toml_help", []):
162 case [] if bool(getattr(loaded, "__doc__")):
163 result.append(loaded.__doc__)
164 case []:
165 pass
166 case [*xs]:
167 for x in xs:
168 result.append(x)
169
170 loaded = getattr(loaded, "__call__", loaded) # noqa: B004
171 match DKeyed.get_keys(loaded):
172 case []:
173 result.append("-- No Declared Kwargs")
174 case [*xs]:
175 result += [
176 "-- Declared kwargs for action:",
177 *(f"---- {x!r}" for x in sorted(xs, key=lambda x: repr(x))),
178 ]
179
180 result.append(NL)
181 result.append("-- Toml Form of an action: ")
182 # TODO customize this with declared annotations
183 if bool(matched):
184 result.append(f"{{ do=\"{matched[0].name}\", args=[], key=val }} ")
185 else:
186 result.append("{ do=\"action name/import path\", args=[]} # plus any kwargs a specific action uses")
187
188 return result
189
[docs]
190class _StubTask_m:
191 """ Mixin for stubbing a task """
192
[docs]
193 def param_specs(self) -> list:
194 return [
195 *super().param_specs(), # type: ignore[misc]
196 self.build_param(name="--task", type=bool, desc="Stub a Task Specification"),
197 self.build_param(name="-out", type=str, default="", desc="If set, append the stub to this file"),
198
199 self.build_param(name="<1>name", type=str, default=None, desc="The Name of the new task"),
200 self.build_param(name="<2>ctor", type=str, default="task", desc="a code ref, or alias of a task class"),
201 ]
202
[docs]
203 def _stub_task_toml(self, idx:int, tasks:ChainGuard, plugins:ChainGuard) -> list[str]:
204 """
205 This creates a toml stub using default values, as best it can
206 """
207 logging.info("---- Stubbing Task Toml")
208 result = []
209
210 # Create stub toml, with some basic information
211 stub = TaskStub()
212 stub['name'].default = self._stub_task_name(idx, tasks)
213 self._add_ctor_specific_stub_fields(idx, stub)
214
215 # Output to doot.report/stdout, or file
216 match doot.args.on_fail("").cmds[self.name][idx].args.out():
217 case "":
218 result.append(stub.to_toml())
219 return result
220 case str() as x:
221 task_fail = pl.Path(x)
222
223 if task_file.is_dir():
224 task_file /= "stub_tasks.toml"
225 doot.report.gen.user("Stubbing task %s into file: %s", stub['name'], task_file)
226 with task_file.open("a") as f:
227 f.write("\n")
228 f.write(stub.to_toml())
229
230 return []
231
[docs]
232 def _stub_task_name(self, idx:int, tasks:ChainGuard) -> str:
233 match doot.args.on_fail(None).cmds[self.name][idx].args.name():
234 case None:
235 raise doot.errors.CommandError("No Name Provided for Stub")
236 case '':
237 name = TaskName("example::task")
238 case x:
239 name = TaskName(x)
240
241 # extend the name if there are already tasks with that name
242 original_name = name
243 count = 0
244 while str(name) in tasks:
245 count += 1
246 name = original_name.push("$conflicted$", str(count))
247 else:
248 return name
249
[docs]
250 def _add_ctor_specific_stub_fields(self, idx:int, stub:TaskStub) -> None:
251 """ add ctor specific fields,
252 such as for dir_walker: roots [], exts [], recursive bool, subtask "", head_task ""
253 works *towards* the task_type, not away, so more specific elements are added over the top of more general elements
254 """
255 task_mro : Iterable
256 ##--|
257 match doot.aliases.task.get((ctor:=doot.args.on_fail("task").cmds[self.name][idx].args.ctor()), None):
258 case None:
259 raise doot.errors.CommandError("Task Ctor was not appliable", ctor)
260 case x:
261 task_ctor : CodeReference = CodeReference(x)
262
263 try:
264 match task_ctor():
265 case type() as ctor:
266 task_mro = ctor.mro()
267 case Exception() as err:
268 raise err
269 except TypeError as err:
270 logging.exception(err.args[0].replace("\n", ""))
271 task_mro = []
272 return
273
274 for cls in reversed(task_mro):
275 try:
276 cls.stub_class(stub)
277 if isinstance(cls, Task_p):
278 stub['doot_version'].default = doot.__version__
279 stub['doc'].default = []
280 except NotImplementedError:
281 pass
282 except AttributeError:
283 pass
284
285 # Convert to aliases
286 stub['ctor'].default = task_ctor
287
[docs]
288class _StubPrinter_m:
289 """
290 Mixin for stubbing printer config
291 """
292
[docs]
293 def param_specs(self) -> list:
294 return [
295 *super().param_specs(), # type: ignore[misc]
296 self.build_param(name="--report", type=bool, default=False, desc="Generate a stub doot.report config"),
297 ]
298
[docs]
299 def _stub_printer(self) -> list[Maybe[str|tuple]]:
300 logging.info("---- Printing Printer Spec Info")
301 result = [
302 ("- Printer Config Spec Form. Use in doot.toml [logging], [logging.subprinters], and [logging.extra]", {"colour":"blue"}),
303 NL,
304 'NAME = { level="", filter=[], target=[""], format="", colour=true, propagate=false, filename_fmt=""}',
305 ]
306 return result
307
308##--|
309
[docs]
310@Proto(Command_p)
311@Mixin(_StubDoot_m, _StubParam_m, _StubAction_m, _StubTask_m, _StubPrinter_m)
312class StubCmd(BaseCommand):
313 """ Called to interactively create a stub task definition
314 with a `target`, outputs to that file, else to stdout for piping
315 """
316 _name = "stub" # type: ignore[misc]
317 _help = tuple(["Create a new stub task either to stdout, or path",
318 "args allow stubbing a config file, cli parameter, or action",
319 ])
320
[docs]
321 @override
322 def param_specs(self) -> list:
323 return [
324 *super().param_specs(),
325 self.build_param(name="--strang", type=bool, default=False, desc="Generate a stub strang/location expansion"),
326 self.build_param(name="--suppress-header", default=True, implicit=True),
327 ]
328
329 def __call__(self, *, idx:int, tasks:ChainGuard, plugins:ChainGuard):
330 match dict(doot.args.cmds[self.name][idx].args):
331 case {"config": True}:
332 result = self._stub_doot_toml()
333 case {"action": True}:
334 result = self._stub_action(idx, plugins)
335 case {"param": True}:
336 result = self._stub_cli_param()
337 case {"report": True}:
338 result = self._stub_printer()
339 case {"strang": True}:
340 result = "TODO"
341 case _:
342 result = self._stub_task_toml(idx, tasks, plugins)
343 ##--|
344 self._print_text(result)