Coverage for sygaldry / codegen.py: 82%
339 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-10 13:02 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import re
6from dataclasses import dataclass, field
7from typing import Any, Optional
9from .types import RESERVED_KEYS
11_NON_IDENTIFIER_RE = re.compile(r"[^a-zA-Z0-9_]")
14@dataclass(frozen=True)
15class SourceMapping:
16 """
17 Map a generated line number to a config path.
18 """
20 line: int
21 config_path: str
24@dataclass
25class ComponentEntry:
26 """
27 A typed component extracted from the config.
28 """
30 config_path: str
31 var_name: str
32 import_path: str
33 args: list[Any] = field(default_factory=list)
34 kwargs: dict[str, Any] = field(default_factory=dict)
35 is_func: bool = False
38@dataclass
39class PlainEntry:
40 """
41 A plain (non-typed) config value.
42 """
44 config_path: str
45 var_name: str
46 value: Any
49@dataclass
50class RefExpression:
51 """
52 A reference expression to be emitted as a variable or attribute access.
53 """
55 target: str
58@dataclass
59class EntriesExpression:
60 """
61 A dict built from _entries.
62 """
64 items: list[tuple[Any, Any]]
67@dataclass
68class AnalysisResult:
69 """
70 Result of analyzing a config dict.
71 """
73 components: list[ComponentEntry] = field(default_factory=list)
74 plains: list[PlainEntry] = field(default_factory=list)
75 topological_order: list[str] = field(default_factory=list)
76 imports: dict[str, tuple[str, str]] = field(default_factory=dict)
79def _to_identifier(key: str) -> str:
80 """
81 Convert a config key to a valid Python identifier.
82 """
83 if result := _NON_IDENTIFIER_RE.sub("_", key):
84 if result[0].isdigit():
85 result = f"_{result}"
86 return result
87 else:
88 return "_unnamed"
91def _split_import_path(dotted: str) -> tuple[str, str]:
92 """
93 Split 'a.b.ClassName' into ('a.b', 'ClassName').
94 """
95 parts = dotted.rsplit(".", 1)
96 if len(parts) == 1:
97 return parts[0], parts[0]
98 else:
99 return parts[0], parts[1]
102class ConfigAnalyzer:
103 """
104 Walk a loaded config dict and extract typed component specs.
105 """
107 def __init__(self, config: dict[str, Any]) -> None:
108 self._config = config
109 self._components: list[ComponentEntry] = list()
110 self._plains: list[PlainEntry] = list()
111 self._imports: dict[str, tuple[str, str]] = dict()
112 self._deps: dict[str, list[str]] = dict()
113 self._variable_names: dict[str, str] = dict()
114 self._anon_counter = 0
115 self._used_names: set[str] = set()
117 def analyze(self) -> AnalysisResult:
118 """
119 Analyze the config and return structured results.
120 """
121 for key, value in self._config.items():
122 self._analyze_top_level(key, value)
123 order = self._topological_sort()
124 result = AnalysisResult(
125 components=self._components,
126 plains=self._plains,
127 topological_order=order,
128 imports=self._imports,
129 )
130 return result
132 def _alloc_name(self, preferred: str) -> str:
133 """
134 Allocate a unique variable name.
135 """
136 name = _to_identifier(preferred)
137 if name not in self._used_names: 137 ↛ 141line 137 didn't jump to line 141 because the condition on line 137 was always true
138 self._used_names.add(name)
139 return name
140 else:
141 counter = 0
142 while f"{name}_{counter}" in self._used_names:
143 counter += 1
144 unique = f"{name}_{counter}"
145 self._used_names.add(unique)
146 return unique
148 def _alloc_anon(self, hint: str) -> str:
149 """
150 Allocate an anonymous variable name.
151 """
152 name = f"__sygaldry_{_to_identifier(hint)}_{self._anon_counter}"
153 self._anon_counter += 1
154 self._used_names.add(name)
155 return name
157 def _analyze_top_level(self, key: str, value: Any) -> None:
158 """
159 Analyze a single top-level config entry.
160 """
161 variable_name = self._alloc_name(key)
162 self._variable_names[key] = variable_name
163 self._deps[key] = list()
165 if isinstance(value, dict):
166 if "_ref" in value:
167 ref_target = value["_ref"]
168 top = ref_target.split(".")[0]
169 self._deps[key].append(top)
170 self._plains.append(
171 PlainEntry(
172 config_path=key,
173 var_name=variable_name,
174 value=RefExpression(target=ref_target),
175 )
176 )
177 elif "_type" in value:
178 self._analyze_type_entry(key, variable_name, value)
179 elif "_func" in value:
180 self._analyze_func_entry(key, variable_name, value)
181 elif value.get("_deep") is False:
182 self._plains.append(
183 PlainEntry(config_path=key, var_name=variable_name, value=...)
184 )
185 elif "_entries" in value:
186 entries_expr = self._analyze_entries(value["_entries"], key, self._deps[key])
187 self._plains.append(
188 PlainEntry(config_path=key, var_name=variable_name, value=entries_expr)
189 )
190 else:
191 self._plains.append(
192 PlainEntry(config_path=key, var_name=variable_name, value=value)
193 )
194 else:
195 self._plains.append(
196 PlainEntry(config_path=key, var_name=variable_name, value=value)
197 )
199 def _analyze_type_entry(
200 self, config_path: str, var_name: str, entry: dict[str, Any]
201 ) -> None:
202 """
203 Analyze a _type entry.
204 """
205 import_path = entry["_type"]
206 mod, name = _split_import_path(import_path)
207 self._imports[import_path] = (mod, name)
209 raw_args = entry.get("_args", [])
210 args = list()
211 for idx, arg in enumerate(raw_args):
212 processed = self._process_arg_value(
213 arg, f"{config_path}._args.{idx}", self._deps.setdefault(config_path, [])
214 )
215 args.append(processed)
217 kwargs: dict[str, Any] = dict()
218 for key, value in entry.items():
219 if key in RESERVED_KEYS:
220 continue
221 processed = self._process_arg_value(
222 value, f"{config_path}.{key}", self._deps.setdefault(config_path, [])
223 )
224 kwargs[key] = processed
226 raw_kwargs = entry.get("_kwargs", {})
227 for key, value in raw_kwargs.items():
228 processed = self._process_arg_value(
229 value, f"{config_path}._kwargs.{key}", self._deps.setdefault(config_path, [])
230 )
231 kwargs[key] = processed
233 self._components.append(
234 ComponentEntry(
235 config_path=config_path,
236 var_name=var_name,
237 import_path=import_path,
238 args=args,
239 kwargs=kwargs,
240 )
241 )
243 def _analyze_func_entry(
244 self, config_path: str, var_name: str, value: dict[str, Any]
245 ) -> None:
246 """
247 Analyze a _func entry.
248 """
249 import_path = value["_func"]
250 module, name = _split_import_path(import_path)
251 self._imports[import_path] = (module, name)
252 component = ComponentEntry(
253 config_path=config_path,
254 var_name=var_name,
255 import_path=import_path,
256 is_func=True,
257 )
258 self._components.append(component)
260 def _process_arg_value(self, value: Any, config_path: str, deps: list[str]) -> Any:
261 """
262 Process a value that appears as an argument to a _type constructor.
264 Returns the value as-is for scalars, as a RefExpression for _ref,
265 or hoists nested _type entries to intermediate variables.
266 """
267 if not isinstance(value, dict):
268 if isinstance(value, list):
269 return [
270 self._process_arg_value(item, f"{config_path}.{idx}", deps)
271 for idx, item in enumerate(value)
272 ]
273 return value
275 if "_ref" in value:
276 ref_target = value["_ref"]
277 top = ref_target.split(".")[0]
278 deps.append(top)
279 return RefExpression(target=ref_target)
281 elif "_type" in value: 281 ↛ 286line 281 didn't jump to line 286 because the condition on line 281 was always true
282 anon_name = self._alloc_anon(config_path.replace(".", "_"))
283 self._analyze_type_entry(config_path, anon_name, value)
284 return RefExpression(target=anon_name)
286 elif "_func" in value:
287 anon_name = self._alloc_anon(config_path.replace(".", "_"))
288 self._analyze_func_entry(config_path, anon_name, value)
289 return RefExpression(target=anon_name)
291 elif value.get("_deep") is False:
292 return ...
294 elif "_entries" in value:
295 return self._analyze_entries(value["_entries"], config_path, deps)
297 else:
298 return value
300 def _analyze_entries(
301 self, entries: Any, config_path: str, deps: list[str]
302 ) -> EntriesExpression:
303 """
304 Analyze an _entries list.
305 """
306 items: list[tuple[Any, Any]] = list()
307 if not isinstance(entries, list): 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 return EntriesExpression(items=[])
309 for idx, entry in enumerate(entries):
310 if not isinstance(entry, dict): 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 continue
312 key = self._process_arg_value(entry.get("_key"), f"{config_path}.{idx}._key", deps)
313 value = self._process_arg_value(
314 entry.get("_value"), f"{config_path}.{idx}._value", deps
315 )
316 item = (key, value)
317 items.append(item)
318 return EntriesExpression(items=items)
320 def _topological_sort(self) -> list[str]:
321 """
322 Sort entries so that dependencies come before dependents.
323 """
324 visited: set[str] = set()
325 order: list[str] = list()
326 visiting: set[str] = set()
328 keys = {component.config_path for component in self._components}
329 keys.update({plain.config_path for plain in self._plains})
331 def visit(key_: str) -> None:
332 # What the absolute fuck?:w
333 if key_ in visited or key_ in visiting:
334 return
336 visiting.add(key_)
337 for dep in self._deps.get(key_, []):
338 if dep in keys: 338 ↛ 337line 338 didn't jump to line 337 because the condition on line 338 was always true
339 visit(dep)
340 visiting.discard(key_)
341 visited.add(key_)
342 order.append(key_)
344 for key in keys:
345 visit(key)
347 return order
350class CodeGenerator:
351 """
352 Generate Python source code from analysis results.
353 """
355 def __init__(self, analysis: AnalysisResult, *, line_length: int = 99) -> None:
356 self._analysis = analysis
357 self._line_length = line_length
358 self._lines: list[str] = list()
359 self._mappings: list[SourceMapping] = list()
360 self._entry_map: dict[str, Any] = dict()
362 for component in analysis.components:
363 self._entry_map[component.config_path] = component
364 for plain in analysis.plains:
365 self._entry_map[plain.config_path] = plain
366 for component in analysis.components:
367 self._entry_map[component.var_name] = component
368 for plain in analysis.plains:
369 self._entry_map[plain.var_name] = plain
371 def generate(
372 self,
373 call_target: str | None = None,
374 call_method: str | None = None,
375 call_args: list[Any] | None = None,
376 ) -> tuple[str, list[SourceMapping]]:
377 """
378 Generate the full source code.
379 """
380 self._emit_header()
381 self._emit_imports()
382 self._emit_blank()
383 self._emit_entries()
384 if call_target is not None: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true
385 self._emit_blank()
386 self._emit_call(call_target, call_method, call_args or [])
387 return "\n".join(self._lines) + "\n", list(self._mappings)
389 def _emit(self, line: str, config_path: str | None = None) -> None:
390 """
391 Emit a line of code, optionally recording its config path.
392 """
393 self._lines.append(line)
394 if config_path is not None:
395 mapping = SourceMapping(line=len(self._lines), config_path=config_path)
396 self._mappings.append(mapping)
398 def _emit_blank(self) -> None:
399 self._lines.append("")
401 def _emit_header(self) -> None:
402 self._emit("# Auto-generated by sygaldry for type checking — do not edit")
403 self._emit("from typing import Any")
405 def _emit_imports(self) -> None:
406 seen: set[str] = set()
407 for _, (mod, name) in sorted(self._analysis.imports.items(), key=lambda x: x[0]):
408 statement = f"from {mod} import {name}"
409 if statement not in seen: 409 ↛ 407line 409 didn't jump to line 407 because the condition on line 409 was always true
410 seen.add(statement)
411 self._emit(statement)
413 def _emit_entries(self) -> None:
414 """
415 Emit all entries in topological order.
416 """
417 emitted: set[str] = set()
418 for config_path in self._analysis.topological_order:
419 if config_path in emitted: 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 continue
421 entry = self._entry_map.get(config_path)
422 if entry is None: 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 continue
425 emitted.add(config_path)
427 if isinstance(entry, ComponentEntry):
428 self._emit_component(entry)
429 elif isinstance(entry, PlainEntry): 429 ↛ 418line 429 didn't jump to line 418 because the condition on line 429 was always true
430 self._emit_plain(entry)
432 def _emit_call(
433 self,
434 target: str,
435 method: str | None,
436 args: list[Any],
437 ) -> None:
438 """
439 Emit a call expression at the end of the generated code.
440 """
441 var_name = _to_identifier(target)
442 if entry := self._entry_map.get(target):
443 if isinstance(entry, (ComponentEntry, PlainEntry)):
444 var_name = entry.var_name
446 callee = f"{var_name}.{method}" if method else var_name
447 arg_parts = [self._value_to_source(arg) for arg in args]
448 args_str = ", ".join(arg_parts)
449 one_line = f"{callee}({args_str})"
450 if len(one_line) <= self._line_length or not arg_parts:
451 self._emit(one_line)
452 else:
453 self._emit(f"{callee}(")
454 for arg_part in enumerate(arg_parts):
455 self._emit(f" {arg_part}{suffix}")
456 self._emit(")")
458 def _emit_component(self, comp: ComponentEntry) -> None:
459 """
460 Emit a component (class instantiation or func import).
461 """
462 _, name = _split_import_path(comp.import_path)
463 if comp.is_func:
464 self._emit(
465 f"{comp.var_name} = {name} # config: {comp.config_path}",
466 config_path=comp.config_path,
467 )
468 return
470 parts = [self._value_to_source(arg) for arg in comp.args]
471 for key, value in comp.kwargs.items():
472 parts.append(f"{key}={self._value_to_source(value)}")
474 comment = f" # config: {comp.config_path}"
475 args_str = ", ".join(parts)
476 one_line = f"{comp.var_name}: {name} = {name}({args_str}){comment}"
477 if len(one_line) <= self._line_length or not parts: 477 ↛ 480line 477 didn't jump to line 480 because the condition on line 477 was always true
478 self._emit(one_line, config_path=comp.config_path)
479 else:
480 self._emit(f"{comp.var_name}: {name} = {name}({comment}", config_path=comp.config_path)
481 for part in parts:
482 self._emit(f" {part},")
483 self._emit(")")
485 def _emit_plain(self, plain: PlainEntry) -> None:
486 """
487 Emit a plain value assignment.
488 """
489 comment = f" # config: {plain.config_path}"
490 if isinstance(plain.value, RefExpression):
491 target = self._resolve_ref_source(plain.value.target)
492 self._emit(
493 f"{plain.var_name} = {target}{comment}",
494 config_path=plain.config_path,
495 )
496 elif plain.value is ...:
497 self._emit(
498 f"{plain.var_name}: Any = ...{comment}",
499 config_path=plain.config_path,
500 )
501 elif isinstance(plain.value, EntriesExpression):
502 items = [
503 (self._value_to_source(key), self._value_to_source(value))
504 for key, value in plain.value.items
505 ]
506 items_str = ", ".join(f"{k}: {v}" for k, v in items)
507 one_line = f"{plain.var_name}: dict = {{{items_str}}}{comment}"
508 if len(one_line) <= self._line_length or not items: 508 ↛ 511line 508 didn't jump to line 511 because the condition on line 508 was always true
509 self._emit(one_line, config_path=plain.config_path)
510 else:
511 self._emit(f"{plain.var_name}: dict = {{{comment}", config_path=plain.config_path)
512 for i, (k, v) in enumerate(items):
513 suffix = ","
514 self._emit(f" {k}: {v}{suffix}")
515 self._emit("}")
516 else:
517 type_name = self._python_type_name(plain.value)
518 self._emit(
519 f"{plain.var_name}: {type_name} = {self._value_to_source(plain.value)}{comment}",
520 config_path=plain.config_path,
521 )
523 def _resolve_ref_source(self, target: str) -> str:
524 """
525 Convert a ref target to a Python expression.
526 """
527 top, _, remainder = target.partition(".")
528 if entry := self._entry_map.get(top): 528 ↛ 534line 528 didn't jump to line 534 because the condition on line 528 was always true
529 if isinstance(entry, (ComponentEntry, PlainEntry)): 529 ↛ 532line 529 didn't jump to line 532 because the condition on line 529 was always true
530 var_name = entry.var_name
531 else:
532 var_name = _to_identifier(top)
533 else:
534 var_name = _to_identifier(top)
535 if remainder:
536 return f"{var_name}.{remainder}"
537 return var_name
539 def _value_to_source(self, value: Any) -> str:
540 """
541 Convert a config value to its Python source representation.
542 """
543 if isinstance(value, RefExpression):
544 return self._resolve_ref_source(value.target)
545 elif isinstance(value, EntriesExpression): 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true
546 items_str = ", ".join(
547 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
548 for k, v in value.items
549 )
550 return f"{{{items_str}}}"
551 elif value is ...: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true
552 return "..."
553 elif value is None:
554 return "None"
555 elif isinstance(value, bool):
556 return "True" if value else "False"
557 elif isinstance(value, (float, int, str)):
558 return repr(value)
559 elif isinstance(value, list):
560 items = ", ".join(self._value_to_source(v) for v in value)
561 return f"[{items}]"
562 elif isinstance(value, dict): 562 ↛ 569line 562 didn't jump to line 569 because the condition on line 562 was always true
563 items = ", ".join(
564 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
565 for k, v in value.items()
566 )
567 return f"{{{items}}}"
568 else:
569 return repr(value)
571 @staticmethod
572 def _python_type_name(value: Any) -> str:
573 """
574 Get a type annotation name for a plain value.
575 """
576 if value is None:
577 return "None"
578 elif isinstance(value, bool):
579 return "bool"
580 elif isinstance(value, int):
581 return "int"
582 elif isinstance(value, float):
583 return "float"
584 elif isinstance(value, str):
585 return "str"
586 elif isinstance(value, list):
587 return "list"
588 elif isinstance(value, dict): 588 ↛ 591line 588 didn't jump to line 591 because the condition on line 588 was always true
589 return "dict"
590 else:
591 return "Any"
594def generate_check_source(
595 config: dict[str, Any],
596 *,
597 line_length: int = 99,
598 call_target: str | None = None,
599 call_method: str | None = None,
600 call_args: list[Any] | None = None,
601) -> tuple[str, list[SourceMapping]]:
602 """
603 Generate type-checkable Python source from a sygaldry config.
605 :param config: Loaded, interpolated config mapping.
606 :param line_length: Maximum line length for generated code.
607 :param call_target: Top-level config key to emit a call expression for.
608 :param call_method: Method name to call on the target.
609 :param call_args: Arguments for the call expression.
610 :type config: dict
611 :type line_length: int
612 :returns: ``(source_code, mappings)`` tuple.
613 """
614 analyzer = ConfigAnalyzer(config)
615 result = analyzer.analyze()
616 generator = CodeGenerator(result, line_length=line_length)
617 return generator.generate(
618 call_target=call_target,
619 call_method=call_method,
620 call_args=call_args,
621 )