feat: add guarded Prashna and Rangacharya evidence
This commit is contained in:
732642856
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""VedAstro-assisted career timing radar.
|
||||
|
||||
External VedAstro signals are secondary evidence only. They do not change
|
||||
local scores, dominant labels, or final career/prashna adjudication by
|
||||
themselves.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import vedastro_service_adapter
|
||||
|
||||
|
||||
def build_career_radar_packet(case: dict[str, Any], *, start_date: str, end_date: str, case_id: str = "user_chart") -> dict[str, Any]:
|
||||
result = vedastro_service_adapter.run_range_scan_for_case(
|
||||
case,
|
||||
"career",
|
||||
start_date,
|
||||
end_date,
|
||||
case_id=case_id,
|
||||
)
|
||||
policy = result.get("adjudicator_policy") if isinstance(result.get("adjudicator_policy"), dict) else {}
|
||||
can_change_score = bool(policy.get("can_change_score", False))
|
||||
status = result.get("status", "blocked")
|
||||
return {
|
||||
"scope": "career_vedastro_radar",
|
||||
"status": "ok" if status == "ok" else "blocked",
|
||||
"blocked_reason": None if status == "ok" else result.get("reason") or status,
|
||||
"domain": "career",
|
||||
"adjudicator_use": "secondary_evidence_only",
|
||||
"can_change_score": can_change_score,
|
||||
"can_set_final_verdict": False,
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
"vedastro_range_scan_result": result,
|
||||
"technique_audit_row": {
|
||||
"technique": "VedAstro Career Range Scan",
|
||||
"used": status == "ok",
|
||||
"status": status,
|
||||
"role": "external_secondary_evidence",
|
||||
"confidence_effect": "raises_attention_only_not_final_score" if status == "ok" else "blocked_no_effect",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _load_case(path: Path) -> dict[str, Any]:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--case-json", type=Path, required=True)
|
||||
parser.add_argument("--start-date", required=True)
|
||||
parser.add_argument("--end-date", required=True)
|
||||
parser.add_argument("--case-id", default="user_chart")
|
||||
parser.add_argument("--output", type=Path)
|
||||
args = parser.parse_args()
|
||||
packet = build_career_radar_packet(
|
||||
_load_case(args.case_json),
|
||||
start_date=args.start_date,
|
||||
end_date=args.end_date,
|
||||
case_id=args.case_id,
|
||||
)
|
||||
text = json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True)
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(text + "\n", encoding="utf-8")
|
||||
print(text)
|
||||
return 0 if packet["status"] == "ok" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,151 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract text from screenshots without requiring Homebrew-installed Tesseract."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
DEFAULT_SHORTCUT_NAME = "Extract Text from Image"
|
||||
VALID_BACKENDS = {"auto", "manual", "shortcuts", "tesseract"}
|
||||
|
||||
|
||||
def choose_backend(requested: str = "auto") -> str:
|
||||
if requested != "auto":
|
||||
if requested not in VALID_BACKENDS:
|
||||
raise ValueError(f"unsupported backend: {requested}")
|
||||
return requested
|
||||
if shutil.which("shortcuts"):
|
||||
return "shortcuts"
|
||||
if shutil.which("tesseract"):
|
||||
return "tesseract"
|
||||
return "manual"
|
||||
|
||||
|
||||
def _manual_transcript_path(image: Path, transcript_dir: Path | None) -> Path:
|
||||
base = transcript_dir or image.parent
|
||||
return base / f"{image.stem}.txt"
|
||||
|
||||
|
||||
def _extract_manual(image: Path, transcript_dir: Path | None) -> dict[str, Any]:
|
||||
transcript = _manual_transcript_path(image, transcript_dir)
|
||||
if not transcript.is_file():
|
||||
return {
|
||||
"image_path": str(image),
|
||||
"text": "",
|
||||
"backend": "manual",
|
||||
"status": "blocked",
|
||||
"reason": "manual_transcript_missing",
|
||||
"expected_transcript_path": str(transcript),
|
||||
}
|
||||
return {
|
||||
"image_path": str(image),
|
||||
"text": transcript.read_text(encoding="utf-8"),
|
||||
"backend": "manual",
|
||||
"status": "ok",
|
||||
}
|
||||
|
||||
|
||||
def _extract_shortcuts(image: Path, shortcut_name: str) -> dict[str, Any]:
|
||||
if not shutil.which("shortcuts"):
|
||||
return {"image_path": str(image), "text": "", "backend": "shortcuts", "status": "blocked", "reason": "shortcuts_cli_missing"}
|
||||
completed = subprocess.run(
|
||||
["shortcuts", "run", shortcut_name, "-i", str(image)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=120,
|
||||
)
|
||||
text = completed.stdout
|
||||
if completed.returncode != 0:
|
||||
return {
|
||||
"image_path": str(image),
|
||||
"text": text,
|
||||
"backend": "shortcuts",
|
||||
"status": "blocked",
|
||||
"reason": "shortcuts_run_failed",
|
||||
"stderr": completed.stderr.strip(),
|
||||
"shortcut_name": shortcut_name,
|
||||
}
|
||||
return {"image_path": str(image), "text": text, "backend": "shortcuts", "status": "ok"}
|
||||
|
||||
|
||||
def _extract_tesseract(image: Path) -> dict[str, Any]:
|
||||
if not shutil.which("tesseract"):
|
||||
return {"image_path": str(image), "text": "", "backend": "tesseract", "status": "blocked", "reason": "tesseract_missing"}
|
||||
completed = subprocess.run(
|
||||
["tesseract", str(image), "stdout", "-l", "eng+chi_sim"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=120,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
return {
|
||||
"image_path": str(image),
|
||||
"text": completed.stdout,
|
||||
"backend": "tesseract",
|
||||
"status": "blocked",
|
||||
"reason": "tesseract_run_failed",
|
||||
"stderr": completed.stderr.strip(),
|
||||
}
|
||||
return {"image_path": str(image), "text": completed.stdout, "backend": "tesseract", "status": "ok"}
|
||||
|
||||
|
||||
def extract_one(image: Path, *, backend: str = "auto", transcript_dir: Path | None = None, shortcut_name: str = DEFAULT_SHORTCUT_NAME) -> dict[str, Any]:
|
||||
selected = choose_backend(backend)
|
||||
if selected == "manual":
|
||||
return _extract_manual(image, transcript_dir)
|
||||
if selected == "shortcuts":
|
||||
return _extract_shortcuts(image, shortcut_name)
|
||||
if selected == "tesseract":
|
||||
return _extract_tesseract(image)
|
||||
raise ValueError(f"unsupported backend: {selected}")
|
||||
|
||||
|
||||
def extract_many(
|
||||
images: list[Path],
|
||||
*,
|
||||
output: Path | None = None,
|
||||
backend: str = "auto",
|
||||
transcript_dir: Path | None = None,
|
||||
shortcut_name: str = DEFAULT_SHORTCUT_NAME,
|
||||
) -> dict[str, Any]:
|
||||
items = [extract_one(image, backend=backend, transcript_dir=transcript_dir, shortcut_name=shortcut_name) for image in images]
|
||||
if output:
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
output.write_text("\n".join(json.dumps(item, ensure_ascii=False, sort_keys=True) for item in items) + "\n", encoding="utf-8")
|
||||
return {
|
||||
"status": "ok" if items and all(item["status"] == "ok" for item in items) else "blocked",
|
||||
"backend": choose_backend(backend),
|
||||
"items": items,
|
||||
}
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("images", nargs="+", type=Path)
|
||||
parser.add_argument("--backend", choices=sorted(VALID_BACKENDS), default="auto")
|
||||
parser.add_argument("--transcript-dir", type=Path)
|
||||
parser.add_argument("--shortcut-name", default=DEFAULT_SHORTCUT_NAME)
|
||||
parser.add_argument("--output", type=Path, default=Path("scratch/local/ocr_extract/ocr.jsonl"))
|
||||
args = parser.parse_args(argv)
|
||||
report = extract_many(
|
||||
args.images,
|
||||
output=args.output,
|
||||
backend=args.backend,
|
||||
transcript_dir=args.transcript_dir,
|
||||
shortcut_name=args.shortcut_name,
|
||||
)
|
||||
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
|
||||
return 0 if report["status"] == "ok" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Experimental Rangacharya/Jaimini variant.
|
||||
|
||||
All outputs are blocked from adjudication until formula-level validation passes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Mapping
|
||||
|
||||
|
||||
SIGNS = [
|
||||
"Aries",
|
||||
"Taurus",
|
||||
"Gemini",
|
||||
"Cancer",
|
||||
"Leo",
|
||||
"Virgo",
|
||||
"Libra",
|
||||
"Scorpio",
|
||||
"Sagittarius",
|
||||
"Capricorn",
|
||||
"Aquarius",
|
||||
"Pisces",
|
||||
]
|
||||
|
||||
SOURCE_CARDS_PATH = Path(__file__).resolve().parent.parent / "references" / "rangacharya_source_cards.json"
|
||||
|
||||
|
||||
class RangacharyaValidationError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def _source_cards() -> Dict[str, Dict[str, Any]]:
|
||||
try:
|
||||
data = json.loads(SOURCE_CARDS_PATH.read_text(encoding="utf-8"))
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return {}
|
||||
return {str(card.get("id")): dict(card) for card in data.get("cards", []) if card.get("id")}
|
||||
|
||||
|
||||
def _card_meta(card_id: str) -> Dict[str, Any]:
|
||||
card = _source_cards().get(card_id, {})
|
||||
status = str(card.get("status") or "blocked")
|
||||
meta = {
|
||||
"source_card_id": card_id,
|
||||
"source_card_status": status,
|
||||
"validation_status": status,
|
||||
"adjudication_enabled": False,
|
||||
}
|
||||
if status != "source_verified":
|
||||
meta["blocked_reason"] = card.get("blocked_reason") or "source card is not verified for adjudication"
|
||||
return meta
|
||||
|
||||
|
||||
def _sign_name(index: int) -> str:
|
||||
return SIGNS[index % 12]
|
||||
|
||||
|
||||
def _placeholder_pada(label: str, asc_sign_idx: int, source_house: int) -> Dict[str, Any]:
|
||||
sign_idx = (asc_sign_idx + source_house - 1) % 12
|
||||
return {
|
||||
"label": label,
|
||||
"sign": _sign_name(sign_idx),
|
||||
"sign_index": sign_idx,
|
||||
"source_house": source_house,
|
||||
"note": "Rangacharya formula pending source-card implementation",
|
||||
**_card_meta("rangacharya_core_arudha"),
|
||||
}
|
||||
|
||||
|
||||
def calc_rangacharya_variant(asc_sign_idx: int, planet_longitudes: Mapping[str, float]) -> Dict[str, Any]:
|
||||
asc_sign_idx %= 12
|
||||
arudha_padas = {
|
||||
"AL": _placeholder_pada("AL", asc_sign_idx, 1),
|
||||
"A7": _placeholder_pada("A7", asc_sign_idx, 7),
|
||||
"A10": _placeholder_pada("A10", asc_sign_idx, 10),
|
||||
"UL": _placeholder_pada("UL", asc_sign_idx, 12),
|
||||
}
|
||||
return {
|
||||
"variant": "rangacharya",
|
||||
"status": "experimental_not_for_adjudication",
|
||||
"adjudication_enabled": False,
|
||||
"source_status": "transcribed",
|
||||
"active_lagna": {
|
||||
"sign": _sign_name(asc_sign_idx),
|
||||
**_card_meta("active_effective_lagna"),
|
||||
},
|
||||
"effective_lagna": {
|
||||
"sign": _sign_name(asc_sign_idx),
|
||||
**_card_meta("active_effective_lagna"),
|
||||
},
|
||||
"arudha_padas": arudha_padas,
|
||||
"input_planets_present": sorted(planet_longitudes),
|
||||
}
|
||||
|
||||
|
||||
def _flatten(prefix: str, value: Any) -> Dict[str, Any]:
|
||||
if not isinstance(value, dict):
|
||||
return {prefix: value}
|
||||
rows: Dict[str, Any] = {}
|
||||
for key, child in value.items():
|
||||
child_key = f"{prefix}.{key}" if prefix else str(key)
|
||||
rows.update(_flatten(child_key, child))
|
||||
return rows
|
||||
|
||||
|
||||
def diff_current_vs_rangacharya(current: Mapping[str, Any], variant: Mapping[str, Any]) -> Dict[str, Any]:
|
||||
current_flat = _flatten("", dict(current))
|
||||
variant_flat = _flatten("", dict(variant.get("arudha_padas", variant)))
|
||||
differences = []
|
||||
for key in sorted(set(current_flat) | set(variant_flat)):
|
||||
current_value = current_flat.get(key)
|
||||
variant_value = variant_flat.get(key)
|
||||
if current_value != variant_value:
|
||||
differences.append({"key": key, "current": current_value, "rangacharya": variant_value})
|
||||
return {
|
||||
"current_algorithm": "current_jaimini",
|
||||
"variant_algorithm": "rangacharya",
|
||||
"adjudication_enabled": False,
|
||||
"differences": differences,
|
||||
}
|
||||
|
||||
|
||||
def validation_summary(result: Mapping[str, Any]) -> Dict[str, Any]:
|
||||
statuses = []
|
||||
for key, value in _flatten("", dict(result)).items():
|
||||
if key.endswith("validation_status"):
|
||||
statuses.append(str(value))
|
||||
blocking = sorted({status for status in statuses if status != "adjudication_enabled"})
|
||||
return {
|
||||
"adjudication_enabled": bool(result.get("adjudication_enabled")) and not blocking,
|
||||
"blocking_statuses": blocking,
|
||||
}
|
||||
|
||||
|
||||
def assert_adjudication_allowed(result: Mapping[str, Any]) -> None:
|
||||
summary = validation_summary(result)
|
||||
if not summary["adjudication_enabled"]:
|
||||
raise RangacharyaValidationError(
|
||||
"Rangacharya variant is not adjudication-enabled; validation gates are incomplete"
|
||||
)
|
||||
@@ -0,0 +1,58 @@
|
||||
"""Readiness report for the experimental Rangacharya variant."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
CARDS_PATH = ROOT / "references" / "rangacharya_source_cards.json"
|
||||
MANIFEST_PATH = ROOT / "references" / "rangacharya_source_manifest.json"
|
||||
|
||||
|
||||
def _load_json(path: Path) -> Dict[str, Any]:
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
||||
|
||||
def build_report() -> Dict[str, Any]:
|
||||
cards_payload = _load_json(CARDS_PATH)
|
||||
manifest_payload = _load_json(MANIFEST_PATH)
|
||||
cards = {}
|
||||
blocked = []
|
||||
transcribed = []
|
||||
for card in cards_payload.get("cards", []):
|
||||
card_id = str(card.get("id") or "")
|
||||
if not card_id:
|
||||
continue
|
||||
status = str(card.get("status") or "blocked")
|
||||
adjudication_enabled = bool(card.get("adjudication_enabled"))
|
||||
cards[card_id] = {
|
||||
"status": status,
|
||||
"adjudication_enabled": adjudication_enabled,
|
||||
"blocked_reason": card.get("blocked_reason") or "",
|
||||
}
|
||||
if status == "blocked" or not adjudication_enabled:
|
||||
blocked.append(card_id)
|
||||
if status == "transcribed":
|
||||
transcribed.append(card_id)
|
||||
return {
|
||||
"scope": "rangacharya_readiness",
|
||||
"manifest_available": bool(manifest_payload),
|
||||
"source_cards_available": bool(cards_payload),
|
||||
"adjudication_enabled": bool(cards) and not blocked,
|
||||
"card_count": len(cards),
|
||||
"blocked_count": len(blocked),
|
||||
"transcribed_count": len(transcribed),
|
||||
"blocked_cards": blocked,
|
||||
"transcribed_cards": transcribed,
|
||||
"cards": cards,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(json.dumps(build_report(), ensure_ascii=False, indent=2, sort_keys=True))
|
||||
@@ -481,7 +481,17 @@ class UnifiedConsultationOrchestrator:
|
||||
replay_manifest_path = Path(__file__).resolve().parents[1] / "references/real_case_calibration/replay_manifest.json"
|
||||
replay_manifest = validate_real_case_replay_manifest(replay_manifest_path)
|
||||
holdout_manifest_path = Path(__file__).resolve().parents[1] / "references/real_case_calibration/replay_manifest_holdout_v2.json"
|
||||
holdout_manifest = validate_real_case_replay_manifest(holdout_manifest_path)
|
||||
holdout_manifest = (
|
||||
validate_real_case_replay_manifest(holdout_manifest_path)
|
||||
if holdout_manifest_path.exists()
|
||||
else {
|
||||
"status": "blocked",
|
||||
"case_count": 0,
|
||||
"replay_ready_count": 0,
|
||||
"blocked_reason": "holdout_replay_manifest_missing",
|
||||
"path": "references/real_case_calibration/replay_manifest_holdout_v2.json",
|
||||
}
|
||||
)
|
||||
benchmark_path = Path(__file__).resolve().parents[1] / "docs/benchmark/public_real_case_20_case_closure_2026_07_11.json"
|
||||
if benchmark_path.exists():
|
||||
benchmark_payload = json.loads(benchmark_path.read_text(encoding="utf-8"))
|
||||
|
||||
Reference in New Issue
Block a user