test: import research oracle probes
This commit is contained in:
732642856
@@ -0,0 +1,308 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Probe jyotishyamitra as an independent oracle without copying implementation."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
|
||||
VERSION = "1.4.0"
|
||||
COMMIT = "86f7eb610a66b06b3f0817d2c53355bec8b3bf8d"
|
||||
LICENSE = "MIT"
|
||||
RETURNVAL = "ASTRODATA_DICTIONARY"
|
||||
|
||||
|
||||
def _sha256_bytes(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _stable_json(data: object) -> str:
|
||||
return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def wheel_metadata(wheel_path: Path) -> dict:
|
||||
with zipfile.ZipFile(wheel_path) as zf:
|
||||
metadata_name = next(name for name in zf.namelist() if name.endswith(".dist-info/METADATA"))
|
||||
wheel_name = next(name for name in zf.namelist() if name.endswith(".dist-info/WHEEL"))
|
||||
license_name = next((name for name in zf.namelist() if name.endswith(".dist-info/licenses/LICENSE")), None)
|
||||
metadata_text = zf.read(metadata_name).decode("utf-8", "replace")
|
||||
wheel_text = zf.read(wheel_name).decode("utf-8", "replace")
|
||||
license_text = zf.read(license_name).decode("utf-8", "replace") if license_name else ""
|
||||
fields = {}
|
||||
for line in metadata_text.splitlines():
|
||||
if ": " in line:
|
||||
key, value = line.split(": ", 1)
|
||||
fields.setdefault(key, value)
|
||||
tags = [line.split(": ", 1)[1] for line in wheel_text.splitlines() if line.startswith("Tag: ")]
|
||||
return {
|
||||
"metadata_name": metadata_name,
|
||||
"wheel_name": wheel_name,
|
||||
"name": fields.get("Name"),
|
||||
"version": fields.get("Version"),
|
||||
"license": fields.get("License"),
|
||||
"license_file_name": license_name,
|
||||
"license_file_sha256": _sha256_bytes(license_text.encode("utf-8")) if license_text else None,
|
||||
"license_file_spdx_inferred": "MIT" if "MIT License" in license_text else None,
|
||||
"requires_python": fields.get("Requires-Python"),
|
||||
"summary": fields.get("Summary"),
|
||||
"wheel_tags": tags,
|
||||
"metadata_sha256": _sha256_bytes(metadata_text.encode("utf-8")),
|
||||
"wheel_record_sha256": _sha256_bytes(wheel_text.encode("utf-8")),
|
||||
}
|
||||
|
||||
|
||||
def canonical_request(case: dict) -> dict:
|
||||
birth = case["birth"]
|
||||
return {
|
||||
"name": case.get("name", "public_case"),
|
||||
"gender": case.get("gender", "male"),
|
||||
"place": case.get("place", "unknown"),
|
||||
"longitude": case["longitude"],
|
||||
"latitude": case["latitude"],
|
||||
"timezone": case["timezone"],
|
||||
"birth": {
|
||||
"year": birth["year"],
|
||||
"month": birth["month"],
|
||||
"day": birth["day"],
|
||||
"hour": birth["hour"],
|
||||
"minute": birth["minute"],
|
||||
"second": birth.get("second", 0),
|
||||
},
|
||||
"ayanamsa": case.get("ayanamsa", "package_default"),
|
||||
"node_mode": case.get("node_mode", "package_default"),
|
||||
"returnval": RETURNVAL,
|
||||
}
|
||||
|
||||
|
||||
def schema_fingerprint(raw: object) -> dict:
|
||||
paths = []
|
||||
|
||||
def walk(value: object, prefix: str = "$") -> None:
|
||||
if isinstance(value, dict):
|
||||
for key in sorted(value):
|
||||
walk(value[key], f"{prefix}.{key}")
|
||||
elif isinstance(value, list):
|
||||
paths.append(f"{prefix}[]")
|
||||
if value:
|
||||
walk(value[0], f"{prefix}[]")
|
||||
else:
|
||||
paths.append(f"{prefix}:{type(value).__name__}")
|
||||
|
||||
walk(raw)
|
||||
joined = "\n".join(paths)
|
||||
return {"path_count": len(paths), "sha256": _sha256_bytes(joined.encode("utf-8")), "sample_paths": paths[:80]}
|
||||
|
||||
|
||||
def normalize_raw(raw: object) -> object:
|
||||
data = copy.deepcopy(raw)
|
||||
try:
|
||||
data["Dashas"]["Vimshottari"]["current"]["date"] = "<volatile_run_time>"
|
||||
except (TypeError, KeyError):
|
||||
pass
|
||||
return data
|
||||
|
||||
|
||||
def run_installed_jyotishyamitra(case: dict) -> object:
|
||||
import importlib
|
||||
|
||||
jm = importlib.import_module("jyotishyamitra")
|
||||
request = canonical_request(case)
|
||||
birth = request["birth"]
|
||||
data = jm.input_birthdata(
|
||||
name=request["name"],
|
||||
gender=request["gender"],
|
||||
place=request["place"],
|
||||
longitude=str(request["longitude"]),
|
||||
lattitude=str(request["latitude"]),
|
||||
timezone=str(request["timezone"]),
|
||||
year=str(birth["year"]),
|
||||
month=str(birth["month"]),
|
||||
day=str(birth["day"]),
|
||||
hour=str(birth["hour"]),
|
||||
min=str(birth["minute"]),
|
||||
sec=str(birth.get("second", 0)),
|
||||
)
|
||||
validation = jm.validate_birthdata()
|
||||
if validation != "SUCCESS":
|
||||
return {"status": "INPUT_ERROR", "validation": validation, "input": data}
|
||||
return jm.generate_astrologicalData(jm.get_birthdata(), returnval=request["returnval"])
|
||||
|
||||
|
||||
def run_isolated(wheel_path: Path, case: dict) -> dict:
|
||||
with TemporaryDirectory() as tmp:
|
||||
subprocess.run(
|
||||
[sys.executable, "-m", "pip", "install", "--no-deps", "--target", tmp, str(wheel_path)],
|
||||
check=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
code = (
|
||||
"import json,sys;"
|
||||
"sys.path.insert(0, sys.argv[1]);"
|
||||
"from scripts.jyotishyamitra_adapter_probe import run_installed_jyotishyamitra;"
|
||||
"case=json.loads(sys.stdin.read());"
|
||||
"print(json.dumps(run_installed_jyotishyamitra(case), ensure_ascii=False, sort_keys=True))"
|
||||
)
|
||||
env = {**os.environ, "PYTHONPATH": str(Path.cwd())}
|
||||
done = subprocess.run(
|
||||
[sys.executable, "-c", code, tmp],
|
||||
input=_stable_json(case),
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
env=env,
|
||||
check=False,
|
||||
)
|
||||
raw = json.loads(done.stdout) if done.returncode == 0 and done.stdout.strip().startswith(("{", "[", '"')) else done.stdout.strip()
|
||||
return {
|
||||
"install_path": tmp,
|
||||
"returncode": done.returncode,
|
||||
"stderr": done.stderr.strip(),
|
||||
"raw": raw,
|
||||
}
|
||||
|
||||
|
||||
def extract_fields(raw: dict) -> dict:
|
||||
keys = ("D1", "D2", "D4", "D9", "D10", "ashtakavarga", "shadbala", "vimshottari")
|
||||
lowered = {str(k).lower(): k for k in raw}
|
||||
out = {}
|
||||
for key in keys:
|
||||
source = lowered.get(key.lower())
|
||||
if source is not None:
|
||||
out[key] = raw[source]
|
||||
return out
|
||||
|
||||
|
||||
def compare_with_existing_oracles(jyotishyamitra: dict, local: dict, xalen: dict) -> dict:
|
||||
rows = []
|
||||
counts = {"local": 0, "xalen": 0}
|
||||
for section, fields in jyotishyamitra.items():
|
||||
if not isinstance(fields, dict):
|
||||
continue
|
||||
for field, value in fields.items():
|
||||
local_value = (local.get(section) or {}).get(field)
|
||||
xalen_value = (xalen.get(section) or {}).get(field)
|
||||
local_match = value == local_value
|
||||
xalen_match = value == xalen_value
|
||||
counts["local"] += int(local_match)
|
||||
counts["xalen"] += int(xalen_match)
|
||||
rows.append(
|
||||
{
|
||||
"section": section,
|
||||
"field": field,
|
||||
"jyotishyamitra_value": value,
|
||||
"local_value": local_value,
|
||||
"xalen_value": xalen_value,
|
||||
"local_status": "match" if local_match else "mismatch",
|
||||
"xalen_status": "match" if xalen_match else "mismatch",
|
||||
}
|
||||
)
|
||||
return {
|
||||
"scope": "jyotishyamitra_field_comparison",
|
||||
"row_count": len(rows),
|
||||
"match_counts": counts,
|
||||
"promotion_allowed": False,
|
||||
"truth_policy": "independent_observation_not_truth",
|
||||
"rows": rows,
|
||||
}
|
||||
|
||||
|
||||
def build_report(
|
||||
wheel_path: Path,
|
||||
commit: str = COMMIT,
|
||||
raw: dict | None = None,
|
||||
comparison: dict | None = None,
|
||||
request: dict | None = None,
|
||||
isolated_run: dict | None = None,
|
||||
) -> dict:
|
||||
if not wheel_path.exists():
|
||||
return {
|
||||
"scope": "jyotishyamitra_pinned_adapter_probe",
|
||||
"oracle": "jyotishyamitra",
|
||||
"version": VERSION,
|
||||
"source_commit": commit,
|
||||
"license": LICENSE,
|
||||
"wheel_path": str(wheel_path),
|
||||
"status": "blocked",
|
||||
"blocked_reason": "fixture_missing",
|
||||
"truth_policy": "independent_observation_not_truth",
|
||||
"promotion_allowed": False,
|
||||
"boundary": "Wheel fixture is required; adapter must not download at test/runtime or use fake zip evidence.",
|
||||
}
|
||||
raw = raw or {}
|
||||
normalized = normalize_raw(raw)
|
||||
wheel_hash = _sha256_bytes(wheel_path.read_bytes()) if wheel_path.exists() else None
|
||||
meta = wheel_metadata(wheel_path) if wheel_path.exists() else {}
|
||||
return {
|
||||
"scope": "jyotishyamitra_pinned_adapter_probe",
|
||||
"oracle": "jyotishyamitra",
|
||||
"version": VERSION,
|
||||
"source_commit": commit,
|
||||
"source_url": f"https://github.com/VicharaVandana/jyotishyamitra/commit/{commit}",
|
||||
"package_url": "https://pypi.org/project/jyotishyamitra/1.4.0/",
|
||||
"license": LICENSE,
|
||||
"package_metadata": meta,
|
||||
"python_runtime": {
|
||||
"executable": sys.executable,
|
||||
"version": sys.version.split()[0],
|
||||
"wheel_tags": meta.get("wheel_tags", []),
|
||||
},
|
||||
"canonical_request": request or {},
|
||||
"isolated_subprocess": {
|
||||
"used": isolated_run is not None,
|
||||
"returncode": None if isolated_run is None else isolated_run["returncode"],
|
||||
"temporary_install_path": None if isolated_run is None else isolated_run["install_path"],
|
||||
"stderr": None if isolated_run is None else isolated_run["stderr"],
|
||||
},
|
||||
"wheel_path": str(wheel_path),
|
||||
"wheel_sha256": wheel_hash,
|
||||
"raw_sha256": _sha256_bytes(_stable_json(raw).encode("utf-8")),
|
||||
"normalized_raw_sha256": _sha256_bytes(_stable_json(normalized).encode("utf-8")),
|
||||
"normalization": {
|
||||
"volatile_paths": ["$.Dashas.Vimshottari.current.date"],
|
||||
"purpose": "remove run timestamp before replay comparison",
|
||||
},
|
||||
"schema_fingerprint": schema_fingerprint(raw),
|
||||
"raw": raw,
|
||||
"comparison": comparison or {},
|
||||
"truth_policy": "independent_observation_not_truth",
|
||||
"promotion_allowed": False,
|
||||
"status": "stable_raw_ready_as_independent_observation" if raw and not (isinstance(raw, dict) and raw.get("status") == "INPUT_ERROR") else "metadata_only_or_input_blocked",
|
||||
"boundary": "Adapter calls the installed MIT package as an oracle; it does not copy implementation or promote conclusions.",
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--wheel", type=Path, default=Path("/tmp/jyotishyamitra_probe/jyotishyamitra-1.4.0-py3-none-any.whl"))
|
||||
parser.add_argument("--case-json", type=Path)
|
||||
parser.add_argument("--output", type=Path)
|
||||
args = parser.parse_args()
|
||||
raw = {}
|
||||
request = {}
|
||||
isolated = None
|
||||
if args.case_json:
|
||||
case = json.loads(args.case_json.read_text(encoding="utf-8"))
|
||||
request = canonical_request(case)
|
||||
isolated = run_isolated(args.wheel, case)
|
||||
raw = isolated["raw"]
|
||||
report = build_report(args.wheel, raw=raw, request=request, isolated_run=isolated)
|
||||
text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(text, encoding="utf-8")
|
||||
sys.stdout.write(text)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,86 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Audit whether candidate sources can promote day/month timing claims."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_SOURCES = [
|
||||
{
|
||||
"name": "Wikidata/EventKG/BiographyNet derived timelines",
|
||||
"positive_events": True,
|
||||
"explicit_non_event_intervals": False,
|
||||
"independent_human_reviewed": False,
|
||||
"observed_before_preregistration": True,
|
||||
"notes": "Useful for positive event dates; missing events are not non-event labels.",
|
||||
},
|
||||
{
|
||||
"name": "existing 40 control dates",
|
||||
"positive_events": False,
|
||||
"explicit_non_event_intervals": True,
|
||||
"independent_human_reviewed": False,
|
||||
"observed_before_preregistration": True,
|
||||
"notes": "May remain diagnostic only; cannot tune or promote claims.",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def evaluate_source(source: dict) -> dict:
|
||||
blockers = []
|
||||
if not source.get("explicit_non_event_intervals"):
|
||||
blockers.append("missing_explicit_non_event_intervals")
|
||||
if not source.get("independent_human_reviewed"):
|
||||
blockers.append("not_independently_human_reviewed")
|
||||
if source.get("observed_before_preregistration"):
|
||||
blockers.append("observed_before_preregistration")
|
||||
return {
|
||||
**source,
|
||||
"usable_for_promotion": not blockers,
|
||||
"blockers": blockers,
|
||||
}
|
||||
|
||||
|
||||
def build_report(sources: list[dict] | None = None) -> dict:
|
||||
rows = [evaluate_source(source) for source in (sources or DEFAULT_SOURCES)]
|
||||
usable = [row for row in rows if row["usable_for_promotion"]]
|
||||
return {
|
||||
"scope": "timing_negative_holdout_source_audit",
|
||||
"claim_status": "exploratory_unvalidated" if not usable else "ready_for_blind_holdout",
|
||||
"timing_precision": "candidate_day_window",
|
||||
"production_tuning_allowed": False if not usable else True,
|
||||
"candidate_window_policy": (
|
||||
"Return ranked candidate days/months with signals and confidence caps; "
|
||||
"do not label them verified predictions until blind negative holdout passes."
|
||||
),
|
||||
"required_label_contract": {
|
||||
"explicit_non_event_intervals": True,
|
||||
"independent_human_reviewed": True,
|
||||
"unobserved_before_preregistration": True,
|
||||
"positive_and_negative_split_locked_before_scoring": True,
|
||||
},
|
||||
"sources": rows,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--source-json", type=Path)
|
||||
parser.add_argument("--output", type=Path)
|
||||
args = parser.parse_args()
|
||||
sources = None
|
||||
if args.source_json:
|
||||
sources = json.loads(args.source_json.read_text(encoding="utf-8"))["sources"]
|
||||
report = build_report(sources)
|
||||
text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(text, encoding="utf-8")
|
||||
sys.stdout.write(text)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,84 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Archive reproducible VedAstro package identity and hosted-version gap."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.request import urlopen
|
||||
|
||||
|
||||
REGISTRATION_URL = (
|
||||
"https://api.nuget.org/v3/registration5-semver1/"
|
||||
"vedastro.library/{version}.json"
|
||||
)
|
||||
|
||||
|
||||
def _json_url(url: str, timeout: float) -> dict:
|
||||
with urlopen(url, timeout=timeout) as response: # noqa: S310 - fixed public NuGet URL.
|
||||
return json.loads(response.read().decode("utf-8"))
|
||||
|
||||
|
||||
def build_archive(version: str = "1.2.0", timeout: float = 20.0) -> dict:
|
||||
registration = _json_url(REGISTRATION_URL.format(version=version), timeout)
|
||||
catalog_url = registration["catalogEntry"]
|
||||
catalog = _json_url(catalog_url, timeout)
|
||||
deps = []
|
||||
for group in catalog.get("dependencyGroups") or []:
|
||||
target = group.get("targetFramework")
|
||||
for dep in group.get("dependencies") or []:
|
||||
deps.append(
|
||||
{
|
||||
"target_framework": target,
|
||||
"id": dep["id"],
|
||||
"range": dep["range"],
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"scope": "vedastro_reproducible_identity_archive",
|
||||
"package": "VedAstro.Library",
|
||||
"version": version,
|
||||
"nuget_registration_url": REGISTRATION_URL.format(version=version),
|
||||
"nuget_catalog_url": catalog_url,
|
||||
"package_content_url": registration["packageContent"],
|
||||
"package_hash_algorithm": catalog.get("packageHashAlgorithm"),
|
||||
"package_hash": catalog.get("packageHash"),
|
||||
"package_size": catalog.get("packageSize"),
|
||||
"published": catalog.get("published") or registration.get("published"),
|
||||
"catalog_commit_id": catalog.get("catalog:commitId"),
|
||||
"catalog_commit_timestamp": catalog.get("catalog:commitTimeStamp"),
|
||||
"license": catalog.get("licenseExpression"),
|
||||
"project_url": catalog.get("projectUrl"),
|
||||
"dependencies": deps,
|
||||
"self_host_candidate_status": "reproducible_package_identity_archived",
|
||||
"hosted_api_status": "blocked",
|
||||
"hosted_api_blocker": (
|
||||
"api.vedastro.org does not expose a verified build commit, package hash, "
|
||||
"DLL hash, assembly version, container digest, or method-semantics contract."
|
||||
),
|
||||
"boundary": (
|
||||
"This archive fixes a NuGet self-host candidate identity only; it does not "
|
||||
"prove the hosted API is running this package or the same method semantics."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--version", default="1.2.0")
|
||||
parser.add_argument("--timeout", type=float, default=20.0)
|
||||
parser.add_argument("--output", type=Path)
|
||||
args = parser.parse_args()
|
||||
report = build_archive(args.version, args.timeout)
|
||||
text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(text, encoding="utf-8")
|
||||
sys.stdout.write(text)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,104 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compare the native Shadbala chain with VP Jain's published worked example."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
|
||||
EXPECTED = {
|
||||
"sthana": (172.04, 77.17, 184.94, 238.16, 152.98, 198.08, 206.31),
|
||||
"dig": (6.59, 12.22, 20.99, 31.97, 31.99, 53.29, 26.67),
|
||||
"kala": (81.80, 205.85, 158.08, 210.68, 144.22, 135.89, 139.56),
|
||||
"chesta": (0.0, 0.0, 20.93, 28.76, 8.43, 28.18, 5.05),
|
||||
"naisargika": (60.0, 51.43, 17.14, 25.71, 34.29, 42.86, 8.57),
|
||||
"drik": (11.24, -0.32, -5.10, 4.29, 4.32, -2.86, 5.82),
|
||||
}
|
||||
LOCAL_KEYS = {
|
||||
"sthana": ("sthana_bala", "total"),
|
||||
"dig": ("dig_bala",),
|
||||
"kala": ("kala_bala", "total"),
|
||||
"chesta": ("chesta_bala",),
|
||||
"naisargika": ("naisargika_bala",),
|
||||
"drik": ("drik_bala",),
|
||||
}
|
||||
|
||||
|
||||
def _native_output() -> dict:
|
||||
command = [
|
||||
sys.executable, str(ROOT / "scripts" / "jyotish_engine.py"), "shadbala",
|
||||
"--year", "1981", "--month", "9", "--day", "13",
|
||||
"--hour", "1", "--minute", "30", "--lat", "28.65",
|
||||
"--lon", "77.2166666667", "--tz", "5.5", "--ayanamsa", "lahiri",
|
||||
]
|
||||
return json.loads(subprocess.check_output(command, cwd=ROOT, text=True))
|
||||
|
||||
|
||||
def _value(row: dict, path: tuple[str, ...]) -> float:
|
||||
value = row
|
||||
for key in path:
|
||||
value = value[key]
|
||||
return float(value)
|
||||
|
||||
|
||||
def _variant(component: str, planet: str) -> str:
|
||||
if component == "sthana":
|
||||
return "moolatrikona_degree_range_vs_whole_sign"
|
||||
if planet in {"Sun", "Moon"}:
|
||||
return "luminary_chesta_policy"
|
||||
return "mean_motion_seeghrochcha_variant"
|
||||
|
||||
|
||||
def build_report() -> dict:
|
||||
actual = _native_output()["planets"]
|
||||
rows = []
|
||||
for component, expected_values in EXPECTED.items():
|
||||
for planet, expected in zip(PLANETS, expected_values):
|
||||
local = _value(actual[planet], LOCAL_KEYS[component])
|
||||
delta = round(local - expected, 4)
|
||||
matched = abs(delta) <= 1.0
|
||||
rows.append({
|
||||
"component": component,
|
||||
"planet": planet,
|
||||
"unit": "Virupa",
|
||||
"published_value": expected,
|
||||
"local_value": local,
|
||||
"delta": delta,
|
||||
"tolerance": 1.0,
|
||||
"status": "within_tolerance" if matched else "method_variant",
|
||||
"variant": None if matched else _variant(component, planet),
|
||||
})
|
||||
matched = sum(row["status"] == "within_tolerance" for row in rows)
|
||||
return {
|
||||
"case": {
|
||||
"name": "VP Jain published Shadbala example",
|
||||
"birth": "1981-09-13T01:30:00+05:30",
|
||||
"latitude": 28.65,
|
||||
"longitude": 77.2166666667,
|
||||
"ayanamsa": "Lahiri",
|
||||
},
|
||||
"source": {
|
||||
"upstream": "PyJHora 4.8.7 pvr_tests.py::shadbala_VPJainBook_tests",
|
||||
"upstream_issue": "https://github.com/naturalstupid/PyJHora/issues/17",
|
||||
"license_boundary": "Numeric published-example expectations only; no AGPL implementation copied.",
|
||||
},
|
||||
"summary": {
|
||||
"row_count": len(rows),
|
||||
"classified_count": len(rows),
|
||||
"within_tolerance_count": matched,
|
||||
"method_variant_count": len(rows) - matched,
|
||||
"absolute_parity": False,
|
||||
},
|
||||
"rows": rows,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
output = ROOT / "references" / "oracle" / "vp_jain_shadbala_component_benchmark_2026_07_17.json"
|
||||
output.write_text(json.dumps(build_report(), ensure_ascii=False, indent=2) + "\n")
|
||||
print(output)
|
||||
@@ -0,0 +1,77 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Gate Xalen Shadbala/Ashtakavarga variants against external arbitration needs."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def build_report(attribution_path: Path, public_batch_path: Path, ephemeris_path: Path) -> dict:
|
||||
attribution = json.loads(attribution_path.read_text(encoding="utf-8"))
|
||||
public_batch = json.loads(public_batch_path.read_text(encoding="utf-8"))
|
||||
ephemeris = json.loads(ephemeris_path.read_text(encoding="utf-8"))
|
||||
unresolved = [
|
||||
row for row in attribution["rows"]
|
||||
if row.get("truth_status") in {
|
||||
"method_variant_unresolved",
|
||||
"requires_external_worked_example_per_contributor",
|
||||
"defer_until_components_arbitrated",
|
||||
}
|
||||
]
|
||||
enough_cases = public_batch.get("case_count", 0) >= 5
|
||||
independent_ephemeris_ok = (
|
||||
ephemeris.get("maximum_absolute_longitude_delta_deg", 999) <= 0.01
|
||||
and ephemeris.get("varga_difference_count") == 0
|
||||
)
|
||||
return {
|
||||
"scope": "xalen_formula_arbitration_gate",
|
||||
"public_case_count": public_batch.get("case_count", 0),
|
||||
"multi_case_replay_status": "ready" if enough_cases else "partial",
|
||||
"independent_ephemeris_status": "ready" if independent_ephemeris_ok else "partial",
|
||||
"unresolved_variant_count": len(unresolved),
|
||||
"truth_status": "blocked" if unresolved else "ready",
|
||||
"promotion_allowed": not unresolved and enough_cases and independent_ephemeris_ok,
|
||||
"required_next_evidence": [
|
||||
"published component-level Shadbala worked examples for every disputed component",
|
||||
"published BAV contributor-table example with per-sign rows",
|
||||
"formula-source citation for each Xalen/local method branch",
|
||||
],
|
||||
"boundary": (
|
||||
"Multi-case replay and independent ephemeris reduce implementation risk; "
|
||||
"they do not arbitrate method variants without external numeric worked examples."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument(
|
||||
"--attribution",
|
||||
type=Path,
|
||||
default=Path("references/oracle/xalen_formula_unit_attribution_2026_07_17.json"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--public-batch",
|
||||
type=Path,
|
||||
default=Path("references/oracle/xalen_public_case_batch_2026_07_17.json"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ephemeris",
|
||||
type=Path,
|
||||
default=Path("references/oracle/xalen_ephemeris_mode_comparison_2026_07_17.json"),
|
||||
)
|
||||
parser.add_argument("--output", type=Path)
|
||||
args = parser.parse_args()
|
||||
report = build_report(args.attribution, args.public_batch, args.ephemeris)
|
||||
text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(text, encoding="utf-8")
|
||||
sys.stdout.write(text)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user