File size: 8,949 Bytes
aa1d0aa | 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 | #!/usr/bin/env python3
"""
render_spec.py β deterministic YAML -> Markdown renderer for the Step 3 binding analysis spec.
The YAML spec is the source of truth. This script ONLY formats what the spec already
contains: no LLM, no catalog lookups, no network. Catalog/instrument certification is
verify_spec.py's job and runs BEFORE this (journal Step 13). render_spec.py assumes a
clean spec and fails loudly on a malformed one.
Usage:
render_spec.py <spec>.yaml [-o <out>.md]
Reference output: 12_layout_exploration/output/format_examples/format_1_REVISED_analysis_spec.md
"""
import argparse
import sys
import yaml
def cell(text):
"""Make a value safe for a single Markdown table cell.
Collapses all whitespace (including newlines from YAML block scalars) to single spaces
and escapes pipes so they don't split the cell.
"""
if text is None:
return "β"
s = str(text)
s = s.replace("\r\n", "\n").replace("\r", "\n")
s = " ".join(s.split()) # collapse all whitespace runs (incl. newlines) to single spaces
s = s.replace("|", "\\|") # escape pipes so they don't split the cell
return s or "β"
DIM_LABEL = {
"sector": "Sector",
"pillar": "Pillar (2D)",
"subpillar": "Subpillar (2D)",
"cross_cutting": "Cross-cutting lens",
}
COLS = (
"| Indicator | Definition (catalog) | Measurable | Reasons (proves / blocks) "
"| Variables in the dataset | Indicator name in the analysis |"
)
SEP = "|---|---|---|---|---|---|"
def die(msg):
sys.stderr.write(f"render_spec.py: {msg}\n")
sys.exit(1)
def require(spec, key):
if key not in spec:
die(f"required key '{key}' missing from spec")
return spec[key]
def fmt_vars(v):
return ", ".join(f"`{x}`" for x in v.get("variables", [])) or "β"
def fmt_result_id(result_id, v):
ids = v.get("result_ids")
if ids:
return ", ".join(f"`{i}`" for i in ids)
return f"`{result_id}`"
def dim_key(dim):
# "sector::WASH" -> ("sector", "WASH")
base = dim.split("::", 1)[0]
return base, dim
def render(spec):
out = []
src = spec.get("_source_file", "<spec>.yaml")
out.append(f"<!-- generated from {src} β do not edit by hand -->")
out.append("")
out.append(f"# Analysis Spec β {require(spec, 'dataset')}")
out.append("")
out.append(f"**Unit of analysis:** {spec.get('unit_of_analysis', 'β')} Β· "
f"**Catalog version (Step 2):** {spec.get('step2_catalog_version', 'β')} Β· "
f"**n:** {spec.get('n_total', 'β')}")
route = spec.get("step1_framework_route", {})
if route:
out.append(f"**Route:** Sector `{route.get('sector','β')}` Β· "
f"Pillar `{route.get('pillar_2d','β')}` Β· "
f"Subpillar `{route.get('subpillar_2d','β')}`"
+ (f" Β· Cross-cutting `{route['cross_cutting']}`"
if route.get("cross_cutting") else ""))
out.append("**Status:** β DRAFT β β REVIEWED β β APPROVED β β ANALYSIS RUN")
out.append("")
out.append("---")
out.append("")
# ---- Β§1 coverage map, one table per dimension present on the indicators ----
# Indicators are grouped by their `dimension` (sector::X or cross_cutting::X).
# The pillar/subpillar route is recorded in the header, NOT as a table: in this
# catalog every indicator shares one pillar/subpillar anchor, so a pillar table
# would only duplicate the sector table. There is therefore no inheritance to
# cross-reference β each indicator simply appears once under its own dimension.
out.append("## 1. Coverage map β by framework dimension (Step 1)")
out.append("")
out.append("*One table per dimension that carries indicators (sector, plus any "
"cross-cutting lens). The pillar/subpillar route is in the header above. "
"Each indicator appears once.*")
out.append("")
indicators = require(spec, "indicators")
# Group by full dimension string; sort within each group MEASURABLEβPROXYβNOT_MEASURABLE then id asc.
# This makes the rendered MD canonical regardless of YAML insertion order (D13).
MEASURABLE_RANK = {"MEASURABLE": 0, "PROXY": 1, "NOT_MEASURABLE": 2}
# Human-readable verdict labels (internal YAML values β display)
VERDICT_LABEL = {"MEASURABLE": "Measurable", "PROXY": "Proxy", "NOT_MEASURABLE": "Not measurable"}
order = []
by_dim = {}
for rid, v in indicators.items():
dim = v.get("dimension", "sector::?")
_, full = dim_key(dim)
by_dim.setdefault(full, []).append((rid, v))
if full not in order:
order.append(full)
# Sort each dimension's indicators: MEASURABLE first, then PROXY, then NOT_MEASURABLE, then id asc
for full in order:
by_dim[full].sort(
key=lambda rv: (MEASURABLE_RANK.get(rv[1].get("measurable", ""), 9), rv[0])
)
for section, full in enumerate(order, start=1):
base, _ = dim_key(full)
label = DIM_LABEL.get(base, base)
out.append(f"### 1.{section} {label} β `{full.split('::', 1)[1]}`")
out.append("")
out.append(COLS)
out.append(SEP)
for rid, v in by_dim[full]:
verdict_raw = v.get('measurable', '')
verdict_display = VERDICT_LABEL.get(verdict_raw, cell(verdict_raw))
# "Indicator name in the analysis" is empty for Not measurable rows
result_cell = "" if verdict_raw == "NOT_MEASURABLE" else fmt_result_id(rid, v)
out.append(
"| " + cell(v.get('label', rid))
+ " | " + cell(v.get('definition', 'β'))
+ " | " + verdict_display
+ " | " + cell(v.get('reasons', 'β'))
+ " | " + fmt_vars(v)
+ " | " + result_cell + " |"
)
out.append("")
# ---- Β§1c uncovered modules ----
uncovered = spec.get("uncovered_modules")
if uncovered:
section_count = len(order) + 1
out.append(f"### 1.{section_count} Survey modules with no catalog indicator")
out.append("")
out.append("| Module | Variables | Note |")
out.append("|---|---|---|")
for entry in uncovered:
mod = cell(entry.get("module", "β"))
vars_str = ", ".join(f"`{v}`" for v in entry.get("variables", [])) or "β"
note = cell(entry.get("note", "no catalog indicator"))
out.append(f"| {mod} | {vars_str} | {note} |")
out.append("")
# ---- Β§1b disaggregation ----
dis = spec.get("disaggregation")
if dis:
out.append("---")
out.append("")
out.append("## 1b. Disaggregation required")
out.append("")
out.append("Report **every indicator row above separately** for:")
out.append("")
groups = dis.get("groups", {})
srcs = ", ".join(f"`{s}`" for s in dis.get("source_variables", []))
# Only show n column if at least one group has a real count
has_counts = any(v is not None for v in groups.values())
if has_counts:
out.append("| Group | n | Source variable |")
out.append("|---|---|---|")
for g, n in groups.items():
n_val = cell(str(n)) if n is not None else "β"
out.append(f"| {cell(g)} | {n_val} | {srcs} |")
else:
out.append("| Group | Source variable |")
out.append("|---|---|")
for g in groups:
out.append(f"| {cell(g)} | {srcs} |")
out.append("")
trig = dis.get("trigger")
if trig:
out.append(f"**Trigger:** {cell(trig)}. A single pooled figure per indicator is "
f"**not** acceptable β each table must carry every group.")
out.append("")
# ---- Β§2 gates ----
gates = spec.get("gates", [])
if gates:
out.append("---")
out.append("")
out.append("## 2. Pass criteria β gates *(all must hold before the report ships)*")
out.append("")
out.append("| Gate | Pass condition | β |")
out.append("|---|---|---|")
for g in gates:
out.append(f"| {cell(g.get('id','β'))} | {cell(g.get('assert','β'))} | β |")
out.append("")
return "\n".join(out) + "\n"
def main():
ap = argparse.ArgumentParser()
ap.add_argument("spec")
ap.add_argument("-o", "--out")
a = ap.parse_args()
with open(a.spec) as f:
spec = yaml.safe_load(f)
if not isinstance(spec, dict):
die("spec did not parse to a mapping")
spec["_source_file"] = a.spec.split("/")[-1]
md = render(spec)
if a.out:
with open(a.out, "w") as f:
f.write(md)
print(f"wrote {a.out}")
else:
sys.stdout.write(md)
if __name__ == "__main__":
main()
|