fix(rectification): restore classic eight methods and observe KP Placidus cusps

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 15:58:28 +08:00
parent a1914686e8
commit 84e6191fc2
42 changed files with 1488 additions and 185 deletions
@@ -0,0 +1,87 @@
+"""Display-only KP Placidus cusp observation for birth-time rectification.
+
+This lane uses Swiss Ephemeris Placidus house cusps and Krishnamurti ayanamsa.
+It must never mix into ranking, propose, or unique-minute confirmation.
+Whole-sign midpoints are not a substitute.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ayanamsa_utils import apply_ayanamsa, current_ayanamsa_name
+from kp_system import KP_LORDS, get_kp_lords
+from scripts.domain_calculation_service import swiss_ephemeris_lock
+
+SCAN_HOUSES = (1, 4, 7, 10)
+_BLOCKED = {
+    "status": "blocked",
+    "house_system": "placidus",
+    "ayanamsa": "krishnamurti",
+    "houses": {},
+    "indices": {},
+    "note": "Swiss Ephemeris Placidus 宫头无法计算，KP 观察未执行。不计分，不挡提出门。",
+}
+
+
+def _cusp_row(degree: float) -> dict[str, Any]:
+    lords = get_kp_lords(degree)
+    sub_lord = str(lords.get("sub_lord") or "")
+    try:
+        sub_index = KP_LORDS.index(sub_lord)
+    except ValueError:
+        sub_index = None
+    return {
+        "cusp_degree": round(float(degree) % 360.0, 4),
+        "sign": lords.get("sign"),
+        "rasi_lord": lords.get("rasi_lord"),
+        "nakshatra": lords.get("nakshatra"),
+        "nakshatra_lord": lords.get("nakshatra_lord"),
+        "pada": lords.get("pada"),
+        "sub_lord": sub_lord,
+        "sub_sub_lord": lords.get("sub_sub_lord"),
+        "sub_index": sub_index,
+    }
+
+
+def observe_kp_cusps(jd: float | None, lat: float | None, lon: float | None) -> dict[str, Any]:
+    """Return Placidus + Krishnamurti 12-cusp snapshot, or a blocked observation."""
+    if not isinstance(jd, (int, float)) or not isinstance(lat, (int, float)) or not isinstance(lon, (int, float)):
+        return dict(_BLOCKED)
+    if abs(float(lat)) >= 66.0:
+        return dict(_BLOCKED)
+    previous = current_ayanamsa_name()
+    try:
+        import swisseph as swe
+    except ImportError:
+        return dict(_BLOCKED)
+    with swiss_ephemeris_lock():
+        try:
+            apply_ayanamsa("krishnamurti", swe)
+            cusps_raw, _ascmc = swe.houses(float(jd), float(lat), float(lon), b"P")
+            ayanamsa = float(swe.get_ayanamsa(float(jd)))
+            houses: dict[str, dict[str, Any]] = {}
+            indices: dict[str, int] = {}
+            for house in range(1, 13):
+                tropical = float(cusps_raw[house - 1])
+                if tropical != tropical or abs(tropical) > 720:  # NaN / absurd
+                    return dict(_BLOCKED)
+                sidereal = (tropical - ayanamsa) % 360.0
+                row = _cusp_row(sidereal)
+                houses[str(house)] = row
+                if house in SCAN_HOUSES and isinstance(row["sub_index"], int):
+                    indices[f"kp{house}_sub_index"] = row["sub_index"]
+            if len(houses) != 12 or len(indices) != 4:
+                return dict(_BLOCKED)
+            return {
+                "status": "executed",
+                "house_system": "placidus",
+                "ayanamsa": "krishnamurti",
+                "houses": houses,
+                "indices": indices,
+                "note": "已按 Swiss Ephemeris Placidus + Krishnamurti 观察 12 宫头；不计分，不参与提出门或确认门。",
+            }
+        except (TypeError, ValueError, OverflowError, OSError, RuntimeError, ArithmeticError):
+            return dict(_BLOCKED)
+        finally:
+            apply_ayanamsa(previous or "lahiri", swe)