1#!/usr/bin/env python3
2"""
3
4"""
5# ruff: noqa: ANN002, ANN003
6# Imports:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import datetime
11import enum
12import functools as ftz
13import itertools as itz
14import logging as logmod
15import pathlib as pl
16import re
17import time
18import collections
19import contextlib
20import hashlib
21from copy import deepcopy
22from uuid import UUID, uuid1
23from weakref import ref
24import atexit # for @atexit.register
25import faulthandler
26# ##-- end stdlib imports
27
28from jgdv._abstract.protocols.general import SpecStruct_p
29from jgdv.structs.strang import Strang, CodeReference
30from .._interface import Key_p, NonKey_p, MultiKey_p, IndirectKey_p, LIFT_EXPANSION_PATTERN
31
32# ##-- types
33# isort: off
34# General
35import abc
36import collections.abc
37import typing
38import types
39from typing import cast, assert_type, assert_never
40from typing import Generic, NewType, Never
41from typing import no_type_check, final, override, overload
42# Protocols and Interfaces:
43from typing import Protocol, runtime_checkable
44from collections.abc import Mapping
45from collections.abc import Sequence
46if typing.TYPE_CHECKING:
47 from typing import Final, ClassVar, Any, Self
48 from typing import Literal, LiteralString
49 from typing import TypeGuard
50 from collections.abc import Iterable, Iterator, Callable, Generator
51 from collections.abc import MutableMapping, Hashable
52
53 from jgdv import Maybe, Rx, Ident, RxStr, CtorFn, CHECKTYPE, FmtStr
54 type LitFalse = Literal[False]
55 type InstructionAlts = list[ExpInst_d]
56 type InstructionList = list[InstructionAlts]
57 type InstructionExpansions = list[ExpInst_d]
58 type ExpOpts = dict
59 type SourceBases = list|Mapping|SpecStruct_p
60
61# isort: on
62# ##-- end types
63
64##-- logging
65logging = logmod.getLogger(__name__)
66##-- end logging
67
68# Vars:
69
70##--| Error Messages
71NestedFailure : Final[str] = "Nested ExpInst_d"
72NoValueFailure : Final[str] = "ExpInst_d's must have a val"
73UnexpectedData : Final[str] = "Unexpected kwargs given to ExpInst_d"
74
75##--| Values
76NO_EXPANSIONS_PERMITTED : Final[int] = 0
77
78EXPANSION_CONVERT_MAPPING : Final[dict[str,Maybe[Callable]]] = {
79 "p" : lambda x: pl.Path(x).expanduser().resolve(),
80 "s" : str,
81 "S" : Strang,
82 "c" : CodeReference,
83 "i" : int,
84 "f" : float,
85 LIFT_EXPANSION_PATTERN : None,
86}
87##--| Data
88
[docs]
89class ExpInst_d:
90 """ The lightweight holder of expansion instructions, passed through the
91 expander mixin.
92 Uses slots to make it as lightweight as possible
93
94 - fallback : the value to use if expansion fails
95 - convert : controls type coercion of expansion result
96 - lift : says to lift expanded values into keys themselves (using !L in the key str)
97 - literal : signals the value needs no more expansion
98 - rec : the remaining recursive expansions available. -1 is unrestrained.
99 - total_recs : tracks the number of expansions have occured
100
101 """
102 __slots__ = ("convert", "fallback", "lift", "literal", "rec", "total_recs", "value")
103 value : Any
104 convert : Maybe[str|bool]
105 fallback : Maybe[str]
106 lift : bool | tuple[bool, bool]
107 literal : bool
108 rec : Maybe[int]
109 total_recs : int
110
111 def __init__(self, **kwargs) -> None:
112 self.value = kwargs.pop("value")
113 self.convert = kwargs.pop("convert", None)
114 self.fallback = kwargs.pop("fallback", None)
115 self.lift = kwargs.pop("lift", False)
116 self.literal = kwargs.pop("literal", False)
117 self.rec = kwargs.pop("rec", None)
118 self.total_recs = kwargs.pop("total_recs", 0)
119
120 assert(self.rec is None or self.rec >= 0)
121 if self.rec == 0:
122 self.literal = True
123 self.process_value()
124 if bool(kwargs):
125 raise ValueError(UnexpectedData, kwargs)
126
127 @override
128 def __repr__(self) -> str:
129 lit = "(Lit)" if self.literal else ""
130 return f"<ExpInst_d:{lit} {self.value!r} / {self.fallback!r} (R:{self.rec},L:{self.lift},C:{self.convert})>"
131
[docs]
132 def process_value(self) -> None:
133 match self.value:
134 case ExpInst_d() as val:
135 raise TypeError(NestedFailure, val)
136 case Key_p() as k if k.data.convert is not None and LIFT_EXPANSION_PATTERN in k.data.convert:
137 self.lift = True
138 case Key_p() | None:
139 pass
140
[docs]
141class ExpInstChain_d:
142 __slots__ = ("chain", "merge", "root")
143
144 root : Key_p
145 chain : tuple[ExpInst_d, ...]
146 merge : Maybe[int]
147
148 def __init__(self, *chain:ExpInst_d, root:Key_p, merge:Maybe[int]=None) -> None:
149 self.root = root
150 self.chain = tuple(chain)
151 self.merge = merge
152
153 def __getitem__(self, i:int) -> ExpInst_d:
154 return self.chain[i]
155
156 def __iter__(self) -> Iterator[ExpInst_d]:
157 return iter(self.chain)
158
159 def __len__(self) -> int:
160 return self.merge or len(self.chain)
161
162 @override
163 def __repr__(self) -> str:
164 if self.merge:
165 val = f"(M:{self.merge})"
166 else:
167 val = f"(C:{len(self.chain)})"
168 return f"<ExpChain{val}: {self.root}>"
169
[docs]
170class SourceChain_d:
171 """ The core logic to lookup a key from a sequence of sources
172
173 | Doesn't perform repeated expansions.
174 | Tries sources in order.
175 | A Source that is a list is copied and each retrieval pops a value off it
176
177 TODO replace this with collections.ChainMap ?
178 """
179 __slots__ = ("sources",)
180 sources : list[Mapping|list]
181
182 def __init__(self, *args:Maybe[SourceBases|SourceChain_d]) -> None:
183 self.sources = []
184 for base in args:
185 match base:
186 case None:
187 pass
188 case SourceChain_d():
189 self.sources += base.sources
190 case list():
191 self.sources.append(base[:])
192 case dict() | collections.ChainMap():
193 self.sources.append(base)
194 case Mapping():
195 self.sources.append(base)
196 case SpecStruct_p():
197 self.sources.append(base.params)
198 case x:
199 raise TypeError(type(x))
200
201 @override
202 def __repr__(self) -> str:
203 source_types = ", ".join([type(x).__name__ for x in self.sources])
204 return f"<{type(self).__name__}: {source_types}>"
205
[docs]
206 def extend(self, *args:SourceBases) -> SourceChain_d:
207 extension = SourceChain_d(*self.sources, *args)
208 return extension
209
[docs]
210 def lookup(self, target:ExpInstChain_d) -> Maybe[ExpInst_d|tuple]:
211 """ Look up alternatives
212
213 | pass through DKeys and (DKey, ..) for recursion
214 | lift (str(), True, fallback)
215 | don't lift (str(), False, fallback)
216
217 """
218 x : Any
219 for inst in target:
220 match inst:
221 case ExpInst_d(value=NonKey_p()) | ExpInst_d(literal=True):
222 return inst
223 case ExpInst_d() as curr:
224 match self.get(str(curr.value)):
225 case None:
226 pass
227 case x:
228 return x, curr
229 case x:
230 msg = "Unrecognized lookup spec"
231 raise TypeError(msg, x)
232 ##--|
233 else:
234 return None
235
[docs]
236 def get(self, key:str, fallback:Maybe=None) -> Maybe:
237 """ Get a key's value from an ordered sequence of potential sources.
238
239 """
240 replacement : Maybe = fallback
241 for lookup in self.sources:
242 match lookup:
243 case None | []:
244 continue
245 case list():
246 replacement = lookup.pop()
247 case _ if hasattr(lookup, "get"):
248 if key not in lookup:
249 continue
250 replacement = lookup.get(key, fallback)
251 case SpecStruct_p():
252 params = lookup.params
253 replacement = params.get(key, fallback)
254 case _:
255 msg = "Unknown Type in get"
256 raise TypeError(msg, key, lookup)
257
258 if replacement is not fallback:
259 return replacement
260 else:
261 return fallback
262
263##--| Protocols
264
[docs]
265@runtime_checkable
266class InstructionFactory_p(Protocol):
267
[docs]
268 def build_chains(self, val:ExpInst_d, opts:ExpOpts) -> list[ExpInstChain_d|ExpInst_d]: ...
269
[docs]
270 def build_inst(self, val:Maybe, root:Maybe[ExpInst_d], opts:ExpOpts, *, decrement:bool=True) -> Maybe[ExpInst_d]: ...
271
[docs]
272 def null_inst(self) -> ExpInst_d: ...
273
[docs]
274 def literal_inst(self, val:Any) -> ExpInst_d: ... # noqa: ANN401
275
[docs]
276 def lift_inst(self, val:str, root:Maybe[ExpInst_d], opts:ExpOpts, *, decrement:bool=False, implicit:bool=False) -> ExpInst_d: ...
[docs]
277class Expander_p[T](Protocol):
278
[docs]
279 def set_ctor(self, ctor:CtorFn[..., T]) -> None: ...
280
[docs]
281 def redirect(self, source:T, *sources:dict, **kwargs:Any) -> list[Maybe[ExpInst_d]]: ... # noqa: ANN401
282
[docs]
283 def expand(self, source:T, *sources:dict, **kwargs:Any) -> Maybe[ExpInst_d]: ... # noqa: ANN401
284
[docs]
285 def extra_sources(self, source:T) -> SourceChain_d: ...
286
[docs]
287 def coerce_result(self, inst:ExpInst_d, opts:ExpOpts, *, source:Key_p) -> Maybe[ExpInst_d]: ...
288
[docs]
289class ExpansionHooks_p(Protocol):
290
[docs]
291 def exp_to_inst_h(self, root:ExpInst_d, factory:InstructionFactory_p, **kwargs:Any) -> Maybe[ExpInst_d]: ... # noqa: ANN401
292
[docs]
293 def exp_generate_chains_h(self, root:ExpInst_d, factory:InstructionFactory_p, opts:ExpOpts) -> list[ExpInstChain_d|ExpInst_d]: ...
294
296
[docs]
297 def exp_flatten_h(self, values:list[Maybe[ExpInst_d]], factory:InstructionFactory_p, opts:dict) -> Maybe[ExpInst_d]: ...
298
[docs]
299 def exp_coerce_h(self, inst:ExpInst_d, factory:InstructionFactory_p, opts:dict) -> Maybe[ExpInst_d]: ...
300
[docs]
301 def exp_final_h(self, inst:ExpInst_d, root:Maybe[ExpInst_d], factory:InstructionFactory_p, opts:dict) -> Maybe[ExpInst_d]: ...
302
[docs]
303 def exp_check_result_h(self, inst:Maybe[ExpInst_d], opts:dict) -> None: ...
304
[docs]
305class Expandable_p(Protocol):
306 """ An expandable, like a DKey,
307 uses these hooks to customise the expansion
308 """
309
[docs]
310 def expand(self, *sources, **kwargs) -> Maybe: ...