Jesse_Chen
324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""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 = 15
|
|
_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_fifteen")
|
|
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_fifteen")
|
|
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}")
|