"""vals.py: the single values file for the project (numerical integrity rules). Every number that appears in any report is a named entry here, written by the script that computed it. One canonical name per quantity; rounded or converted presentations are produced by the formatters below from that one entry at render time, never stored twice. Every interval records its estimand, and it must equal the estimand of its point estimate. """ import json, os, re, math, datetime VALUES_PATH = os.environ.get("MILL_VALUES", "/home/claude/mill/outputs/values.json") REQUIRED = ("value", "unit", "estimand", "source") class Values: def __init__(self, path=VALUES_PATH, script=None): self.path = path self.script = script self.v = {} if os.path.exists(path): with open(path) as fh: self.v = json.load(fh) self._touched = set() # --- writers ------------------------------------------------------------------------- def set(self, name, value, unit, estimand, note=None, kind="value"): if not re.fullmatch(r"[a-z0-9_]+", name): raise ValueError(f"bad value name {name!r}") if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): raise ValueError(f"{name}: non-finite value") if name in self._touched: raise ValueError(f"{name}: set twice in one script run (one canonical name per quantity)") prev = self.v.get(name) if prev is not None and prev.get("source") not in (None, self.script): raise ValueError(f"{name}: already owned by {prev.get('source')}, not {self.script}") rec = {"value": _plain(value), "unit": unit, "estimand": estimand, "source": self.script, "kind": kind} if note: rec["note"] = note self.v[name] = rec self._touched.add(name) return value def interval(self, name, point, lo, hi, level, method, estimand, note=None): """An interval for the point estimate `point`; estimand must match the point's.""" if point not in self.v: raise KeyError(f"{name}: point estimate {point} not in values file") if self.v[point]["estimand"] != estimand: raise ValueError(f"{name}: estimand differs from that of {point}") if name in self._touched: raise ValueError(f"{name}: set twice in one script run") rec = {"value": None, "lo": _plain(lo), "hi": _plain(hi), "level": level, "method": method, "point": point, "unit": self.v[point]["unit"], "estimand": estimand, "source": self.script, "kind": "interval"} if note: rec["note"] = note self.v[name] = rec self._touched.add(name) def get(self, name): return self.v[name]["value"] def drop_owned(self): """Remove entries previously written by this script (so a re-run cannot leave stale names).""" for k in [k for k, r in self.v.items() if r.get("source") == self.script and k not in self._touched]: del self.v[k] def save(self): self.drop_owned() os.makedirs(os.path.dirname(self.path), exist_ok=True) tmp = self.path + ".tmp" with open(tmp, "w") as fh: json.dump(dict(sorted(self.v.items())), fh, indent=1) os.replace(tmp, self.path) def _plain(x): """Convert numpy scalars to plain Python numbers.""" try: import numpy as np if isinstance(x, np.generic): return x.item() except ImportError: pass return x # --- formatters (presentation only; each is a pure function of one entry) ----------------- def _num(x, nd): return f"{x:,.{nd}f}" def fmt_value(rec, spec): v = rec.get("value") if rec.get("kind") == "interval": lo, hi = rec["lo"], rec["hi"] if spec.startswith("ci_ppt"): # difference of two shares, with the unit nd = int(spec[6:] or 0) return f"{_num(100 * lo, nd)} to {_num(100 * hi, nd)} percentage points" if spec.startswith("ci_pp"): # difference of two shares, number only (unit in a table header) nd = int(spec[5:] or 0) return f"{_num(100 * lo, nd)} to {_num(100 * hi, nd)}" if spec.startswith("ci_pct"): nd = int(spec[6:] or 0) return f"{_num(100 * lo, nd)} to {_num(100 * hi, nd)} percent" if spec.startswith("ci_f"): nd = int(spec[4:]) return f"{_num(lo, nd)} to {_num(hi, nd)}" if spec.startswith("ci_g"): nd = int(spec[4:]) return f"{_sig(lo, nd)} to {_sig(hi, nd)}" raise ValueError(f"interval needs a ci_ format, got {spec!r}") if spec in ("", "raw"): return str(v) if spec == "int": return f"{int(round(v)):,}" if spec.startswith("f"): return _num(v, int(spec[1:])) if spec.startswith("pct"): return _num(100 * v, int(spec[3:] or 0)) + " percent" if spec.startswith("ppt"): # difference of two shares, with the unit return _num(100 * v, int(spec[3:] or 0)) + " percentage points" if spec.startswith("pp"): # percent number without the word (unit in a table header) return _num(100 * v, int(spec[2:] or 0)) if spec.startswith("g"): return _sig(v, int(spec[1:])) if spec == "hms": s = int(round(v)); h, r = divmod(s, 3600); m, s2 = divmod(r, 60) return f"{h} h {m} min {s2} s" if spec == "gb": return _num(v / 1e9, 1) + " GB" if spec == "mb": return _num(v / 1e6, 1) + " MB" if spec == "khz": return _num(v / 1000, 1) + " kHz" if spec in ("word", "wordcap"): words = ["zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve", "thirteen", "fourteen", "fifteen", "sixteen", "seventeen", "eighteen", "nineteen", "twenty"] iv = int(round(v)) if abs(v - iv) > 1e-9 or not 0 <= iv <= 20: raise ValueError(f"word format needs an integer 0 to 20, got {v}") w = words[iv] return w[0].upper() + w[1:] if spec == "wordcap" else w if spec == "date": return str(v) if spec == "text": return str(v) raise ValueError(f"unknown format {spec!r}") def _sig(x, n): if x == 0: return "0" d = n - int(math.floor(math.log10(abs(x)))) - 1 d = max(d, 0) return f"{round(x, d):,.{d}f}" TOKEN = re.compile(r"\{\{\s*([a-z0-9_]+)\s*(?:\|\s*([a-z0-9_]+)\s*)?\}\}") def render(text, values): missing = [] def sub(m): name, spec = m.group(1), (m.group(2) or "") if name not in values: missing.append(name) return m.group(0) return fmt_value(values[name], spec) out = TOKEN.sub(sub, text) return out, missing def today(): return datetime.date.today().isoformat()