root/Jyotisha
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Classify VedAstro capability catalog domains

Browse Source
This commit is contained in:
732642856
2026-07-02 20:12:43 +08:00
parent 28e920d866
commit cdd0f51da5
6 changed files with 331 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/vedastro_evidence_orchestrator.py
+37
View File
@@ -31,6 +31,12 @@ ROUTE_DOMAIN_MAP = {
"career": ["career"],
"finance": ["wealth"],
"wealth": ["wealth"],
"health": ["health"],
"education": ["education"],
"property": ["property"],
"children": ["children"],
"migration": ["migration"],
"prashna": ["prashna"],
"rectification": ["marriage", "career", "wealth"],
"timing": ["career", "marriage", "wealth"],
"general": ["career", "marriage", "wealth"],
@@ -64,6 +70,36 @@ ROUTE_THEME_REQUIREMENTS = {
"requires_dual_dasha": True,
"required_local_supplements": ["wealth_structure_explainer", "narayana_dasha", "functional_benefic_malefic"],
},
"health": {
"route": "health",
"requires_dual_dasha": True,
"required_local_supplements": ["d30_health_axis", "sixth_eighth_twelfth_houses", "functional_benefic_malefic"],
},
"education": {
"route": "education",
"requires_dual_dasha": True,
"required_local_supplements": ["d24_learning_axis", "fifth_ninth_houses", "functional_benefic_malefic"],
},
"property": {
"route": "property",
"requires_dual_dasha": True,
"required_local_supplements": ["d4_property_axis", "fourth_house", "functional_benefic_malefic"],
},
"children": {
"route": "children",
"requires_dual_dasha": True,
"required_local_supplements": ["d7_children_axis", "fifth_house", "functional_benefic_malefic"],
},
"migration": {
"route": "migration",
"requires_dual_dasha": True,
"required_local_supplements": ["d4_d9_foreign_axis", "ninth_twelfth_houses", "functional_benefic_malefic"],
},
"prashna": {
"route": "prashna",
"requires_dual_dasha": False,
"required_local_supplements": ["prashna_chart", "question_context_required", "functional_benefic_malefic"],
},
"overview": {
"route": "overview",
"requires_dual_dasha": True,
@@ -120,6 +156,7 @@ def orchestrate_vedastro_evidence(
window_start, window_end = (start_date, end_date) if start_date and end_date else _default_window(reference_date)
case = _normalize_case(birth_payload)
case["reference_date"] = str(reference_date or window_start)[:10]
case["themes"] = list(domains)
domain_reports: dict[str, Any] = {}
evidence_ledger: list[dict[str, Any]] = []
top_events_by_domain: dict[str, Any] = {}
scripts/vedastro_official_capability_runner.py
+91 -6
View File
@@ -22,7 +22,21 @@ CATALOG_STUB_ENV = "VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB"
PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu", "Ascendant"]
HOUSES = [f"House{i}" for i in range(1, 13)]
DEFAULT_SIGIL_SAMPLE_LIMIT = int(os.environ.get("VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT", "0") or 0)
DOMAIN_ORDER = ["career", "marriage", "wealth", "rectification", "timing", "general"]
DOMAIN_ORDER = [
"career",
"marriage",
"wealth",
"health",
"education",
"property",
"children",
"migration",
"prashna",
"rectification",
"timing",
"general",
"unknown",
]
DEFAULT_DYNAMIC_THEMES = ["career", "marriage", "wealth", "rectification", "timing"]
POLICY_BUCKETS = {
"needs_user_context": "needs_user_context_methods",
@@ -74,7 +88,25 @@ def _domain_routing_for_method(method: str, capability: dict[str, Any], paramete
if any(token in text for token in ("wealth", "money", "finance", "income", "gain", "house2", "house11", "ashtakvarga")):
domains.add("wealth")
priority = "high" if priority == "low" else priority
if any(token in text for token in ("birth", "rectification", "appearance", "body", "height", "shape", "complexion")):
if any(token in text for token in ("health", "illness", "disease", "medical", "medicine", "hospital", "accident", "injury", "surgery")):
domains.add("health")
priority = "high" if any(token in text for token in ("health", "disease", "illness", "accident")) else priority
if any(token in text for token in ("education", "school", "college", "degree", "study", "studies", "learning", "exam")):
domains.add("education")
priority = "high" if any(token in text for token in ("education", "degree", "exam")) else priority
if any(token in text for token in ("property", "vehicle", "home", "house4", "house 4", "land", "realestate", "real estate", "residence")):
domains.add("property")
priority = "high" if any(token in text for token in ("property", "vehicle", "land")) else priority
if any(token in text for token in ("children", "child", "progeny", "putra", "pregnancy", "fertility")):
domains.add("children")
priority = "high" if any(token in text for token in ("children", "progeny", "pregnancy")) else priority
if any(token in text for token in ("foreign", "travel", "migration", "relocation", "abroad", "immigration", "journey")):
domains.add("migration")
priority = "high" if any(token in text for token in ("migration", "relocation", "abroad")) else priority
if any(token in text for token in ("prashna", "horary", "questiontext", "question text", "muhurta")):
domains.add("prashna")
priority = "high" if any(token in text for token in ("prashna", "horary")) else priority
if any(token in text for token in ("rectification", "appearance", "body", "height", "shape", "complexion")):
domains.add("rectification")
if priority == "low":
priority = "medium"
@@ -83,9 +115,6 @@ def _domain_routing_for_method(method: str, capability: dict[str, Any], paramete
if priority == "low":
priority = "medium"
if not domains:
domains.add("general")
if parameter_strategy in {"requires_user_context", "requires_user_text", "requires_rectification_profile"}:
execution_policy = {
"requires_user_context": "needs_user_context",
@@ -97,14 +126,46 @@ def _domain_routing_for_method(method: str, capability: dict[str, Any], paramete
else:
execution_policy = "auto"
if not domains:
domains.add("unknown" if execution_policy == "blocked" else "general")
blocked_reason = None
if execution_policy == "blocked":
blocked_reason = parameter_strategy
elif execution_policy == "needs_user_context":
blocked_reason = "requires_additional_user_context"
elif execution_policy == "needs_user_text":
blocked_reason = "requires_user_text_or_question"
elif execution_policy == "needs_rectification_profile":
blocked_reason = "requires_rectification_profile"
ordered_domains = [domain for domain in DOMAIN_ORDER if domain in domains]
return {
"domains": ordered_domains,
"execution_policy": execution_policy,
"priority": priority,
"adjudicator_use": _adjudicator_use(execution_policy, priority),
"confidence_role": _confidence_role(execution_policy, priority),
"blocked_reason": blocked_reason,
}
def _adjudicator_use(execution_policy: str, priority: str) -> str:
if execution_policy == "auto":
return "primary_candidate" if priority == "high" else "secondary_context"
if execution_policy in {"needs_user_context", "needs_user_text", "needs_rectification_profile"}:
return "secondary_context"
return "not_used"
def _confidence_role(execution_policy: str, priority: str) -> str:
if execution_policy == "auto":
return "confidence_support" if priority == "high" else "background_reference"
if execution_policy in {"needs_user_context", "needs_user_text", "needs_rectification_profile"}:
return "confidence_cap_until_context_available"
return "blocked"
def _build_domain_routing(method_statuses: dict[str, Any]) -> dict[str, Any]:
routing: dict[str, dict[str, Any]] = {}
for method, status in method_statuses.items():
@@ -161,6 +222,13 @@ def _requested_dynamic_themes(payload: dict[str, Any]) -> list[str]:
"婚恋": "marriage",
"婚姻": "marriage",
"财富": "wealth",
"健康": "health",
"教育": "education",
"房产": "property",
"子女": "children",
"迁移": "migration",
"问卜": "prashna",
"卜卦": "prashna",
"校时": "rectification",
"应期": "timing",
}
@@ -217,6 +285,9 @@ def _capability_reference(method: str, status: dict[str, Any], theme: str) -> di
"status": status.get("status"),
"execution_policy": status.get("execution_policy"),
"priority": status.get("priority"),
"adjudicator_use": status.get("adjudicator_use"),
"confidence_role": status.get("confidence_role"),
"blocked_reason": status.get("blocked_reason"),
"domains": status.get("domains") or [],
"bucket": status.get("bucket"),
"signature": status.get("signature"),
@@ -537,7 +608,7 @@ def _full_catalog_method_payload(method: str, capability: dict[str, Any], case:
return None, "requires_user_context"
if any(name in lowered for name in ("bodyheight", "bodyshape", "hair", "lips", "nose", "complexion", "faceshape", "constitution", "personality")):
return None, "requires_rectification_profile"
if any(name in lowered for name in ("rawtextdata", "birthdatarawtext", "inputtext", "textinput", "query", "fullname", "personfullname", "address", "locationname", "ipaddress")):
if any(name in lowered for name in ("rawtextdata", "birthdatarawtext", "inputtext", "textinput", "query", "questiontext", "question", "fullname", "personfullname", "address", "locationname", "ipaddress")):
return None, "requires_user_text"
return None, "unsupported_signature"
@@ -624,6 +695,18 @@ def run_full_capability_catalog(birth_payload: dict[str, Any]) -> dict[str, Any]
domain_routing = _build_domain_routing(method_statuses)
requested_themes = _requested_dynamic_themes(birth_payload)
dynamic_selection = _build_dynamic_selection(method_statuses, domain_routing, requested_themes)
unknown_method_count = sum(
1
for status in method_statuses.values()
if isinstance(status, dict) and status.get("domains") == ["unknown"]
)
misrouted_general_method_count = sum(
1
for status in method_statuses.values()
if isinstance(status, dict)
and "general" in (status.get("domains") or [])
and len(status.get("domains") or []) > 1
)
return {
"runner": "vedastro_official_capability_runner",
@@ -639,6 +722,8 @@ def run_full_capability_catalog(birth_payload: dict[str, Any]) -> dict[str, Any]
"ok_method_count": ok_count,
"unsupported_method_count": unsupported_count,
"blocked_method_count": blocked_count,
"unknown_method_count": unknown_method_count,
"misrouted_general_method_count": misrouted_general_method_count,
"sample_limit": sample_limit,
"domain_routing_count": len(domain_routing),
"dynamic_selection_theme_count": len(dynamic_selection),
scripts/vedastro_service_adapter.py
+3 -2
View File
@@ -2417,6 +2417,7 @@ def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_
"reference_date": case.get("reference_date") or case.get("today") or case.get("transit_date") or case.get("current_date"),
"dasha_levels": case.get("dasha_levels"),
"dasha_precision_hours": case.get("dasha_precision_hours"),
"themes": case.get("themes") or case.get("theme"),
}
manifest = _official_full_snapshot_manifest(user_case, case_id)
budget_started_at = time.monotonic()
@@ -2441,7 +2442,8 @@ def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_
and official_python_bundle.get("status") != "official_snapshot_budget_exhausted"
):
official_python_bundle = _try_official_python_bridge_snapshot_bundle(user_case)
if time.monotonic() - budget_started_at < _timeout_seconds():
budget_exhausted = official_python_bundle.get("status") == "official_snapshot_budget_exhausted"
if not budget_exhausted and time.monotonic() - budget_started_at < _timeout_seconds():
official_full_capability_catalog = _try_official_full_capability_catalog_bundle(user_case)
else:
official_full_capability_catalog = {
@@ -2470,7 +2472,6 @@ def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_
)
endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
network_enabled = os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() in {"1", "true", "yes"}
budget_exhausted = official_python_bundle.get("status") == "official_snapshot_budget_exhausted"
if budget_exhausted and endpoint and network_enabled and _is_official_public_endpoint(endpoint):
result = {
"backend": "vedastro_service_adapter_candidate",
tests/test_vedastro_evidence_orchestrator.py
+67
View File
@@ -190,3 +190,70 @@ def test_vedastro_orchestrator_surfaces_daily_windows_by_domain(monkeypatch) ->
assert result["daily_windows_by_domain"]["career"][0]["date"] == "2026-07-18"
assert result["top_daily_window_by_domain"]["career"]["score"] == 5
def test_vedastro_orchestrator_passes_non_core_themes_to_official_catalog(monkeypatch) -> None:
from scripts import vedastro_evidence_orchestrator as orchestrator
seen_snapshot_cases = []
seen_scan_domains = []
def fake_snapshot(case, *, case_id="user_chart"):
seen_snapshot_cases.append(case)
return {
"status": "partial",
"available": True,
"section_statuses": {},
"source_metadata": {
"official_full_capability_catalog": {
"status": "partial",
"summary": {"catalog_method_count": 641, "unknown_method_count": 0},
"domain_routing": {
"health": {"method_count": 3, "auto_method_count": 1, "high_priority_methods": ["HealthProblemEvent"]},
},
"dynamic_selection": {
"health": {
"requested_theme": "health",
"selected_methods": [
{
"method": "HealthProblemEvent",
"citation_id": "vedastro:health:HealthProblemEvent",
"execution_policy": "auto",
}
],
"report_reference": {
"theme": "health",
"citation_ids": ["vedastro:health:HealthProblemEvent"],
"auto_count": 1,
},
}
},
},
},
}
def fake_scan(case, domain, start_date, end_date, case_id):
seen_scan_domains.append(domain)
return {
"status": "unsupported_range_scan_domain",
"available": False,
"reason": f"Unsupported range scan domain: {domain}",
"event_count": 0,
"evidence_ledger": [],
}
monkeypatch.setattr(orchestrator, "run_official_full_snapshot_for_case", fake_snapshot)
monkeypatch.setattr(orchestrator, "run_range_scan_for_case", fake_scan)
result = orchestrator.orchestrate_vedastro_evidence(
{"year": REDACTED_YEAR, "month": 4, "day": 17, "hour": 14, "minute": 49, "lat": 36.42, "lon": 114.2, "tz": 8},
route="health",
reference_date="2026-06-30",
)
assert seen_snapshot_cases[0]["themes"] == ["health"]
assert seen_scan_domains == ["health"]
assert result["source_metadata"]["official_report_references"]["health"]["citation_ids"] == [
"vedastro:health:HealthProblemEvent"
]
assert result["source_metadata"]["domain_statuses"]["health"] == "unsupported_range_scan_domain"
tests/test_vedastro_official_capability_runner.py
+131
View File
@@ -374,3 +374,134 @@ def test_official_capability_runner_builds_dynamic_theme_selection_and_citations
assert marriage["needs_user_context_methods"][0]["method"] == "MatchReport"
assert marriage["needs_user_context_methods"][0]["citation_id"] == "vedastro:marriage:MatchReport"
assert marriage["report_reference"]["needs_user_context_count"] == 1
def test_full_catalog_classification_marks_non_core_domains_and_unknowns() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_official_capability_runner.py",
"--bundle",
"official_full_capability_catalog",
"--birth-json",
json.dumps(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
"themes": ["health", "education", "property", "children", "migration", "prashna"],
}
),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**dict(**__import__("os").environ),
"VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(
{
"available": True,
"status": "ok",
"capabilities": [
{
"method": "HealthProblemEvent",
"signature": "(birthTime, startTime, endTime)",
"bucket": "event_range",
"parameter_names": ["birthTime", "startTime", "endTime"],
"callable": True,
},
{
"method": "EducationDegreeYoga",
"signature": "(birthTime)",
"bucket": "birth_time_only",
"parameter_names": ["birthTime"],
"callable": True,
},
{
"method": "PropertyHouseVehicleResult",
"signature": "(birthTime)",
"bucket": "birth_time_only",
"parameter_names": ["birthTime"],
"callable": True,
},
{
"method": "ChildrenProgenyPromise",
"signature": "(birthTime)",
"bucket": "birth_time_only",
"parameter_names": ["birthTime"],
"callable": True,
},
{
"method": "ForeignTravelMigrationEvent",
"signature": "(birthTime, startTime, endTime)",
"bucket": "event_range",
"parameter_names": ["birthTime", "startTime", "endTime"],
"callable": True,
},
{
"method": "PrashnaHoraryJudgement",
"signature": "(queryTime, questionText)",
"bucket": "query_text",
"parameter_names": ["queryTime", "questionText"],
"callable": True,
},
{
"method": "OpaqueExperimentalCalculator",
"signature": "(complexPayload)",
"bucket": "opaque",
"parameter_names": ["complexPayload"],
"callable": True,
},
],
"buckets": {
"event_range": {"count": 2, "examples": ["HealthProblemEvent", "ForeignTravelMigrationEvent"]},
"birth_time_only": {"count": 3, "examples": ["EducationDegreeYoga"]},
"query_text": {"count": 1, "examples": ["PrashnaHoraryJudgement"]},
"opaque": {"count": 1, "examples": ["OpaqueExperimentalCalculator"]},
},
}
),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(
{
"HealthProblemEvent": {"available": True, "status": "ok", "result": {}},
"EducationDegreeYoga": {"available": True, "status": "ok", "result": {}},
"PropertyHouseVehicleResult": {"available": True, "status": "ok", "result": {}},
"ChildrenProgenyPromise": {"available": True, "status": "ok", "result": {}},
"ForeignTravelMigrationEvent": {"available": True, "status": "ok", "result": {}},
}
),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
required_fields = {"domains", "execution_policy", "adjudicator_use", "confidence_role", "blocked_reason"}
for status in report["method_statuses"].values():
assert required_fields.issubset(status)
assert "health" in report["method_statuses"]["HealthProblemEvent"]["domains"]
assert "education" in report["method_statuses"]["EducationDegreeYoga"]["domains"]
assert "property" in report["method_statuses"]["PropertyHouseVehicleResult"]["domains"]
assert "children" in report["method_statuses"]["ChildrenProgenyPromise"]["domains"]
assert "migration" in report["method_statuses"]["ForeignTravelMigrationEvent"]["domains"]
assert "prashna" in report["method_statuses"]["PrashnaHoraryJudgement"]["domains"]
assert report["method_statuses"]["PrashnaHoraryJudgement"]["execution_policy"] == "needs_user_text"
assert report["method_statuses"]["PrashnaHoraryJudgement"]["adjudicator_use"] == "secondary_context"
assert report["method_statuses"]["OpaqueExperimentalCalculator"]["domains"] == ["unknown"]
assert report["method_statuses"]["OpaqueExperimentalCalculator"]["execution_policy"] == "blocked"
assert report["method_statuses"]["OpaqueExperimentalCalculator"]["blocked_reason"] == "unsupported_signature"
for domain in ("health", "education", "property", "children", "migration", "prashna"):
assert domain in report["domain_routing"]
assert domain in report["dynamic_selection"]
assert "general" not in report["domain_routing"]
assert report["summary"]["unknown_method_count"] == 1
assert report["summary"]["misrouted_general_method_count"] == 0
tests/test_vedastro_service_adapter_executor.py
+2 -2
View File
@@ -166,7 +166,7 @@ def test_vedastro_official_snapshot_stops_when_foreground_budget_is_exhausted(mo
assert result["status"] == "official_snapshot_budget_exhausted"
assert result["available"] is False
assert result["source_metadata"]["official_python_bundle"]["status"] == "official_snapshot_budget_exhausted"
assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "official_full_capability_catalog_timeout"
assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "official_full_capability_catalog_skipped_budget_exhausted"
def test_vedastro_official_snapshot_skips_bridge_after_runner_consumes_budget(monkeypatch) -> None:
@@ -226,7 +226,7 @@ def test_vedastro_official_snapshot_skips_bridge_after_runner_consumes_budget(mo
assert bridge_called["value"] is False
assert result["status"] == "official_snapshot_budget_exhausted"
assert result["source_metadata"]["official_python_bundle"]["status"] == "official_snapshot_budget_exhausted"
assert result["source_metadata"]["official_full_capability_catalog"]["summary"]["catalog_method_count"] == 641
assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "official_full_capability_catalog_skipped_budget_exhausted"
def test_vedastro_official_snapshot_budget_does_not_mask_mock_rest_endpoint(monkeypatch) -> None: