"""Annual Varshaphala / Tajika report pack contract. The pack normalizes existing annual producers into a stable, audit-friendly shape. It does not adjudicate conflicting annual methods yet; that belongs to the next conflict-gate layer. """ from __future__ import annotations from datetime import datetime, timedelta from types import SimpleNamespace from typing import Any try: # pragma: no cover - import path differs under CLI vs pytest from calculation_profile_contract import build_calculation_profile except ImportError: # pragma: no cover from scripts.calculation_profile_contract import build_calculation_profile try: # pragma: no cover - import path differs under CLI vs pytest from tajika_named_yoga_authority import build_tajika_named_yoga_authority except ImportError: # pragma: no cover from scripts.tajika_named_yoga_authority import build_tajika_named_yoga_authority SCHEMA = "jyotish.annual_tajika_pack.v1" def build_annual_tajika_pack( payload: dict[str, Any], *, solar_return_report: dict[str, Any] | None = None, tajika_report: dict[str, Any] | None = None, pyjhora_replay: dict[str, Any] | None = None, vedastro_reference: dict[str, Any] | None = None, ) -> dict[str, Any]: """Build a JSON-safe annual pack for arbitrary birth payloads.""" profile = build_calculation_profile(payload) args = _args_from_payload(payload) solar_report = solar_return_report if solar_return_report is not None else _safe_solar_return(args) tajika_report = tajika_report if tajika_report is not None else _safe_tajika(args) solar_return = _field_from_report(solar_report, "solar_return", "Solar Return") annual_chart = _annual_chart_from_solar_report(solar_report) muntha = _field_with_conflict_gate("muntha", solar_report, tajika_report) year_lord = _field_with_conflict_gate("year_lord", solar_report, tajika_report) tajika_yogas = _tajika_yogas_field(solar_report) sahams = _field_from_report(solar_report, "sahams", "Sahams") mudda_dasha = _dasha_field("Mudda Dasha", solar_report, tajika_report, "mudda_dasha") patyayini_dasha = _dasha_field("Patyayini Dasha", solar_report, tajika_report, "patyayini_dasha") pack = { "schema": SCHEMA, "profile": { **profile, "target_year": int(payload["target_year"]), }, "solar_return": solar_return, "annual_chart": annual_chart, "muntha": muntha, "year_lord": year_lord, "tajika_yogas": tajika_yogas, "sahams": sahams, "mudda_dasha": mudda_dasha, "patyayini_dasha": patyayini_dasha, "monthly_windows": _monthly_windows(mudda_dasha), "external_engine_comparison": _external_engine_comparison(pyjhora_replay, vedastro_reference), "report_sections": _report_sections( profile, muntha=muntha, year_lord=year_lord, tajika_yogas=tajika_yogas, sahams=sahams, ), "exports": {}, "audit": _audit(profile, solar_report, tajika_report, pyjhora_replay, vedastro_reference), } pack["exports"] = _exports(pack) return pack def _args_from_payload(payload: dict[str, Any]) -> SimpleNamespace: birth = dict(payload.get("birth") or {}) settings = dict(payload.get("settings") or {}) year, month, day = [int(part) for part in str(birth["date"]).split("-")] time_parts = [int(part) for part in str(birth.get("time") or "00:00:00").split(":")] while len(time_parts) < 3: time_parts.append(0) age = payload.get("age") if age is None and payload.get("target_year") is not None: age = int(payload["target_year"]) - year return SimpleNamespace( year=year, month=month, day=day, hour=time_parts[0], minute=time_parts[1], second=time_parts[2], lat=float(birth["latitude"]), lon=float(birth["longitude"]), tz=_offset_to_hours(birth.get("utc_offset")), target_year=int(payload["target_year"]), age=age, mode="all", ayanamsa=settings.get("ayanamsa", "lahiri"), node_mode=settings.get("node_mode", "mean"), house_system=settings.get("house_system", "whole_sign"), position_mode=settings.get("position_mode", "legacy"), dasha_year_days=settings.get("dasha_year_days", 365.25), solar_return_location_mode=settings.get("solar_return_location_mode", "birth_place"), annual_year_policy=settings.get("annual_year_policy", "solar_return_exact"), ) def _offset_to_hours(offset: Any) -> float: if offset is None: return 0.0 if isinstance(offset, (int, float)): return float(offset) text = str(offset).strip() sign = -1 if text.startswith("-") else 1 text = text.lstrip("+-") hours, minutes = [int(part) for part in text.split(":")[:2]] return sign * (hours + minutes / 60) def _safe_solar_return(args: SimpleNamespace) -> dict[str, Any]: try: from scripts.cmd_solar_return import cmd_solar_return except ImportError: # pragma: no cover from cmd_solar_return import cmd_solar_return try: result = cmd_solar_return(args) return result if isinstance(result, dict) else {"error": "solar_return_non_dict"} except Exception as exc: # pragma: no cover - defensive boundary return {"error": str(exc)} def _safe_tajika(args: SimpleNamespace) -> dict[str, Any]: try: from scripts.jyotish_engine import cmd_tajika except ImportError: # pragma: no cover from jyotish_engine import cmd_tajika try: result = cmd_tajika(args) return result if isinstance(result, dict) else {"error": "tajika_non_dict"} except Exception as exc: # pragma: no cover - defensive boundary return {"error": str(exc)} def _field_from_report(report: dict[str, Any], key: str, label: str) -> dict[str, Any]: if not isinstance(report, dict) or report.get("error"): return {"status": "blocked", "producer": label, "reason": report.get("error", "producer_failed")} value = report.get(key) if value is None: return {"status": "blocked", "producer": label, "reason": f"{key}_missing"} if isinstance(value, dict) and value.get("status") in {"blocked", "conflict", "not_applicable", "parameter_sensitive", "partial_verified", "verified"}: normalized = { "status": value.get("status"), "producer": label, "data": value, } if value.get("reason") is not None: normalized["reason"] = value.get("reason") return normalized return {"status": "partial_verified", "producer": label, "data": value} def _annual_chart_from_solar_report(report: dict[str, Any]) -> dict[str, Any]: field = _field_from_report(report, "sr_chart_info", "Solar Return Annual Chart") if field["status"] == "partial_verified": field["data_contract"] = "normalized_from_sr_chart_info" return field def _tajika_yogas_field(solar_report: dict[str, Any]) -> dict[str, Any]: if not isinstance(solar_report, dict) or solar_report.get("error"): return _field_from_report(solar_report, "tajika_yogas", "Tajika Yogas") authority = _build_governed_tajika_named_yoga_authority(solar_report) if authority is None: return _field_from_report(solar_report, "tajika_yogas", "Tajika Yogas") return { "status": "partial_verified" if authority.get("status") == "authority_ready" else authority.get("status", "blocked"), "producer": "Tajika Named Yoga Authority", "data": authority, "data_contract": "governed_named_yoga_authority_surface", } def _build_governed_tajika_named_yoga_authority(solar_report: dict[str, Any]) -> dict[str, Any] | None: planets = _extract_tajika_motion_planets(solar_report) if not planets: return None authority = build_tajika_named_yoga_authority(planets) if not isinstance(authority, dict): return None return authority def _extract_tajika_motion_planets(solar_report: dict[str, Any]) -> dict[str, dict[str, float]]: chart = solar_report.get("chart") if isinstance(solar_report.get("chart"), dict) else {} planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {} extracted: dict[str, dict[str, float]] = {} for name in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"): payload = planets.get(name) if not isinstance(payload, dict): continue longitude = payload.get("degree_raw", payload.get("degree")) speed = payload.get("speed") if longitude is None or speed is None: continue extracted[name] = {"longitude": float(longitude), "speed": float(speed)} return extracted def _first_available_field(label: str, *reports_and_key: Any) -> dict[str, Any]: *reports, key = reports_and_key producers = [] for report in reports: field = _field_from_report(report, key, label) producers.append(field) if field["status"] == "partial_verified": field["producers_checked"] = producers return field return {"status": "blocked", "producer": label, "reason": f"{key}_missing", "producers_checked": producers} def _field_with_conflict_gate(key: str, solar_report: dict[str, Any], tajika_report: dict[str, Any]) -> dict[str, Any]: values = [] for producer, report in (("solar_return", solar_report), ("tajika", tajika_report)): if isinstance(report, dict) and report.get(key) is not None and not report.get("error"): values.append({"producer": producer, "value": report[key]}) if len(values) >= 2 and values[0]["value"] != values[1]["value"]: return { "status": "conflict", "field": key, "values": values, "reason": "same_profile_annual_producers_disagree", } if values: return { "status": "partial_verified", "producer": values[0]["producer"], "data": values[0]["value"], "producers_checked": values, } return { "status": "blocked", "field": key, "reason": f"{key}_missing", "producers_checked": values, } def _dasha_field(label: str, solar_report: dict[str, Any], tajika_report: dict[str, Any], key: str) -> dict[str, Any]: field = _first_available_field(label, solar_report, tajika_report, key) data = field.get("data") if isinstance(field.get("data"), dict) else {} periods = data.get("periods") or data.get("dasha_sequence") or [] field["periods"] = periods if isinstance(periods, list) else [] return field def _monthly_windows(mudda_dasha: dict[str, Any]) -> list[dict[str, Any]]: windows = [] for idx, period in enumerate(mudda_dasha.get("periods", []), start=1): if not isinstance(period, dict): continue windows.append( { "index": idx, "source": "mudda_dasha", "lord": period.get("lord"), "duration_months": period.get("months"), "status": "partial_verified", } ) return windows def _report_sections( profile: dict[str, Any], *, muntha: dict[str, Any] | None = None, year_lord: dict[str, Any] | None = None, tajika_yogas: dict[str, Any] | None = None, sahams: dict[str, Any] | None = None, ) -> dict[str, Any]: blocked_fields = [ name for name, field in (("muntha", muntha), ("year_lord", year_lord)) if isinstance(field, dict) and field.get("status") == "blocked" ] conflict_fields = [ name for name, field in (("muntha", muntha), ("year_lord", year_lord)) if isinstance(field, dict) and field.get("status") == "conflict" ] all_flagged_fields = [ name for name in (*blocked_fields, *conflict_fields) if name ] muntha_brief = _field_brief("Muntha", muntha) year_lord_brief = _field_brief("Year Lord", year_lord) tajika_brief = _field_brief("Tajika Yogas", tajika_yogas) sahams_brief = _field_brief("Sahams", sahams) quick_takeaways = [item for item in (muntha_brief, year_lord_brief) if item] narrative_preview = [ "年度层已经有可读壳层，但仍需要带着冲突标签阅读。", muntha_brief, year_lord_brief, ] if blocked_fields: summary_status = "blocked" summary_reason = "interpretive annual narrative waits for blocked field closure" thematic_status = "blocked" thematic_reason = "annual technique closure incomplete" elif conflict_fields: summary_status = "parameter_sensitive" summary_reason = "interpretive annual narrative remains conflict-labeled until producer adjudication closes" thematic_status = "parameter_sensitive" thematic_reason = "annual technique conflict requires labeled reading" else: summary_status = "partial_verified" summary_reason = "annual shell is readable but still awaits broader external parity closure" thematic_status = "partial_verified" thematic_reason = "annual technique shell available with current native evidence" return { "executive_summary": { "status": summary_status, "reason": summary_reason, "blocked_fields": all_flagged_fields, "summary_lines": narrative_preview, "quick_takeaways": quick_takeaways, }, "thematic_narrative": { "status": thematic_status, "reason": thematic_reason, "highlights": [ "Solar Return / Tajika annual shell is available.", "Muntha and Year Lord are visible, but Year Lord remains producer-disagree sensitive.", "The report should read annual structure first, then the evidence appendix.", ], }, "evidence_appendix": { "status": "partial_verified", "profile_id": profile["profile_id"], "field_briefs": { "muntha": muntha_brief, "year_lord": year_lord_brief, "tajika_yogas": tajika_brief, "sahams": sahams_brief, }, "must_not_claim": ["exact_annual_event_prediction"], }, } def _exports(pack: dict[str, Any]) -> dict[str, Any]: exports = { "json": { "schema": pack["schema"], "profile_id": pack["profile"]["profile_id"], "field_statuses": { key: value.get("status") for key, value in pack.items() if isinstance(value, dict) and "status" in value }, }, "markdown": _markdown_summary(pack), "ai_evidence_bundle": { "schema": "jyotish.annual_tajika_pack.ai_evidence.v1", "profile_id": pack["profile"]["profile_id"], "allowed_claim_status": "blocked_until_conflict_gate", "raw_field_paths": [ "solar_return", "annual_chart", "muntha", "year_lord", "tajika_yogas", "sahams", "mudda_dasha", "patyayini_dasha", ], }, } try: # pragma: no cover - import path differs under CLI vs pytest from report_pack_contract import normalize_report_pack_contract except ImportError: # pragma: no cover from scripts.report_pack_contract import normalize_report_pack_contract exports["unified_report_pack_contract"] = normalize_report_pack_contract( {**pack, "exports": exports}, pack_id="annual_tajika_pack", ) return exports def _markdown_summary(pack: dict[str, Any]) -> str: rows = ["| field | status |", "| --- | --- |"] for key in ( "solar_return", "annual_chart", "muntha", "year_lord", "tajika_yogas", "sahams", "mudda_dasha", "patyayini_dasha", "external_engine_comparison", ): field = pack[key] status = field.get('status') if key == "year_lord": values = field.get("values") or [] if values: first = values[0].get("value") if isinstance(values[0], dict) else {} if isinstance(first, dict): status = f"{status}: {first.get('year_lord') or first.get('year_lord_sign')}" rows.append(f"| {key} | {status} |") summary = [ "", "### Annual Reading Preview", "", f"- blocked fields: {', '.join((pack['report_sections']['executive_summary'].get('blocked_fields') or [])) or 'none'}", ] return "\n".join(rows + summary) def _external_engine_comparison( pyjhora_replay: dict[str, Any] | None, vedastro_reference: dict[str, Any] | None, ) -> dict[str, Any]: if not pyjhora_replay and not vedastro_reference: return { "status": "blocked", "reason": "external annual replay not integrated in annual_tajika_pack.v1", "engines": [], } comparison: dict[str, Any] = {"status": "blocked", "engines": []} statuses = [] if pyjhora_replay: comparison["engines"].append("PyJHora/JHora") pyjhora = dict(pyjhora_replay) patyayini = pyjhora.get("patyayini_dasha") if isinstance(patyayini, dict): patyayini = dict(patyayini) patyayini["normalized_rows"] = _normalize_patyayini_replay_rows(patyayini) pyjhora["patyayini_dasha"] = patyayini comparison["pyjhora"] = pyjhora statuses.append(pyjhora_replay.get("status", "blocked")) year_lord_replay = pyjhora_replay.get("year_lord_replay") if isinstance(year_lord_replay, dict): raw_artifact_path = year_lord_replay.get("raw_artifact_path") comparison["evidence_scope"] = "pyjhora_behavior_only" comparison["parity_status"] = "not_multiengine_parity" comparison["raw_evidence_paths"] = [raw_artifact_path] if raw_artifact_path else [] comparison["raw_evidence_status"] = ( "archived_artifact" if raw_artifact_path else "runtime_observation_not_archived" ) comparison["year_lord_replay"] = { key: year_lord_replay[key] for key in ( "status", "reason", "callable", "pyjhora_version", "effective_ayanamsa", "request_hash", "node_mode", ) if key in year_lord_replay } if vedastro_reference: comparison["engines"].append("VedAstro official") comparison["vedastro"] = _sanitize_vedastro_reference(vedastro_reference) statuses.append(comparison["vedastro"].get("status", "blocked")) comparison["status"] = "partial_verified" if any(str(s).startswith("partial_verified") for s in statuses) else "blocked" return comparison def _normalize_patyayini_replay_rows(field: dict[str, Any]) -> list[dict[str, Any]]: """Expose PyJHora tuple shape without inferring a local Patyayini contract.""" raw = field.get("raw") if isinstance(field.get("raw"), dict) else {} periods = raw.get("periods") if isinstance(raw.get("periods"), list) else [] rows = [] for order, period in enumerate(periods, start=1): if not isinstance(period, (list, tuple)) or len(period) != 3: continue codes, boundary, duration = period if not isinstance(codes, (list, tuple)) or len(codes) != 2: continue if not isinstance(boundary, (list, tuple)) or len(boundary) != 4: continue try: year, month, day = (int(value) for value in boundary[:3]) hour_decimal = float(boundary[3]) boundary_display = (datetime(year, month, day) + timedelta(hours=hour_decimal)).strftime("%Y-%m-%d %H:%M:%S") except (TypeError, ValueError, OverflowError): continue rows.append({ "order": order, "main_code": codes[0], "sub_code": codes[1], "boundary_components": { "year": year, "month": month, "day": day, "hour_decimal": hour_decimal, }, "boundary_display": boundary_display, "duration_raw": duration, "timezone_semantics": "not_returned_by_pyjhora_tuple", "boundary_semantics": "unresolved_external_tuple_boundary", "evidence_status": "pyjhora_behavior_only / not_multiengine_parity", }) return rows def _sanitize_vedastro_reference(reference: dict[str, Any]) -> dict[str, Any]: allowed = { "engine", "status", "chart_core", "dasha_all", "varshaphala_status", "supported_annual_methods", "secret_redaction", } sanitized = {key: value for key, value in reference.items() if key in allowed} sanitized.setdefault("engine", "VedAstro official") sanitized.setdefault("status", "blocked") sanitized.setdefault("varshaphala_status", "blocked") sanitized.setdefault("supported_annual_methods", []) sanitized["secret_redaction"] = {"secret_material_included": False} return sanitized def _field_brief(label: str, field: dict[str, Any] | None) -> str: if not isinstance(field, dict): return f"{label}: unavailable" status = field.get("status", "blocked") if status == "conflict": values = field.get("values") or [] render_values = [] for item in values[:2]: if not isinstance(item, dict): continue value = item.get("value") if isinstance(value, dict): render_values.append( ", ".join( str(part) for part in ( value.get("year_lord"), value.get("year_lord_sign"), value.get("muntha_sign"), value.get("muntha_lord"), ) if part ) ) if render_values: return f"{label}: conflict between { ' vs '.join(render_values) }" if status == "partial_verified": data = field.get("data") if isinstance(field.get("data"), dict) else {} if label == "Muntha": return f"{label}: {data.get('muntha_sign') or data.get('muntha_sign_idx')} / {data.get('muntha_lord') or 'unknown'}" if label == "Year Lord": return f"{label}: {data.get('year_lord') or 'unknown'}" if label == "Tajika Yogas": authority_status = data.get("status") if authority_status == "authority_ready": summary = data.get("summary") if isinstance(data.get("summary"), dict) else {} row_count = summary.get("row_count") supported = data.get("supported_named_yogas") if isinstance(data.get("supported_named_yogas"), list) else [] supported_text = "/".join(str(item) for item in supported) if supported else "governed named yogas" return ( f"Tajika Yogas: governed authority surface visible for {supported_text}" + (f" ({row_count} rows)" if row_count is not None else "") ) return "Tajika Yogas: annual candidate structures visible" if label == "Sahams": return "Sahams: annual sensitive points visible" return f"{label}: {status}" def _audit( profile: dict[str, Any], solar_report: dict[str, Any], tajika_report: dict[str, Any], pyjhora_replay: dict[str, Any] | None = None, vedastro_reference: dict[str, Any] | None = None, ) -> dict[str, Any]: muntha_status = _field_with_conflict_gate("muntha", solar_report, tajika_report).get("status", "blocked") year_lord_status = _field_with_conflict_gate("year_lord", solar_report, tajika_report).get("status", "blocked") audit = { "profile_id": profile["profile_id"], "technique_audit": [ {"technique": "Calculation Profile", "status": "verified", "profile_id": profile["profile_id"]}, {"technique": "Solar Return", "status": _producer_status(solar_report)}, {"technique": "Annual Chart", "status": "partial_verified" if solar_report.get("sr_chart_info") else "blocked"}, {"technique": "Muntha", "status": muntha_status}, {"technique": "Year Lord", "status": year_lord_status}, {"technique": "Tajika Yogas", "status": _tajika_yogas_field(solar_report).get("status", "blocked")}, {"technique": "Sahams", "status": "partial_verified" if solar_report.get("sahams") else "blocked"}, {"technique": "Mudda Dasha", "status": "partial_verified" if (solar_report.get("mudda_dasha") or tajika_report.get("mudda_dasha")) else "blocked"}, {"technique": "Patyayini Dasha", "status": "blocked", "reason": "native producer not integrated"}, {"technique": "External Engine Comparison", "status": "blocked", "reason": "pending Task 5/6"}, ], } if pyjhora_replay: year_lord_replay = pyjhora_replay.get("year_lord_replay") audit["technique_audit"].append( { "technique": "PyJHora Annual Replay", "status": pyjhora_replay.get("status", "blocked"), "license_boundary": pyjhora_replay.get("license_boundary"), **( { "evidence_scope": "pyjhora_behavior_only", "parity_status": "not_multiengine_parity", "raw_artifact_path": year_lord_replay.get("raw_artifact_path"), "raw_evidence_status": ( "archived_artifact" if year_lord_replay.get("raw_artifact_path") else "runtime_observation_not_archived" ), "year_lord_replay_status": year_lord_replay.get("status", "blocked"), "node_mode": year_lord_replay.get("node_mode"), } if isinstance(year_lord_replay, dict) else {} ), } ) if vedastro_reference: vedastro = _sanitize_vedastro_reference(vedastro_reference) audit["technique_audit"].append( { "technique": "VedAstro Annual Boundary", "status": vedastro.get("status", "blocked"), "varshaphala_status": vedastro.get("varshaphala_status", "blocked"), } ) return audit def _producer_status(report: dict[str, Any]) -> str: return "blocked" if report.get("error") else "partial_verified"