From 710d4ab65f7c08fdd933283ebddf25324914e946 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Sun, 19 Jul 2026 22:46:04 +0800 Subject: [PATCH] refactor: sync shadbala same unit normalizer --- scripts/shadbala_same_unit_normalizer.py | 132 +++++++++++++---------- 1 file changed, 74 insertions(+), 58 deletions(-) diff --git a/scripts/shadbala_same_unit_normalizer.py b/scripts/shadbala_same_unit_normalizer.py index e8f5a138..2085100b 100644 --- a/scripts/shadbala_same_unit_normalizer.py +++ b/scripts/shadbala_same_unit_normalizer.py @@ -1,10 +1,11 @@ #!/usr/bin/env python3 -"""Normalize local/jyotishganit/Xalen/VP Jain Shadbala component rows. +"""Normalize Shadbala component rows to the same Virupa/Rupa unit. -All component values are represented as Virupa plus derived Rupa=Virupa/60 -when numeric. The output classifies row-level deltas without declaring -absolute truth. +This is an arbitration aid only. It aligns local-observation, jyotishganit, +Xalen, and VP Jain numeric fields into one 42-row matrix; it does not select +an absolute formula truth. """ + from __future__ import annotations import hashlib @@ -12,10 +13,12 @@ import json from pathlib import Path from typing import Any + ROOT = Path(__file__).resolve().parents[1] -JYO = ROOT / "references/oracle/jyotishganit_shadbala_surface_probe_steve_jobs_2026_07_19.json" +JYOTISHGANIT = ROOT / "references/oracle/jyotishganit_shadbala_surface_probe_steve_jobs_2026_07_19.json" XALEN = ROOT / "references/oracle/xalen_shadbala_av_component_delta_report_2026_07_19.json" -VPJ = ROOT / "references/oracle/vp_jain_shadbala_component_benchmark_2026_07_17.json" +VP_JAIN = ROOT / "references/oracle/vp_jain_shadbala_component_benchmark_2026_07_17.json" + PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"] COMPONENTS = { "sthana": "Sthanabala", @@ -27,46 +30,47 @@ COMPONENTS = { } -def stable(data: Any) -> str: +def stable_json(data: Any) -> str: return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) -def rupa(v: float | None) -> float | None: - return None if v is None else round(v / 60.0, 6) +def as_float(value: Any) -> float | None: + return float(value) if isinstance(value, (int, float)) else None -def get_jyo_value(jyo: dict[str, Any], planet: str, canonical: str) -> float | None: - val = jyo["raw"]["shadbala"].get(planet, {}).get(canonical) - if isinstance(val, dict): - val = val.get("Total") - return val if isinstance(val, (int, float)) else None +def to_rupa(virupa: float | None) -> float | None: + return None if virupa is None else round(virupa / 60.0, 6) -def xalen_index(xalen: dict[str, Any]) -> dict[tuple[str, str], dict[str, Any]]: - out: dict[tuple[str, str], dict[str, Any]] = {} - for group in xalen.get("component_groups", []): +def jyotishganit_value(raw: dict[str, Any], planet: str, component: str) -> float | None: + value = raw.get("raw", {}).get("shadbala", {}).get(planet, {}).get(component) + if isinstance(value, dict): + value = value.get("Total") + return as_float(value) + + +def xalen_rows(raw: dict[str, Any]) -> dict[tuple[str, str], dict[str, Any]]: + indexed: dict[tuple[str, str], dict[str, Any]] = {} + for group in raw.get("component_groups", []): component = group.get("component") if component not in COMPONENTS: continue for row in group.get("rows", []): - planet = row.get("field") or row.get("planet") - out[(planet, component)] = row - return out + field = str(row.get("field") or row.get("planet") or "") + planet = field.split(".", 1)[0] + indexed[(planet, component)] = row + return indexed -def vpj_index(vpj: dict[str, Any]) -> dict[tuple[str, str], dict[str, Any]]: - return { - (row.get("planet"), row.get("component")): row - for row in vpj.get("rows", []) - } +def vp_jain_rows(raw: dict[str, Any]) -> dict[tuple[str, str], dict[str, Any]]: + return {(row.get("planet"), row.get("component")): row for row in raw.get("rows", [])} def classify(values: dict[str, float | None], statuses: dict[str, str | None]) -> str: - nums = {k: v for k, v in values.items() if isinstance(v, (int, float))} - if len(nums) < 2: + numeric_values = [v for v in values.values() if isinstance(v, (int, float))] + if len(numeric_values) < 2: return "insufficient_numeric_sources" - span = max(nums.values()) - min(nums.values()) - if span <= 1.0: + if max(numeric_values) - min(numeric_values) <= 1.0: return "within_1_virupa_observation" if any(status == "method_variant" for status in statuses.values() if status): return "method_variant" @@ -74,47 +78,56 @@ def classify(values: dict[str, float | None], statuses: dict[str, str | None]) - def build() -> dict[str, Any]: - jyo = json.loads(JYO.read_text(encoding="utf-8")) - xalen = xalen_index(json.loads(XALEN.read_text(encoding="utf-8"))) - vpj = vpj_index(json.loads(VPJ.read_text(encoding="utf-8"))) - rows = [] + jyotishganit = json.loads(JYOTISHGANIT.read_text(encoding="utf-8")) + xalen = xalen_rows(json.loads(XALEN.read_text(encoding="utf-8"))) + vp_jain = vp_jain_rows(json.loads(VP_JAIN.read_text(encoding="utf-8"))) + + rows: list[dict[str, Any]] = [] for planet in PLANETS: - for short, canonical in COMPONENTS.items(): - jyo_v = get_jyo_value(jyo, planet, canonical) - xrow = xalen.get((planet, short), {}) - vrow = vpj.get((planet, short), {}) + for short_name, canonical_name in COMPONENTS.items(): + jyo_virupa = jyotishganit_value(jyotishganit, planet, canonical_name) + xalen_row = xalen.get((planet, short_name), {}) + vp_jain_row = vp_jain.get((planet, short_name), {}) + values = { - "jyotishganit_virupa": jyo_v, - "xalen_virupa": xrow.get("xalen_value"), - "local_from_xalen_report_virupa": xrow.get("local_value"), - "vp_jain_published_virupa": vrow.get("published_value"), - "vp_jain_local_virupa": vrow.get("local_value"), + "jyotishganit_virupa": jyo_virupa, + "xalen_virupa": as_float(xalen_row.get("xalen_value")), + "local_from_xalen_report_virupa": as_float(xalen_row.get("local_value")), + "vp_jain_published_virupa": as_float(vp_jain_row.get("published_value")), + "vp_jain_local_virupa": as_float(vp_jain_row.get("local_value")), } statuses = { - "xalen_status": xrow.get("status"), - "vp_jain_status": vrow.get("status"), + "xalen_status": xalen_row.get("status"), + "vp_jain_status": vp_jain_row.get("status"), } + rows.append( { "planet": planet, - "component": short, - "canonical_component": canonical, + "component": short_name, + "canonical_component": canonical_name, **values, - "jyotishganit_rupa": rupa(jyo_v), - "xalen_rupa": rupa(xrow.get("xalen_value")), - "vp_jain_published_rupa": rupa(vrow.get("published_value")), + "jyotishganit_rupa": to_rupa(values["jyotishganit_virupa"]), + "xalen_rupa": to_rupa(values["xalen_virupa"]), + "local_from_xalen_report_rupa": to_rupa(values["local_from_xalen_report_virupa"]), + "vp_jain_published_rupa": to_rupa(values["vp_jain_published_virupa"]), + "vp_jain_local_rupa": to_rupa(values["vp_jain_local_virupa"]), **statuses, "normalization_unit": "Virupa", "classification": classify(values, statuses), - "claim_boundary": "Same-unit row only; formula truth still requires source variant selection and public worked example.", + "claim_boundary": ( + "Same-unit observation row only; formula truth still requires " + "component source variant selection and public numeric worked examples." + ), } ) + summary = { "row_count": len(rows), - "within_1_virupa_observation_count": sum(1 for r in rows if r["classification"] == "within_1_virupa_observation"), - "method_variant_count": sum(1 for r in rows if r["classification"] == "method_variant"), - "formula_or_unit_mismatch_count": sum(1 for r in rows if r["classification"] == "formula_or_unit_mismatch"), - "insufficient_numeric_sources_count": sum(1 for r in rows if r["classification"] == "insufficient_numeric_sources"), + "within_1_virupa_observation_count": sum(1 for row in rows if row["classification"] == "within_1_virupa_observation"), + "method_variant_count": sum(1 for row in rows if row["classification"] == "method_variant"), + "formula_or_unit_mismatch_count": sum(1 for row in rows if row["classification"] == "formula_or_unit_mismatch"), + "insufficient_numeric_sources_count": sum(1 for row in rows if row["classification"] == "insufficient_numeric_sources"), } return { "scope": "shadbala_same_unit_normalizer", @@ -124,14 +137,17 @@ def build() -> dict[str, Any]: "production_tuning_allowed": False, "truth_matrix_allowed": False, "sources": { - "jyotishganit": str(JYO.relative_to(ROOT)), + "jyotishganit": str(JYOTISHGANIT.relative_to(ROOT)), "xalen": str(XALEN.relative_to(ROOT)), - "vp_jain": str(VPJ.relative_to(ROOT)), + "vp_jain": str(VP_JAIN.relative_to(ROOT)), }, "summary": summary, - "matrix_hash": hashlib.sha256(stable(rows).encode("utf-8")).hexdigest(), + "matrix_hash": hashlib.sha256(stable_json(rows).encode("utf-8")).hexdigest(), "rows": rows, - "boundary": "All 42 rows are normalized to Virupa/Rupa fields where available. Classifications are arbitration queues, not absolute parity closure.", + "boundary": ( + "All 42 rows are normalized to Virupa/Rupa fields where available. " + "Classifications are arbitration queues, not absolute parity closure." + ), }