Repository navigation
Expand file tree
/
Copy pathboard.py
More file actions
1667 lines (1503 loc) · 75.1 KB
/
Copy pathboard.py
File metadata and controls
1667 lines (1503 loc) · 75.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""scripts/board.py — the PyAutoNerves Board: every config file and option.
A **read-only browser** of the configuration layer the Nerves own: every YAML
file under each library's ``<package>/config/`` and under each workspace's
``config/`` that overrides them, with its keys, its comments (the comments
*are* the option docs), its source and a link to it on GitHub. Nothing on the
board acts; it is a map, grown later if it needs to be.
**What it collects** (``collect()`` — the only I/O, degrading per source into
``errors[]``):
* every ``*.yaml`` / ``*.yml`` file of every source in ``SOURCES``: its line
count, ``yaml.safe_load`` result (or the parse error), the top-level keys,
every key path (dotted) with its line number and comment — the inline
``# …`` on the key's own line, else the comment block directly above it;
* **prior files** (under ``priors/``, or every top-level value a mapping of
``param → {type, …}``) as table rows — Class · param · type · mean/sigma
or lower/upper · width modifier · limits;
* **the override map**: autonerves resolves a key workspace → last-imported
library → … → PyAutoFit (``conf.Config.push(keep_first=True)``, keys
lowercased), so every workspace file is looked up across that
workspace's library stack in the same order — by relative path, where a
library ``X.yaml``'s section ``f`` is the workspace's ``X/f.yaml`` and the
other way round, as ``RecursiveConfig`` merges them (prior files: same
path only). The *counterpart* is the first library holding the file;
the diff is against the merged stack (the value autonerves would fall
back to): keys whose value differs, keys only the workspace sets
(**orphans** — no library defines them), keys only the stack sets. The
version-handshake keys autonerves reads itself (``NERVES_OWNED``) are
*owned*, never orphans. ``build/*.yaml`` is workspace tooling (CI/build
lists, not library settings) and is grouped apart;
* **possibly unused library keys**: every library package's ``.py`` (and
``autonerves/``) is scanned for config lookups (``scan_lookups``) and each
key of a library settings file is classed ``used`` / ``section-read`` /
``unused`` against the reads of the whole stack (``classify_files``) —
static and untrusted, so it only ever adds ``info`` feed items;
* the ``PYAUTO_*`` environment variables read by ``autonerves/test_mode.py``,
``workspace.py`` and ``__init__.py`` — the non-YAML options the Nerves also
own — each with the comment above it (or its function's docstring).
**Where it reads from.** Locally, each source resolves through the body map
(``PyAutoMind/repos.yaml`` ``path:``) under ``--root``; in the workflow,
``--sources DIR`` points at sparse clones laid out ``DIR/<Repo>/<config dir>``
(and ``DIR/<Repo>/<package>`` for a library's code).
GitHub slugs come from the body map's ``github:``; the Nerves' own owner comes
from ``git remote`` (the tenant firewall: no owner is written here).
**Shape** (mirrors ``PyAutoGut/scripts/board.py``): ``render(snapshot, fmt)``
is pure — ``md | md-brief | json | badge | state | html-index | html-repo`` —
and ``--site DIR`` writes the whole Pages site (``index.html``,
``repos/<Repo>.html``, ``badge.json``, ``board.json``, ``state.json``,
``dashboard.md``). This script is NOT part of the ``autonerves`` package.
Usage:
python scripts/board.py [--brain P] [--mind P] [--root P] [--sources DIR]
(--collect OUT | --snapshot F)
(--md|--md-brief|--json|--badge|--state|
--html-index|--html-repo NAME|--site DIR)
"""
from __future__ import annotations
import ast
import datetime
import html as _html
import json
import os
import re
import subprocess
import sys
import warnings
from pathlib import Path
import yaml
NERVES_HOME = Path(__file__).resolve().parents[1]
CHECKIN_PROMPT = (
"Use this chat as an ongoing place to understand and improve configuration across "
"PyAutoLabs. Read PyAutoNerves/AGENTS.md and the current Nerves board, then inspect the "
"relevant source repositories and configuration evidence. Check scan freshness and "
"coverage, keeping unavailable sources explicit.\n\n"
"When I give no particular direction, review configuration sources, workspace overrides, "
"YAML parse errors, orphan keys, possibly unused options, environment variables and "
"documentation gaps. Summarize the findings that need attention and recommend priorities "
"with reasons.\n\n"
"When I name an option, file, workspace or question, make that the main focus. Help me "
"find a setting, explain its purpose and default, trace where it is read, understand "
"override precedence or establish which value applies in a particular context. Bring in "
"related configuration where relevant; do not repeat the full board review on every "
"follow-up.\n\n"
"Ground explanations in configuration definitions and the code that consumes them. "
"Distinguish what static scanning suggests from behavior verified at runtime. Treat "
"possibly unused keys as investigation leads, and check compatibility and downstream use "
"before proposing removal or renaming.\n\n"
"Help me plan configuration changes, improve option documentation, resolve inconsistent "
"defaults or investigate serialization and version-handshake problems. Identify affected "
"libraries and workspaces, explain migration needs and define how the resulting behavior "
"should be verified.\n\n"
"Route accepted changes through the development workflow, checking existing tasks and "
"claims first. Carry clearly authorized work through its procedure, retaining approvals "
"already given in this conversation. Preserve downstream API validation and applicable "
"approval requirements.\n\n"
"After changes, refresh the relevant evidence and report what changed, what was verified "
"and what remains unresolved. Keep this conversation available for subsequent "
"configuration questions and work."
)
BOARD_KEY = "nerves" # this board's entry in the Brain's palette table
SCHEMA_VERSION = 1
STATE_ITEMS_MAX = 20
# The config sources, in one table. `stack` is the lookup order autonerves
# uses behind a workspace (the last-imported library first, PyAutoFit last),
# so a workspace file is compared against what the libraries would give.
# `package` is a library's source dir, scanned for the config keys its code
# reads (the "possibly unused keys" flag).
SOURCES = (
{"repo": "PyAutoFit", "config": "autofit/config", "kind": "library",
"package": "autofit", "stack": ()},
{"repo": "PyAutoArray", "config": "autoarray/config", "kind": "library",
"package": "autoarray", "stack": ()},
{"repo": "PyAutoGalaxy", "config": "autogalaxy/config", "kind": "library",
"package": "autogalaxy", "stack": ()},
{"repo": "PyAutoLens", "config": "autolens/config", "kind": "library",
"package": "autolens", "stack": ()},
{"repo": "PyAutoCTI", "config": "autocti/config", "kind": "library",
"package": "autocti", "stack": ()},
{"repo": "autofit_workspace", "config": "config", "kind": "workspace",
"stack": ("PyAutoFit",)},
{"repo": "autogalaxy_workspace", "config": "config", "kind": "workspace",
"stack": ("PyAutoGalaxy", "PyAutoArray", "PyAutoFit")},
{"repo": "autolens_workspace", "config": "config", "kind": "workspace",
"stack": ("PyAutoLens", "PyAutoGalaxy", "PyAutoArray", "PyAutoFit")},
{"repo": "autocti_workspace", "config": "config", "kind": "workspace",
"stack": ("PyAutoCTI", "PyAutoArray", "PyAutoFit")},
)
# Workspace files that are tooling (build/CI lists), not library settings.
TOOLING_DIRS = ("build",)
YAML_SUFFIXES = (".yaml", ".yml")
ENV_MODULES = ("autonerves/test_mode.py", "autonerves/workspace.py",
"autonerves/__init__.py")
ENV_RE = re.compile(r"\bPYAUTO_[A-Z0-9_]*[A-Z0-9]\b")
# --- locating the Brain and the Mind ------------------------------------------
_THEME_BRAIN: list = [None]
def _brain_home(brain: str | None = None) -> Path | None:
for cand in (brain, os.environ.get("PYAUTO_BRAIN"),
NERVES_HOME / "PyAutoBrain", NERVES_HOME.parent / "PyAutoBrain"):
if cand and (Path(cand) / "board" / "_theme.py").is_file():
return Path(cand)
return None
def theme():
"""The shared theme module (PyAutoBrain/board/_theme.py). Only the html
surfaces need it; every other surface renders with no Brain in reach."""
home = _brain_home(_THEME_BRAIN[0])
board = home / "board" if home else None
if board and (board / "_theme.py").is_file():
if str(board) not in sys.path:
sys.path.insert(0, str(board))
import _theme
return _theme
raise RuntimeError(
"the shared board theme (PyAutoBrain/board/_theme.py) is not in reach "
"— check PyAutoBrain out beside this repo or pass --brain")
def _repos_yaml(root: Path, mind: str | None = None) -> Path | None:
for cand in (mind, os.environ.get("PYAUTO_MIND") if not mind else None,
root / "PyAutoMind", root / "organs" / "PyAutoMind"):
if cand and (Path(cand) / "repos.yaml").is_file():
return Path(cand) / "repos.yaml"
return None
def load_body_map(root: Path, mind: str | None = None) -> dict:
"""``{repo: {path, github, …}}`` from the Mind's body map, or ``{}``."""
path = _repos_yaml(root, mind)
if not path:
return {}
try:
return (yaml.safe_load(path.read_text(encoding="utf-8")) or {}) \
.get("repos") or {}
except (OSError, yaml.YAMLError):
return {}
# --- identity (derived, never hardcoded — tenant firewall) -------------------
def parse_owner_repo(url: str) -> tuple[str, str]:
m = re.search(r"github\.com[:/]+([^/]+)/([^/]+?)(?:\.git)?/?$", url.strip())
return (m.group(1), m.group(2)) if m else ("", "")
def _self_owner_repo() -> tuple[str, str]:
slug = os.environ.get("GITHUB_REPOSITORY", "")
if "/" in slug:
o, r = slug.split("/", 1)
return o, r
try:
out = subprocess.run(["git", "-C", str(NERVES_HOME), "remote",
"get-url", "origin"], capture_output=True,
text=True, timeout=10).stdout
except (OSError, subprocess.SubprocessError):
out = ""
return parse_owner_repo(out)
# --- pure YAML reading ----------------------------------------------------------
_KEY_RE = re.compile(
r"^(?P<ind>[ ]*)(?P<dash>-[ ]+)?"
r"(?P<key>\"[^\"]*\"|'[^']*'|[^\s#:\-\[\]{}][^#:]*?|[\-][^\s#:][^#:]*?)"
r"[ \t]*:(?=[ \t]|$)(?P<rest>.*)$")
_BLOCK_SCALAR = re.compile(r"^\s*[|>][+-]?\d*\s*(#.*)?$")
def _strip_quotes(text: str) -> str:
return re.sub(r"\"[^\"]*\"|'[^']*'", lambda m: " " * len(m.group(0)), text)
def split_comment(line: str) -> tuple[str, str]:
"""``(code, comment)`` for one YAML line — the comment is the text after a
``#`` that starts the line or follows whitespace, outside quotes."""
masked = _strip_quotes(line)
m = re.search(r"(^|\s)#", masked)
if not m:
return line, ""
at = m.start() + len(m.group(1))
return line[:at], line[at:]
def _clip(text: object, limit: int = 160) -> str:
text = " ".join(str(text or "").split())
return text if len(text) <= limit else text[: limit - 1] + "…"
def _comment_text(raw: str) -> str:
return raw.lstrip("#").strip()
def scan_keys(text: str) -> list[dict]:
"""Every mapping key in a YAML file, in order: ``{k, line, c}`` — ``k`` the
dotted path, ``c`` the key's comment (its inline ``#``, else the comment
lines directly above it). A line scan, so it keeps what ``safe_load``
throws away: line numbers and comments."""
out = []
stack: list[tuple[int, str]] = []
above: list[str] = []
block_indent = None
for n, raw in enumerate(text.splitlines(), start=1):
line = raw.rstrip("\r")
stripped = line.strip()
indent = len(line) - len(line.lstrip(" "))
if block_indent is not None:
if stripped and indent <= block_indent:
block_indent = None
else:
continue
if not stripped:
above = []
continue
if stripped.startswith("#"):
above.append(_comment_text(stripped))
continue
if stripped in ("---", "..."):
above = []
continue
m = _KEY_RE.match(line)
if not m:
above = []
continue
key = m.group("key").strip().strip("\"'")
ind = len(m.group("ind")) + len(m.group("dash") or "")
while stack and stack[-1][0] >= ind:
stack.pop()
path = ".".join([k for _, k in stack] + [key])
code, comment = split_comment(m.group("rest"))
c = _comment_text(comment) if comment else " ".join(above[-3:])
out.append({"k": path, "line": n, "c": _clip(c, 200)})
stack.append((ind, key))
above = []
if _BLOCK_SCALAR.match(m.group("rest") or ""):
block_indent = ind
return out
def flatten(data, prefix: str = "", depth: int | None = None) -> dict:
"""Leaf key paths of parsed YAML, keys lowercased as autonerves reads
them; an empty mapping is itself a leaf. ``depth`` stops the descent
(a prior file compares whole ``Class.param`` specs, depth 2)."""
out = {}
stop = depth is not None and prefix.count(".") + 1 >= depth and prefix
if isinstance(data, dict) and data and not stop:
for k, v in data.items():
path = f"{prefix}.{str(k).lower()}" if prefix else str(k).lower()
out.update(flatten(v, path, depth))
elif prefix:
out[prefix] = data
return out
def _is_prior_param(v) -> bool:
return isinstance(v, dict) and "type" in v
def is_prior_file(rel: str, data) -> bool:
"""Under ``priors/``, or every top-level value a mapping of
``param → {type, …}``."""
if rel.startswith("priors/") or "/priors/" in rel:
return True
if not isinstance(data, dict) or not data:
return False
return all(isinstance(v, dict) and v and all(_is_prior_param(p)
for p in v.values())
for v in data.values())
def _num(v) -> str:
if isinstance(v, float) and v.is_integer() and abs(v) < 1e15:
return str(v)
return str(v)
def prior_rows(data, keys: list[dict]) -> list[dict]:
"""One row per ``Class.param`` of a prior file."""
lines = {k["k"]: k["line"] for k in keys}
rows = []
if not isinstance(data, dict):
return rows
for cls, params in data.items():
if not isinstance(params, dict):
continue
for param, spec in params.items():
if not isinstance(spec, dict):
continue
kind = str(spec.get("type", ""))
if "mean" in spec or "sigma" in spec:
a = f"mean {_num(spec.get('mean'))}"
b = f"σ {_num(spec.get('sigma'))}"
elif "lower_limit" in spec or "upper_limit" in spec:
a = f"lower {_num(spec.get('lower_limit'))}"
b = f"upper {_num(spec.get('upper_limit'))}"
elif "value" in spec:
a, b = f"value {_num(spec.get('value'))}", ""
else:
a, b = "", ""
wm = spec.get("width_modifier")
width = (f"{wm.get('type', '')} {_num(wm.get('value'))}".strip()
if isinstance(wm, dict) else ("" if wm is None else str(wm)))
lim = spec.get("limits")
limits = (f"[{_num(lim.get('lower'))}, {_num(lim.get('upper'))}]"
if isinstance(lim, dict) else "")
rows.append({"cls": str(cls), "param": str(param), "type": kind,
"a": a, "b": b, "width": width, "limits": limits,
"line": lines.get(f"{cls}.{param}", 0)})
return rows
def read_file(repo: str, rel: str, text: str, kind: str) -> dict:
"""One file's record — pure over its text."""
rec = {"repo": repo, "path": rel, "lines": len(text.splitlines()),
"text": text, "error": None, "top_keys": [], "keys": [],
"prior": False, "priors": [],
"tooling": kind == "workspace" and rel.split("/", 1)[0] in TOOLING_DIRS}
try:
data = yaml.safe_load(text)
except yaml.YAMLError as e:
rec["error"] = _clip(str(e).splitlines()[0] if str(e) else repr(e), 200)
mark = getattr(e, "problem_mark", None)
rec["error_line"] = (mark.line + 1) if mark is not None else 1
data = None
rec["keys"] = scan_keys(text)
if isinstance(data, dict):
rec["top_keys"] = [str(k) for k in data]
elif rec["error"] is None:
rec["top_keys"] = [k["k"] for k in rec["keys"] if "." not in k["k"]]
if rec["error"] is None and not rec["tooling"] and is_prior_file(rel, data):
rec["prior"] = True
rec["priors"] = prior_rows(data, rec["keys"])
rec["_data"] = data
return rec
# Keys autonerves itself reads from a workspace's config — the version
# handshake (``autonerves/workspace.py``, ``autonerves/__init__.py``). The
# Nerves ship no YAML, so no library defines them: owned, never orphans.
NERVES_OWNED = {
"general.yaml": ("version.workspace_version",
"version.minimum_library_version",
"version.workspace_version_check",
"version.python_version_check"),
}
_MISSING = object()
def _section(data, parts: list[str]):
"""``data`` descended through ``parts`` (keys lowercased, as autonerves
reads them), else ``_MISSING``."""
for p in parts:
if not isinstance(data, dict):
return _MISSING
data = next((v for k, v in data.items() if str(k).lower() == p),
_MISSING)
if data is _MISSING:
return _MISSING
return data
def resolve_in_library(lib_files: dict, rel: str) -> dict | None:
"""What one library gives for a workspace file ``rel`` (``lib_files`` is
``{path: record}``), resolved the way autonerves' ``RecursiveConfig``
walks a config dir: at each level of the path a ``<name>.yml``/``.yaml``
file shadows a ``<name>/`` directory, and a file's top-level sections are
the files of the directory form. So a library ``visualize.yaml``'s
``general:`` section is the workspace's ``visualize/general.yaml``, and a
library ``visualize/`` directory is a workspace ``visualize.yaml``.
Returns a stack record (``repo``, ``path`` — the library file, with a
``§ section`` when one — and ``_data``), or ``None``."""
# file stems reach autonerves lowercased (DictWrapper): match them so
files = {p.lower(): r for p, r in lib_files.items()}
parts = rel.rsplit(".", 1)[0].lower().split("/")
for i in range(len(parts)):
stem = "/".join(parts[:i + 1])
rec = next((files[stem + s] for s in YAML_SUFFIXES[::-1]
if stem + s in files), None)
if rec is not None:
if rec["error"]:
return None
data = _section(rec.get("_data"), parts[i + 1:])
if data is _MISSING:
return None
where = rec["path"] + (f" § {'.'.join(parts[i + 1:])}"
if parts[i + 1:] else "")
return {"repo": rec["repo"], "path": where, "_data": data}
if not any(p.startswith(stem + "/") for p in files):
return None
# the whole path is a library directory: each entry is a section
children = set()
for p in files:
if p.startswith(stem + "/"):
head, _, tail = p[len(stem) + 1:].partition("/")
children.add(head if tail else head.rsplit(".", 1)[0])
data = {}
for child in sorted(children - {"priors"}): # RecursiveConfig skips priors
sub = resolve_in_library(lib_files, f"{stem}/{child}.yaml")
if sub is not None:
data[child] = sub["_data"]
if not data:
return None
repo = next(iter(files.values()))["repo"]
return {"repo": repo, "path": stem + "/", "_data": data}
def diff_against_stack(ws: dict, stack_recs: list[dict]) -> dict:
"""The override record for one workspace file against its library stack
(``stack_recs`` in autonerves lookup order, only the libraries that hold
it — by relative path, or through ``resolve_in_library``)."""
# A prior file is compared per parameter: a workspace that swaps a
# Gaussian for a Uniform changes one spec, it does not orphan the new
# prior's lower/upper fields.
depth = 2 if ws.get("prior") else None
wflat = flatten(ws.get("_data"), depth=depth)
merged: dict = {}
for rec in stack_recs:
for k, v in flatten(rec.get("_data"), depth=depth).items():
merged.setdefault(k, v)
# An empty mapping (`fit_imaging: {}`) sets nothing, so it can neither
# differ nor be orphaned.
wflat = {k: v for k, v in wflat.items() if v != {}}
lines = {k["k"].lower(): k["line"] for k in ws.get("keys") or []}
differs = sorted(k for k in wflat if k in merged and merged[k] != wflat[k])
ws_only = sorted(k for k in wflat if k not in merged)
owned = set(NERVES_OWNED.get(ws["path"], ()))
nerves_owned = [k for k in ws_only if k in owned]
ws_only = [k for k in ws_only if k not in owned]
lib_only = sorted(k for k in merged if k not in wflat)
first = stack_recs[0] if stack_recs else None
return {"repo": ws["repo"], "path": ws["path"],
"counterpart": first["repo"] if first else None,
"counterpart_path": first["path"] if first else None,
"stack": [r["repo"] for r in stack_recs],
"differs": differs, "workspace_only": ws_only,
"nerves_owned": nerves_owned,
"library_only": lib_only[:50], "library_only_count": len(lib_only),
"first_orphan_line": min((lines.get(k, 0) for k in ws_only),
default=0) or 1}
def env_vars_from(texts: dict) -> list[dict]:
"""``PYAUTO_*`` names from module texts (``{relpath: text}``), each with
the ``#`` comment lines directly above a mention, else the first
paragraph of the enclosing function's docstring — the first mention that
has either."""
seen: dict = {}
for rel, text in texts.items():
lines = text.splitlines()
funcs = []
try:
for node in ast.walk(ast.parse(text)):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
doc = ast.get_docstring(node) or ""
funcs.append((node.lineno, getattr(node, "end_lineno",
node.lineno), doc))
except SyntaxError:
pass
for n, line in enumerate(lines, start=1):
for name in ENV_RE.findall(line):
if name in seen:
if rel not in seen[name]["modules"]:
seen[name]["modules"].append(rel)
if seen[name]["comment"]:
continue
comment = ""
j = n - 2
block = []
while j >= 0 and lines[j].strip().startswith("#"):
block.insert(0, lines[j].strip().lstrip("#").strip())
j -= 1
if block:
comment = " ".join(block)
else:
inner = [f for f in funcs if f[0] <= n <= f[1] and f[2]]
if inner:
doc = min(inner, key=lambda f: f[1] - f[0])[2]
para = doc.strip().split("\n\n")[0]
comment = " ".join(para.split())
if name in seen:
# the first mention had nothing to say (e.g. a bare
# constant); a later one inside a documented function does
seen[name]["comment"] = _clip(comment, 220)
continue
seen[name] = {"name": name, "modules": [rel], "line": n,
"comment": _clip(comment, 220)}
return sorted(seen.values(), key=lambda e: e["name"])
# --- config lookups in library code (the "not in use anymore" scan) ------------
# autonerves resolves ``conf.instance["general"]["output"]["remove_files"]`` as
# file ``general.yaml`` → section ``output`` → key ``remove_files`` (keys
# lowercased, every layer merged), so a read is a dotted path whose head is the
# config file's path without its suffix.
#
# Helpers that take a key name and read it under a fixed prefix — a literal
# first argument is a read of ``prefix.<arg>``.
HELPER_READS = {"should_output": ("output",)}
# Files autonerves loads whole outside the subscript API
# (``Config.logging_config`` opens ``logging.yaml`` and hands it to
# ``logging.config.dictConfig``) — every key in them is a section read.
WHOLESALE_FILES = ("logging",)
# Config lookups are only counted once every library's package was scanned:
# a partial scan would call every key of the missing library unused.
LOOKUP_CLASSES = ("used", "section-read", "unused")
# autonerves' own package — a config reader too (swappable for tests).
NERVES_PACKAGE = NERVES_HOME / "autonerves"
def _is_conf_root(node, names: set) -> bool:
"""``conf.instance`` / ``x.conf.instance`` / ``conf.instance.dict`` /
a bare ``instance`` imported from ``autonerves.conf``."""
if isinstance(node, ast.Attribute) and node.attr == "dict":
node = node.value
if isinstance(node, ast.Attribute) and node.attr == "instance":
v = node.value
return (isinstance(v, ast.Name) and v.id == "conf") or \
(isinstance(v, ast.Attribute) and v.attr == "conf")
return isinstance(node, ast.Name) and node.id in names
def _literal(node):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value.lower()
return None
def _chain(node, roots: set, local: dict):
"""Unwind a subscript chain ending at ``node``: ``(keys, open)`` where
``keys`` is the literal path from the root and ``open`` says a non-literal
subscript ended it (everything under ``keys`` may be read); ``None`` when
the chain does not start at the config."""
steps = []
cur = node
while True:
if isinstance(cur, ast.Subscript):
steps.append(cur.slice)
cur = cur.value
elif (isinstance(cur, ast.Call) and isinstance(cur.func, ast.Attribute)
and cur.func.attr == "get" and cur.args):
steps.append(cur.args[0])
cur = cur.func.value
else:
break
if isinstance(cur, ast.Name) and cur.id in local:
base = list(local[cur.id])
elif _is_conf_root(cur, roots):
base = []
else:
return None
keys = base
for s in reversed(steps):
lit = _literal(s)
if lit is None:
return keys, True
keys.append(lit)
return keys, False
def _parents(tree) -> dict:
par = {}
for node in ast.walk(tree):
for child in ast.iter_child_nodes(node):
par[child] = node
return par
def _is_chain_link(node, parent) -> bool:
"""Whether ``node`` continues into ``parent`` as part of a longer chain."""
if isinstance(parent, ast.Subscript) and parent.value is node:
return True
if isinstance(parent, ast.Attribute) and parent.value is node and \
parent.attr in ("get", "dict"):
return True
return False
def lookups_from(text: str) -> tuple[set, set, set]:
"""``(reads, wildcards, bound)`` — the dotted config paths one module
reads.
* a literal subscript chain from ``conf.instance`` (``.get("k")`` counts as
a subscript) is a read of that path;
* a non-literal subscript ends the chain: its literal prefix becomes a
wildcard (anything under it may be read);
* ``x = conf.instance["a"]["b"]`` then ``x["c"]`` is followed within the
function (or module, or a closure inside it) — the assignment reads
``a.b`` itself (``bound``: the path is used, its keys are not covered),
each use of ``x`` reads ``a.b`` + its own literal keys, and a bare use
of ``x`` (passed on, iterated) reads ``a.b`` whole;
* ``HELPER_READS`` calls with a literal first argument.
Raises ``SyntaxError`` on a module that does not parse."""
tree = ast.parse(text)
roots = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and (node.module or "").endswith(
"conf") and (node.module or "").split(".")[0] in (
"autonerves", "autoconf"):
roots |= {a.asname or a.name for a in node.names
if a.name == "instance"}
par = _parents(tree)
reads, wild, bound = set(), set(), set()
funcs = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)
scopes = [n for n in ast.walk(tree) if isinstance(n, funcs)]
scopes.append(tree)
owned_by, own_locals, skip = {}, {}, set()
for scope in scopes:
# nodes owned by this scope (a nested def is its own scope)
owned, stack = [], list(ast.iter_child_nodes(scope))
while stack:
n = stack.pop()
if isinstance(n, funcs):
continue
owned.append(n)
stack.extend(ast.iter_child_nodes(n))
owned_by[scope] = owned
local: dict = {}
for n in owned:
if isinstance(n, ast.Assign) and len(n.targets) == 1 and \
isinstance(n.targets[0], ast.Name):
got = _chain(n.value, roots, {})
if got and not got[1] and got[0]:
local[n.targets[0].id] = tuple(got[0])
bound.add(".".join(got[0]))
skip.add(n.value)
own_locals[scope] = local
seen: set = set()
for scope in scopes:
# a closure sees the names bound in the functions around it
local, up = {}, par.get(scope)
chain_up = []
while up is not None:
if isinstance(up, funcs):
chain_up.append(up)
up = par.get(up)
for outer in reversed(chain_up):
local.update(own_locals[outer])
local.update(own_locals[scope])
owned = owned_by[scope]
for n in owned:
if n in seen or n in skip:
continue
if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and \
n.func.id in HELPER_READS and n.args:
lit = _literal(n.args[0])
if lit is not None:
reads.add(".".join(HELPER_READS[n.func.id] + (lit,)))
continue
if not isinstance(n, (ast.Subscript, ast.Call, ast.Name,
ast.Attribute)):
continue
if isinstance(n, ast.Name) and not isinstance(n.ctx, ast.Load):
continue
if _is_chain_link(n, par.get(n)):
continue
got = _chain(n, roots, local)
if not got:
continue
seen.add(n)
keys, open_ = got
if not keys:
if open_:
wild.add("")
continue
(wild if open_ else reads).add(".".join(keys))
return reads, wild, bound
def scan_lookups(pkg_dir) -> dict:
"""Every config read in a package's ``.py`` files:
``{files, reads, wildcards, bound, unparsed}`` (sorted lists)."""
reads, wild, bound, unparsed, n = set(), set(), set(), [], 0
for path in sorted(Path(pkg_dir).rglob("*.py")):
try:
text = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
unparsed.append(path.relative_to(pkg_dir).as_posix())
continue
if "instance" not in text and not any(h in text for h in HELPER_READS):
n += 1
continue
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore") # old escapes in library code
r, w, b = lookups_from(text)
except SyntaxError:
unparsed.append(path.relative_to(pkg_dir).as_posix())
continue
n += 1
reads |= r
wild |= w
bound |= b
return {"files": n, "reads": sorted(reads), "wildcards": sorted(wild),
"bound": sorted(bound), "unparsed": unparsed}
def config_path(rel: str, key: str) -> str:
"""A file key's full lookup path: ``visualize/general.yaml`` +
``general.backend`` → ``visualize.general.general.backend``."""
stem = rel.rsplit(".", 1)[0].replace("/", ".")
return f"{stem}.{key}".lower() if key else stem.lower()
def classify_key(path: str, reads: set, prefixes: set) -> str:
"""``used`` — the path is read (``reads``: exact reads, wildcard stems and
bound locals), or a read runs through it (a section one of whose keys is
read); ``section-read`` — a whole-section read or wildcard (``prefixes``)
covers one of its ancestors; ``unused`` — nothing references it."""
if path in reads:
return "used"
dotted = path + "."
if any(r.startswith(dotted) for r in reads):
return "used"
parts = path.split(".")
for i in range(len(parts) - 1, -1, -1):
if ".".join(parts[:i]) in prefixes:
return "section-read"
return "unused"
def classify_files(files: list[dict], lookups: dict) -> None:
"""Stamp ``use`` on every key of every library settings file (in place),
against the reads of the whole stack — autonerves merges every layer, so a
PyAutoFit key read by PyAutoLens is used."""
reads, prefixes = set(), set(WHOLESALE_FILES)
for lk in lookups.values():
# a chain that stops on a section reads it whole; a wildcard stem
# ``a.b[x]`` reads ``a.b`` and may read anything under it
prefixes |= set(lk["reads"]) | set(lk["wildcards"])
reads |= set(lk["reads"]) | set(lk["wildcards"]) | \
set(lk.get("bound") or ())
for f in files:
if not f.get("_library") or f.get("prior") or f.get("error") or \
f.get("tooling"):
continue
counts = dict.fromkeys(LOOKUP_CLASSES, 0)
for k in f["keys"]:
k["use"] = classify_key(config_path(f["path"], k["k"]), reads,
prefixes)
counts[k["use"]] += 1
f["use_counts"] = counts
# --- collection (the only I/O) --------------------------------------------------
def _source_dir(src: dict, root: Path, body: dict, sources: Path | None,
field: str = "config") -> Path | None:
sub = src.get(field)
if not sub:
return None
if sources is not None:
cand = sources / src["repo"] / sub
return cand if cand.is_dir() else None
rel = (body.get(src["repo"]) or {}).get("path")
for base in ((root / rel) if rel else None, root / src["repo"]):
if base and (base / sub).is_dir():
return base / sub
return None
def _now_z() -> str:
return datetime.datetime.now(datetime.timezone.utc) \
.strftime("%Y-%m-%dT%H:%M:%SZ")
def build_snapshot(files: list[dict], sources: list[dict], env: list[dict],
errors: list[str], owner: str = "", repo: str = "",
generated: str | None = None,
lookups: dict | None = None) -> dict:
"""Assemble the snapshot from read file records (pure). ``lookups``
(``{reader: scan_lookups(…)}``) classifies every library settings key —
only when every library's package was scanned (``None`` skips it)."""
if lookups is not None:
classify_files(files, lookups)
by_repo: dict = {}
for f in files:
by_repo.setdefault(f["repo"], {})[f["path"]] = f
overrides = []
for src in sources:
if src["kind"] != "workspace":
continue
for rel, rec in sorted((by_repo.get(src["repo"]) or {}).items()):
if rec["tooling"] or rec["error"]:
continue
if rec["prior"]:
# priors resolve by class through JSONPriorConfig, not
# RecursiveConfig: same relative path only
stack = [by_repo[lib][rel] for lib in src.get("stack") or ()
if rel in (by_repo.get(lib) or {})
and not by_repo[lib][rel]["error"]]
else:
stack = [r for r in (resolve_in_library(by_repo[lib], rel)
for lib in src.get("stack") or ()
if by_repo.get(lib)) if r is not None]
if not stack:
continue
overrides.append(diff_against_stack(rec, stack))
for src in sources:
recs = list((by_repo.get(src["repo"]) or {}).values())
src["files"] = len(recs)
src["lines"] = sum(r["lines"] for r in recs)
src["keys"] = sum(len(r["keys"]) for r in recs)
src["priors"] = sum(1 for r in recs if r["prior"])
src["errors"] = sum(1 for r in recs if r["error"])
if src["kind"] == "library" and lookups is not None:
src["unused"] = sum((r.get("use_counts") or {}).get("unused", 0)
for r in recs)
clean = [{k: v for k, v in f.items() if not k.startswith("_")}
for f in files]
return {"schema_version": SCHEMA_VERSION,
"generated": generated or _now_z(),
"owner": owner, "repo": repo or "PyAutoNerves",
"sources": sources, "files": clean, "overrides": overrides,
"env_vars": env, "errors": errors,
"lookups": lookups or {}}
def collect(root=None, brain=None, mind=None, sources_dir=None,
generated=None) -> dict:
"""Walk every ``SOURCES`` config dir and read the autonerves env vars."""
root = Path(root or os.environ.get("PYAUTO_ROOT") or Path.cwd())
sdir = Path(sources_dir) if sources_dir else None
body = load_body_map(root, mind)
owner, repo = _self_owner_repo()
errors: list[str] = []
if not body:
errors.append("PyAutoMind/repos.yaml not found — GitHub links use this "
"repo's owner and local paths fall back to <root>/<Repo>")
srcs, files = [], []
scans: dict | None = {}
for src in SOURCES:
meta = body.get(src["repo"]) or {}
github = meta.get("github") or (f"{owner}/{src['repo']}" if owner else "")
entry = {"repo": src["repo"], "github": github, "config": src["config"],
"kind": src["kind"], "stack": list(src["stack"]),
"found": False}
where = _source_dir(src, root, body, sdir)
if where is None:
errors.append(f"{src['repo']}: config dir {src['config']}/ not found")
srcs.append(entry)
if src["kind"] == "library":
scans = None # its code's reads are unknown: classify nothing
continue
entry["found"] = True
for path in sorted(where.rglob("*")):
if not path.is_file() or path.suffix not in YAML_SUFFIXES:
continue
rel = path.relative_to(where).as_posix()
try:
text = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as e:
errors.append(f"{src['repo']}/{rel}: unreadable ({e})")
continue
rec = read_file(src["repo"], rel, text, src["kind"])
rec["_library"] = src["kind"] == "library"
files.append(rec)
srcs.append(entry)
if src["kind"] == "library":
pkg = _source_dir(src, root, body, sdir, "package")
if pkg is None:
errors.append(f"{src['repo']}: package {src.get('package')}/ "
"not found — config keys are not classified")
scans = None
elif scans is not None:
scans[src["repo"]] = scan_lookups(pkg)
for rel in scans[src["repo"]]["unparsed"]:
errors.append(f"{src['repo']}: {src['package']}/{rel} did "
"not parse — its config reads are not counted")
texts = {}
for rel in ENV_MODULES:
try:
texts[rel] = (NERVES_HOME / rel).read_text(encoding="utf-8")
except OSError as e:
errors.append(f"env vars: {rel} unreadable ({e})")
if scans is not None:
# autonerves reads config itself (should_output, the backend) — its
# own package is a reader too, whatever repo it runs from.
scans["PyAutoNerves"] = scan_lookups(NERVES_PACKAGE)
return build_snapshot(files, srcs, env_vars_from(texts), errors, owner,
repo, generated, scans)
# --- derived views ----------------------------------------------------------------
def repo_url(snap: dict) -> str:
o, r = snap.get("owner"), snap.get("repo")
return f"https://github.com/{o}/{r}" if o and r else ""
def pages_url(snap: dict) -> str:
o, r = str(snap.get("owner") or "").lower(), snap.get("repo") or ""
return f"https://{o}.github.io/{r}/" if o and r else ""
def _source(snap: dict, repo: str) -> dict:
return next((s for s in snap.get("sources") or [] if s["repo"] == repo), {})
def file_url(snap: dict, repo: str, rel: str, line: int = 1) -> str:
s = _source(snap, repo)
if not s.get("github"):
return ""
return (f"https://github.com/{s['github']}/blob/main/{s['config']}/{rel}"
f"#L{max(int(line or 1), 1)}")
def slug(rel: str) -> str:
return "f-" + re.sub(r"[^A-Za-z0-9]+", "-", rel).strip("-").lower()
def unused_anchor(lib: str) -> str:
return "unused-" + re.sub(r"[^A-Za-z0-9]+", "-", lib).strip("-").lower()
def orphans(snap: dict) -> list[dict]:
return [o for o in snap.get("overrides") or [] if o["workspace_only"]]
def parse_errors(snap: dict) -> list[dict]:
return [f for f in snap.get("files") or [] if f.get("error")]
def unused_keys(snap: dict) -> dict:
"""``{library: [{path, k, line}]}`` — the library settings keys no code
in the stack reads, in source order (only classified snapshots)."""
out: dict = {}
for f in snap.get("files") or []:
for k in f.get("keys") or []:
if k.get("use") == "unused":
out.setdefault(f["repo"], []).append(
{"path": f["path"], "k": k["k"], "line": k["line"]})
return out
def status(snap: dict) -> str:
if not snap.get("files"):
return "grey"
if parse_errors(snap) or orphans(snap):
return "yellow"
return "green"
def _summary(snap: dict) -> str:
files = snap.get("files") or []
found = sum(1 for s in snap.get("sources") or [] if s.get("found"))
bits = [f"{len(files)} config files across {found} sources",
f"{len(snap.get('overrides') or [])} workspace overrides"]
if parse_errors(snap):