tools/m0_vocabulary_gap.py

1#!/usr/bin/env python3

2"""Enumerate the source fields that have NOWHERE TO GO in the frozen vocabulary.

4WHY THIS EXISTS

5---------------

6In M0 an unrepresentable field is a warning in an audit record. In M2/M3 it is

7PERMANENT DATA LOSS on the real corpus, discovered at cutover. The list this tool

8produces is the input to a vocabulary decision that has to happen before M1, not

9before M3.

11READ-ONLY. It opens no file for writing and never touches the estate it scans. Run

12it against the canonical estate to size the problem, or against any corpus copy.

14 python3 tools/m0_vocabulary_gap.py /home/daviwil/Ops/folio

15 python3 tools/m0_vocabulary_gap.py docs/f0/pressure-corpus --json out.json

17The FIELD SETS ARE IMPORTED FROM THE CONVERTER, which transcribes them from

18`src/folio/doc/model.sgl`'s define-struct forms. Restating them here would be a

19second list that goes stale the moment the model moves.

20"""

22from __future__ import annotations

24import argparse

25import json

26import sys

27from collections import Counter, defaultdict

28from pathlib import Path

30sys.path.insert(0, str(Path(__file__).resolve().parent))

32from f0_inventory import META_RE, TASK_RE, classify, comment_mask # noqa: E402

33from m0_convert import ( # noqa: E402

34 NOTE_FIELDS,

35 PROJECT_FIELDS,

36 TASK_FIELDS,

37 Source,

38 parse_metadata,

39)

42def scan(root: Path) -> dict:

43 task_keys: Counter[str] = Counter()

44 fm_note_keys: Counter[str] = Counter()

45 fm_project_keys: Counter[str] = Counter()

46 examples: dict[str, list[str]] = defaultdict(list)

47 files = 0

48 skipped: Counter[str] = Counter()

50 for path in sorted(root.rglob("*.md")):

51 rel = str(path.relative_to(root)).replace("\\", "/")

52 kind, _ = classify(rel)

53 if kind not in {"inbox", "note", "project", "archived-note", "archived-project"}:

54 skipped[kind] += 1

55 continue

56 try:

57 data = path.read_bytes()

58 src = Source(rel, data)

59 except (UnicodeDecodeError, ValueError):

60 skipped["unreadable"] += 1

61 continue

62 files += 1

64 masked = comment_mask(src.lines)

65 for i, line in enumerate(masked):

66 if src.in_fence(i) or i in src.frontmatter_lines:

67 continue

68 if not TASK_RE.match(line):

69 continue

70 body = TASK_RE.match(src.lines[i]).group("body")

71 m = META_RE.search(body)

72 if not m:

73 continue

74 for key, value in parse_metadata(m.group(1)):

75 task_keys[key] += 1

76 if key not in TASK_FIELDS and len(examples[f"task:{key}"]) < 3:

77 examples[f"task:{key}"].append(f"{rel}:{i + 1}: {key}: {value}")

79 target = fm_project_keys if kind in ("project", "archived-project") else fm_note_keys

80 allowed = PROJECT_FIELDS if kind in ("project", "archived-project") else NOTE_FIELDS

81 for key, value in src.frontmatter.items():

82 target[key] += 1

83 if key not in allowed and len(examples[f"frontmatter:{key}"]) < 3:

84 examples[f"frontmatter:{key}"].append(f"{rel}: {key}: {value}")

86 def split(counter: Counter[str], allowed: set[str]) -> dict:

87 has_home = {k: v for k, v in sorted(counter.items()) if k in allowed}

88 no_home = {k: v for k, v in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))

89 if k not in allowed}

90 return {

91 "population": sum(counter.values()),

92 "distinct_keys": len(counter),

93 "has_a_frozen_field": has_home,

94 "NO_HOME": no_home,

95 "no_home_occurrences": sum(no_home.values()),

96 }

98 return {

99 "files_scanned": files,

100 "files_skipped_by_class": dict(sorted(skipped.items())),

101 "task_metadata": split(task_keys, TASK_FIELDS),

102 "note_frontmatter": split(fm_note_keys, NOTE_FIELDS),

103 "project_frontmatter": split(fm_project_keys, PROJECT_FIELDS),

104 "frozen_fields": {

105 "task-record": sorted(TASK_FIELDS),

106 "note-record": sorted(NOTE_FIELDS),

107 "project-record": sorted(PROJECT_FIELDS),

108 },

109 "examples": {k: v for k, v in sorted(examples.items())},

110 }

111

112

113def main() -> int:

114 ap = argparse.ArgumentParser(description=__doc__)

115 ap.add_argument("root", type=Path)

116 ap.add_argument("--json", type=Path)

117 args = ap.parse_args()

118

119 report = scan(args.root)

120 if args.json:

121 args.json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")

122

123 print(f"files scanned: {report['files_scanned']}")

124 for section in ("task_metadata", "note_frontmatter", "project_frontmatter"):

125 d = report[section]

126 print(f"\n== {section} == population {d['population']} across "

127 f"{d['distinct_keys']} distinct keys")

128 print(f" representable: {d['has_a_frozen_field']}")

129 print(f" NO HOME ({d['no_home_occurrences']} occurrences): {d['NO_HOME']}")

130 total = sum(report[s]["no_home_occurrences"]

131 for s in ("task_metadata", "note_frontmatter", "project_frontmatter"))

132 print(f"\nM0_VOCAB_GAP unrepresentable_occurrences={total} files={report['files_scanned']}")

133 return 0

134

135

136if __name__ == "__main__":

137 raise SystemExit(main())