Jyotisha/scripts/flexible_birth_time_profile.py

"""Build a read-only sensitivity profile for an unresolved birth-time window."""

from __future__ import annotations

from copy import deepcopy
from datetime import datetime
from hashlib import sha256
import json
import re
from typing import Any, Mapping, Sequence

SCHEMA_VERSION = "jyotish.flexible_birth_time_profile.v1"
MAX_CANDIDATE_MINUTES = 31
_CLOCK_RE = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d")
_PROHIBITED_AUTHORITY_FIELDS = frozenset({
    "approved_birth_time",
    "final_birth_time",
    "winner",
    "approval",
    "approval_authority",
})
_PROHIBITED_SOURCE_STATUSES = frozenset({"approved", "confirmed"})


class FlexibleBirthTimeProfileError(ValueError):
    """Raised when a candidate window cannot be represented safely."""


def build_flexible_birth_time_profile(
    candidates: Sequence[Mapping[str, Any]],
    *,
    source_reference: Mapping[str, Any],
) -> dict[str, Any]:
    _reject_candidate_window_authority(candidates)
    rows = _normalize_candidates(candidates)
    source = _normalize_source_reference(source_reference)
    stable: dict[str, Any] = {}
    sensitive: dict[str, dict[str, Any]] = {}
    for key in sorted({key for row in rows for key in row["evidence"]}):
        values = {row["candidate_time"]: row["evidence"].get(key) for row in rows}
        if len({_canonical(value) for value in values.values()}) == 1:
            stable[key] = deepcopy(next(iter(values.values())))
        else:
            sensitive[key] = deepcopy(values)
    times = [row["candidate_time"] for row in rows]
    profile_id = _profile_id(times, source["review_id"])
    return {
        "schema_version": SCHEMA_VERSION,
        "flexible_profile_id": profile_id,
        "birth_time_window": {
            "start_time": times[0],
            "end_time": times[-1],
            "candidate_count": len(times),
            "candidate_times": times,
        },
        "candidate_references": [
            {"candidate_id": row["candidate_id"], "candidate_time": row["candidate_time"]}
            for row in rows
        ],
        "stable_evidence": stable,
        "sensitive_evidence": sensitive,
        "source_reference": source,
        "trace": [
            {"kind": "flexible_birth_time_profile", "reference": profile_id},
            {"kind": "candidate_window", "reference": source["review_id"]},
            *(
                {"kind": "candidate", "reference": f"candidate://{row['candidate_id']}"}
                for row in rows
            ),
        ],
        "status": "candidate_window_only",
        "claim_boundary": (
            "Read-only candidate-window comparison. It cannot select or confirm a birth minute, "
            "replace chart identity, or grant authority to a candidate chart."
        ),
    }


def build_flexible_birth_time_profile_from_window(
    *,
    birth_date: str,
    start_time: str,
    end_time: str,
    candidate_times: Sequence[str],
    lat: float,
    lon: float,
    tz: float,
    ayanamsa: str = "raman",
    node_mode: str = "mean",
    source_reference: Mapping[str, Any],
) -> dict[str, Any]:
    start, end = _parse_window(birth_date, start_time, end_time)
    normalized_times = _normalize_candidate_times(birth_date, start, end, candidate_times)
    candidates = [
        _candidate_from_recast(
            candidate_at=value,
            recast=_recast_candidate_layers(
                value, lat=lat, lon=lon, tz=tz, ayanamsa=ayanamsa, node_mode=node_mode,
            ),
            ayanamsa=ayanamsa,
            node_mode=node_mode,
        )
        for value in normalized_times
    ]
    profile = build_flexible_birth_time_profile(candidates, source_reference=source_reference)
    profile["calculation_profile"] = {
        "ayanamsa": ayanamsa,
        "node_mode": node_mode,
        "coordinate_mode": "explicit_lat_lon_tz",
        "candidate_recast": "native_domain_calculation_service",
    }
    return profile


def _recast_candidate_layers(
    candidate: datetime,
    *,
    lat: float,
    lon: float,
    tz: float,
    ayanamsa: str,
    node_mode: str,
) -> dict[str, Any]:
    try:
        import domain_calculation_service
        import jaimini
        import kp_system
        import varga
    except ModuleNotFoundError:  # pragma: no cover - package import
        from scripts import domain_calculation_service, jaimini, kp_system, varga

    chart = domain_calculation_service.compute_chart({
        "year": candidate.year,
        "month": candidate.month,
        "day": candidate.day,
        "hour": candidate.hour,
        "minute": candidate.minute,
        "second": candidate.second,
        "lat": lat,
        "lon": lon,
        "tz": tz,
        "ayanamsa": ayanamsa,
        "node_mode": node_mode,
    })
    planets = {
        name: row["lon"]
        for name, row in (chart.get("planets") or {}).items()
        if name in {"Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu"}
    }
    ascendant = chart.get("ascendant") or {}
    asc_lon = float(ascendant["lon"])
    divisions = [2, 3, 4, 6, 7, 8, 9, 10, 11, 12, 16, 24, 30, 40, 45, 60]
    vargas = varga.calc_all_vargas(planets, asc_lon, divisions=divisions)
    arudha = jaimini.calc_arudha_padas(int(asc_lon // 30), planets)
    padas = arudha.get("padas") or {}
    kp_cusps: dict[str, Any] = {}
    for house_key in ("house_1", "house_4", "house_7", "house_10"):
        degree = ((chart.get("houses") or {}).get(house_key) or {}).get("cusp_degree")
        if degree is None:
            continue
        lords = kp_system.get_kp_lords(float(degree))
        kp_cusps[house_key] = {
            "sign": lords.get("sign"),
            "nakshatra_lord": lords.get("nakshatra_lord"),
            "sub_lord": lords.get("sub_lord"),
        }
    return {
        "ascendant": ascendant,
        "varga_lagna": {key: value.get("Ascendant") or {} for key, value in vargas.items()},
        "arudha": {"A7": padas.get("A7") or {}, "A10": padas.get("A10") or {}, "UL": arudha.get("upapada") or {}},
        "kp_cusps": kp_cusps,
    }


def _candidate_from_recast(
    *,
    candidate_at: datetime,
    recast: Mapping[str, Any],
    ayanamsa: str,
    node_mode: str,
) -> dict[str, Any]:
    evidence: dict[str, Any] = {"D1.ascendant": (recast.get("ascendant") or {}).get("sign")}
    for key, value in (recast.get("varga_lagna") or {}).items():
        if isinstance(key, str) and key.startswith("D") and isinstance(value, Mapping):
            evidence[f"{key.split('_', 1)[0]}.ascendant"] = value.get("sign")
    for key in ("A7", "A10", "UL"):
        value = (recast.get("arudha") or {}).get(key)
        if isinstance(value, Mapping):
            evidence[f"arudha.{key}"] = value.get("sign")
    evidence["KP.cusp_observation"] = recast.get("kp_cusps") or {}
    candidate_time = candidate_at.strftime("%H:%M")
    candidate_id = f"candidate-{candidate_time.replace(':', '')}"
    return {
        "candidate_id": candidate_id,
        "candidate_time": candidate_time,
        "evidence": evidence,
        "trace": [
            {"kind": "candidate_chart_recast", "reference": f"candidate-chart://{candidate_id}"},
            {"kind": "calculation_profile", "reference": f"ayanamsa://{ayanamsa}/node/{node_mode}"},
        ],
    }


def _parse_window(birth_date: str, start_time: str, end_time: str) -> tuple[datetime, datetime]:
    if not _is_hh_mm(start_time) or not _is_hh_mm(end_time):
        raise FlexibleBirthTimeProfileError("birth_date_or_candidate_time_invalid")
    try:
        start = datetime.strptime(f"{birth_date} {start_time}", "%Y-%m-%d %H:%M")
        end = datetime.strptime(f"{birth_date} {end_time}", "%Y-%m-%d %H:%M")
    except (TypeError, ValueError) as exc:
        raise FlexibleBirthTimeProfileError("birth_date_or_candidate_time_invalid") from exc
    if end < start:
        raise FlexibleBirthTimeProfileError("candidate_window_must_not_cross_midnight")
    return start, end


def _normalize_candidate_times(
    birth_date: str,
    start: datetime,
    end: datetime,
    candidate_times: Sequence[str],
) -> list[datetime]:
    if isinstance(candidate_times, (str, bytes)) or not isinstance(candidate_times, Sequence):
        raise FlexibleBirthTimeProfileError("candidate_times_required")
    if len(candidate_times) < 2 or len(candidate_times) > MAX_CANDIDATE_MINUTES:
        raise FlexibleBirthTimeProfileError("candidate_count_must_be_two_to_thirty_one")
    values: list[datetime] = []
    for raw in candidate_times:
        if not _is_hh_mm(raw):
            raise FlexibleBirthTimeProfileError("candidate_time_invalid")
        try:
            value = datetime.strptime(f"{birth_date} {raw}", "%Y-%m-%d %H:%M")
        except (TypeError, ValueError) as exc:
            raise FlexibleBirthTimeProfileError("candidate_time_invalid") from exc
        if value < start or value > end:
            raise FlexibleBirthTimeProfileError("candidate_time_outside_window")
        values.append(value)
    if len(set(values)) != len(values):
        raise FlexibleBirthTimeProfileError("candidate_times_must_be_unique")
    return sorted(values)


def _normalize_candidates(candidates: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
    if isinstance(candidates, (str, bytes)) or not isinstance(candidates, Sequence):
        raise FlexibleBirthTimeProfileError("candidates_required")
    if len(candidates) < 2 or len(candidates) > MAX_CANDIDATE_MINUTES:
        raise FlexibleBirthTimeProfileError("candidate_count_must_be_two_to_thirty_one")
    rows: list[dict[str, Any]] = []
    for candidate in candidates:
        if not isinstance(candidate, Mapping):
            raise FlexibleBirthTimeProfileError("candidate_must_be_mapping")
        candidate_id = candidate.get("candidate_id")
        candidate_time = candidate.get("candidate_time")
        evidence = candidate.get("evidence")
        trace = candidate.get("trace")
        if not isinstance(candidate_id, str) or not candidate_id:
            raise FlexibleBirthTimeProfileError("candidate_id_required")
        if not _is_hh_mm(candidate_time):
            raise FlexibleBirthTimeProfileError("candidate_time_required")
        if not isinstance(evidence, Mapping) or not evidence:
            raise FlexibleBirthTimeProfileError("candidate_evidence_required")
        if not isinstance(trace, list) or not trace:
            raise FlexibleBirthTimeProfileError("candidate_trace_required")
        rows.append({"candidate_id": candidate_id, "candidate_time": candidate_time, "evidence": dict(evidence), "trace": trace})
    if len({row["candidate_id"] for row in rows}) != len(rows):
        raise FlexibleBirthTimeProfileError("candidate_ids_must_be_unique")
    if len({row["candidate_time"] for row in rows}) != len(rows):
        raise FlexibleBirthTimeProfileError("candidate_times_must_be_unique")
    return sorted(rows, key=lambda row: row["candidate_time"])


def _normalize_source_reference(value: Mapping[str, Any]) -> dict[str, str]:
    if not isinstance(value, Mapping):
        raise FlexibleBirthTimeProfileError("source_reference_must_be_mapping")
    _reject_candidate_window_authority(value)
    status = value.get("status")
    if isinstance(status, str) and status.strip().lower() in _PROHIBITED_SOURCE_STATUSES:
        raise FlexibleBirthTimeProfileError("approved_or_confirmed_source_reference_forbidden")
    review_id = value.get("review_id")
    if not isinstance(review_id, str) or not review_id:
        raise FlexibleBirthTimeProfileError("source_review_id_required")
    return {"review_id": review_id, "status": "review_required"}


def _profile_id(candidate_times: Sequence[str], review_id: str) -> str:
    digest = sha256(repr((tuple(candidate_times), review_id)).encode("utf-8")).hexdigest()[:24]
    return f"flexible-birth-time://{digest}"


def _canonical(value: Any) -> str:
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str)


def _is_hh_mm(value: Any) -> bool:
    return isinstance(value, str) and _CLOCK_RE.fullmatch(value) is not None


def _candidate_window_authority_violation(value: Any, path: str = "$") -> str | None:
    if isinstance(value, Mapping):
        for raw_key, item in value.items():
            key = str(raw_key).strip().lower()
            child_path = f"{path}.{raw_key}"
            if key in _PROHIBITED_AUTHORITY_FIELDS:
                return child_path
            if key == "source_reference" and isinstance(item, Mapping):
                status = item.get("status")
                if isinstance(status, str) and status.strip().lower() in _PROHIBITED_SOURCE_STATUSES:
                    return f"{child_path}.status"
            violation = _candidate_window_authority_violation(item, child_path)
            if violation:
                return violation
    elif isinstance(value, (list, tuple)):
        for index, item in enumerate(value):
            violation = _candidate_window_authority_violation(item, f"{path}[{index}]")
            if violation:
                return violation
    return None


def _reject_candidate_window_authority(value: Any) -> None:
    violation = _candidate_window_authority_violation(value)
    if violation:
        raise FlexibleBirthTimeProfileError(f"candidate_window_authority_forbidden:{violation}")