1## base_action.py -*- mode: python -*-
2# Imports:
3from __future__ import annotations
4
5# ##-- stdlib imports
6import datetime
7import enum
8import functools as ftz
9import itertools as itz
10import logging as logmod
11import pathlib as pl
12import re
13import shutil
14import time
15import types
16from time import sleep
17from uuid import UUID, uuid1
18
19# ##-- end stdlib imports
20
21# ##-- 3rd party imports
22import sh
23from jgdv import Mixin, Proto
24from jgdv.mixins.path_manip import PathManip_m
25from jgdv.structs.strang import CodeReference
26
27# ##-- end 3rd party imports
28
29# ##-- 1st party imports
30import doot
31from doot.errors import TaskError, TaskFailed
32from doot.mixins.path_manip import PathManip_m
33from doot.util.dkey import DKey, DKeyed
34
35# ##-- end 1st party imports
36
37# ##-| Local
38from ..._interface import Action_p
39
40# # End of Imports.
41
42# ##-- types
43# isort: off
44import abc
45import collections.abc
46from typing import TYPE_CHECKING, cast, assert_type, assert_never
47from typing import Generic, NewType
48# Protocols:
49from typing import Protocol, runtime_checkable
50# Typing Decorators:
51from typing import no_type_check, final, override, overload
52
53if TYPE_CHECKING:
54 from jgdv import Maybe
55 from typing import Final
56 from typing import ClassVar, Any, LiteralString
57 from typing import Never, Self, Literal
58 from typing import TypeGuard
59 from collections.abc import Iterable, Iterator, Callable, Generator
60 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
61
62# isort: on
63# ##-- end types
64
65
[docs]
66@Proto(Action_p)
67class AddStateAction:
68 """
69 add to task state in the task description toml,
70 adds kwargs directly, without expansion
71 """
72
73 @DKeyed.kwargs
74 def __call__(self, spec, state:dict, kwargs) -> dict|bool|None:
75 result = {}
76 for k,v in kwargs.items():
77 key = DKey(v)
78 val = key.expand(spec, state)
79 result[k] = val
80 return result
81
[docs]
82@Proto(Action_p)
83class AddStateFn:
84 """ for each toml kwarg, import its value and set the state[kwarg] = val
85 with expansion
86 """
87
88 @DKeyed.kwargs
89 def __call__(self, spec, state:dict, kwargs) -> dict|bool|None:
90 result = {}
91 for kwarg, val in kwargs:
92 key = DKey(val)
93 val = key.expand(spec, state)
94 ref = CodeReference(val)
95 result[kwarg] = ref()
96
97 return result
98
[docs]
99@Proto(Action_p)
100class PushState:
101 """
102 state[update_] += [state[x] for x in spec.args]
103 """
104
105 @DKeyed.args
106 @DKeyed.types("update_", check=set|list, fallback=list)
107 @DKeyed.redirects("update_")
108 def __call__(self, spec, state, args, _data, _update) -> dict|bool|None:
109 target_data = data.copy()
110 to_add = []
111 for x in args:
112 match DKey(x).expand(spec, state):
113 case None:
114 pass
115 case list()|set() as xs:
116 to_add += xs
117 case x:
118 to_add.append(x)
119
120 match target_data:
121 case set():
122 target_data.update(to_add)
123 case list():
124 target_data.extend(to_add)
125 case _:
126 raise TypeError("Unknown state target to push to", type(target_data), _update)
127
128 return { _update : target_data }
129
[docs]
130@Proto(Action_p)
131class AddNow:
132 """
133 Add the current date, as a string, to the state
134 """
135
136 @DKeyed.expands("format")
137 @DKeyed.redirects("update_")
138 def __call__(self, spec, state, format, _update):
139 now = datetime.datetime.now()
140 return { _update : now.strftime(format) }
141
[docs]
142@Proto(Action_p)
143@Mixin(PathManip_m, allow_inheritance=True)
144class PathParts:
145 """ take a path and add fstem, fpar, fname to state """
146
147 @DKeyed.paths("from")
148 @DKeyed.types("roots")
149 @DKeyed.returns("fstem", "fpar", "fname", "fext", "pstem", "rpath")
150 def __call__(self, spec, state, _from, roots):
151 root_paths = self._build_roots(spec, state, roots)
152 return self._calc_path_parts(_from, root_paths)
153
[docs]
154@Proto(Action_p)
155@Mixin(PathManip_m, allow_inheritance=True)
156class ShadowPath:
157
158 @DKeyed.paths("shadow_root")
159 @DKeyed.types("base", check=pl.Path)
160 def __call__(self, spec, state, shadow_root, base):
161 shadow_dir = self._shadow_path(base, shadow_root)
162 return { "shadow_path" : shadow_dir }