#!/usr/bin/env python3 """User-facing romantic/marriage compatibility contract. This module deliberately does not reuse the historical ``calc_kuja_dosha`` score. That function combines Mars with five pressure planets and contains a male/female asymmetric comparison. It remains available for legacy reports, but is not a valid default for a new relationship product. """ from __future__ import annotations from copy import deepcopy from typing import Any, Dict, Iterable, Mapping, Optional from ashtakoot import ( FRIENDSHIP, NADI, SIGN_LORDS, SIGNS, calculate_ashtakoot, get_nakshatra, ) PRODUCT_VERSION = "marriage_match_product_v1" DEFAULT_PROFILE = "practitioner_consensus_v1" VALID_RELATIONSHIP_TYPES = {"romantic_or_marriage", "romantic", "marriage"} KUJA_PLANET = "Mars" PAPASAMYA_PLANETS = ("Mars", "Saturn", "Sun", "Rahu", "Ketu") # These profiles are deliberately qualitative. The consensus did not support # an invented severity percentage or a sex-specific tolerance threshold. RULE_PROFILES = { "practitioner_consensus_v1": { "label": "Practitioner consensus v1", "kuja_houses": (1, 2, 4, 7, 8, 12), "reference_frames": ("lagna", "moon", "venus"), "exact_house_sign_exceptions": { 2: {"Gemini", "Virgo"}, 4: {"Aries", "Scorpio"}, 7: {"Capricorn", "Cancer"}, 8: {"Sagittarius", "Pisces"}, 12: {"Taurus", "Libra"}, }, "leo_aquarius_exception": "variant_only", "own_or_exalted": "mitigates", "jupiter_contact": "mitigates", "moon_contact": "mitigates", "papasamya_planets": PAPASAMYA_PLANETS, "gender_asymmetry": False, "notes": [ "Kuja is Mars-only; Papasamya is a separately reported five-planet pressure layer.", "Lagna, Moon and Venus are shown separately; they are not converted into an opaque weighted score.", "Own/exalted Mars and Jupiter or Moon contact mitigate a signal in this default profile; they do not erase the raw placement.", "Age, caste and sex-asymmetric comparison rules are not used for a user-facing conclusion.", ], }, "raman_muhurtha_v1": { "label": "B. V. Raman Muhurtha variant", "kuja_houses": (2, 4, 7, 8, 12), "reference_frames": ("lagna", "moon", "venus"), "exact_house_sign_exceptions": { 2: {"Gemini", "Virgo"}, 4: {"Aries", "Scorpio"}, 7: {"Capricorn", "Cancer"}, 8: {"Sagittarius", "Pisces"}, 12: {"Taurus", "Libra"}, }, "leo_aquarius_exception": "cancels", "own_or_exalted": "mitigates", "jupiter_contact": "mitigates", "moon_contact": "mitigates", "papasamya_planets": PAPASAMYA_PLANETS, "gender_asymmetry": False, "notes": [ "Raman-style house set omits the first house. It is a selectable lineage, not the product default.", "The traditional Leo/Aquarius exception is applied only in this variant profile.", ], }, } def available_profiles() -> Dict[str, Dict[str, Any]]: """Return a presentation-safe copy of available rule profiles.""" return deepcopy(RULE_PROFILES) def _profile(profile_id: str) -> Dict[str, Any]: if profile_id not in RULE_PROFILES: raise ValueError("Unknown marriage-match rule profile: %s" % profile_id) return RULE_PROFILES[profile_id] def _lon(value: Any) -> Optional[float]: if isinstance(value, bool): return None if isinstance(value, (int, float)): return float(value) % 360.0 if isinstance(value, Mapping): for key in ("longitude", "degree_raw", "degree", "lon"): if isinstance(value.get(key), (int, float)): return float(value[key]) % 360.0 return None def _sign_index(value: Any) -> Optional[int]: if isinstance(value, str) and value in SIGNS: return SIGNS.index(value) longitude = _lon(value) return int(longitude // 30) if longitude is not None else None def _planet_data(chart: Mapping[str, Any], planet: str) -> Mapping[str, Any]: planets = chart.get("planets", {}) if isinstance(chart, Mapping) else {} value = planets.get(planet, {}) if isinstance(planets, Mapping) else {} return value if isinstance(value, Mapping) else {} def _planet_lon(chart: Mapping[str, Any], planet: str) -> Optional[float]: value = _planet_data(chart, planet) longitude = _lon(value) return longitude if longitude is not None else _lon(chart.get(planet + "_lon")) def _reference_lon(chart: Mapping[str, Any], reference: str) -> Optional[float]: if reference == "lagna": for value in (chart.get("ascendant"), chart.get("lagna"), chart.get("asc_lon")): longitude = _lon(value) if longitude is not None: return longitude return None return _planet_lon(chart, "Moon" if reference == "moon" else "Venus") def _reference_sign(chart: Mapping[str, Any], reference: str) -> Optional[int]: value = _reference_lon(chart, reference) if value is not None: return int(value // 30) if reference == "lagna": # Aries is index 0, which is a valid reference sign rather than a # missing value. Do not use boolean fallback here. ascendant_sign = _sign_index(chart.get("ascendant")) return ascendant_sign if ascendant_sign is not None else _sign_index(chart.get("lagna")) return _sign_index(_planet_data(chart, "Moon" if reference == "moon" else "Venus")) def _planet_sign(chart: Mapping[str, Any], planet: str) -> Optional[int]: lon = _planet_lon(chart, planet) return int(lon // 30) if lon is not None else _sign_index(_planet_data(chart, planet)) def _house_from(reference_sign: Optional[int], planet_sign: Optional[int]) -> Optional[int]: if reference_sign is None or planet_sign is None: return None return ((planet_sign - reference_sign) % 12) + 1 def _same_sign_contact(chart: Mapping[str, Any], planet_a: str, planet_b: str) -> bool: a, b = _planet_sign(chart, planet_a), _planet_sign(chart, planet_b) return a is not None and a == b def _mars_dignity(chart: Mapping[str, Any]) -> str: sign_idx = _planet_sign(chart, "Mars") if sign_idx is None: return "unknown" sign = SIGNS[sign_idx] if sign == "Capricorn": return "exalted" if sign in {"Aries", "Scorpio"}: return "own" if sign == "Cancer": return "debilitated" return "other" def _frame_kuja(chart: Mapping[str, Any], reference: str, profile: Mapping[str, Any]) -> Dict[str, Any]: mars_sign_idx = _planet_sign(chart, "Mars") ref_sign_idx = _reference_sign(chart, reference) house = _house_from(ref_sign_idx, mars_sign_idx) frame = { "reference": reference, "reference_sign": SIGNS[ref_sign_idx] if ref_sign_idx is not None else None, "mars_sign": SIGNS[mars_sign_idx] if mars_sign_idx is not None else None, "mars_house": house, "raw_signal": house in set(profile["kuja_houses"]) if house is not None else None, "status": "not_available" if house is None else "not_triggered", "mitigations": [], } if house is None or not frame["raw_signal"]: return frame mars_sign = frame["mars_sign"] exact_exceptions = profile["exact_house_sign_exceptions"] if mars_sign in exact_exceptions.get(house, set()): frame["status"] = "cancelled_by_profile_exception" frame["mitigations"].append("house_sign_exception") return frame if mars_sign in {"Leo", "Aquarius"}: mode = profile["leo_aquarius_exception"] if mode == "cancels": frame["status"] = "cancelled_by_profile_exception" frame["mitigations"].append("leo_aquarius_profile_exception") return frame if mode == "variant_only": frame["mitigations"].append("leo_aquarius_exception_disputed") frame["status"] = "active" dignity = _mars_dignity(chart) if dignity in {"own", "exalted"}: frame["mitigations"].append("mars_%s" % dignity) if _same_sign_contact(chart, "Mars", "Jupiter"): frame["mitigations"].append("jupiter_conjunction") if _same_sign_contact(chart, "Mars", "Moon"): frame["mitigations"].append("moon_conjunction") return frame def analyze_kuja(chart: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]: """Analyze Mars-only Kuja / Mangal Dosha across visible reference frames.""" profile = _profile(profile_id) if _planet_sign(chart, "Mars") is None: return { "system": "Kuja / Mangal Dosha", "scope": "Mars only", "profile_id": profile_id, "status": "not_available", "reason": "Mars position is required.", "frames": [], } frames = [_frame_kuja(chart, ref, profile) for ref in profile["reference_frames"]] active = [frame for frame in frames if frame["status"] == "active"] cancelled = [frame for frame in frames if frame["status"] == "cancelled_by_profile_exception"] severity = {0: "none", 1: "mild", 2: "material", 3: "strong"}[len(active)] return { "system": "Kuja / Mangal Dosha", "scope": "Mars only", "profile_id": profile_id, "status": "executed", "active_reference_count": len(active), "cancelled_reference_count": len(cancelled), "qualitative_level": severity, "frames": frames, "boundary": "Reference count is a presentation label, not a probability or a compatibility score.", } def analyze_papasamya(chart: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]: """Report five-planet pressure separately from Mars-only Kuja.""" profile = _profile(profile_id) rows = [] for planet in profile["papasamya_planets"]: planet_sign = _planet_sign(chart, planet) if planet_sign is None: rows.append({"planet": planet, "status": "not_available", "frames": []}) continue frames = [] for reference in profile["reference_frames"]: house = _house_from(_reference_sign(chart, reference), planet_sign) frames.append({ "reference": reference, "house": house, "in_pressure_house": house in set(profile["kuja_houses"]) if house is not None else None, }) rows.append({"planet": planet, "status": "executed", "frames": frames}) active = [ {"planet": row["planet"], "reference": frame["reference"], "house": frame["house"]} for row in rows for frame in row.get("frames", []) if frame.get("in_pressure_house") ] return { "system": "Papasamya pressure", "scope": "Mars, Saturn, Sun, Rahu and Ketu; separate from Kuja.", "profile_id": profile_id, "status": "executed" if any(row["status"] == "executed" for row in rows) else "not_available", "planet_rows": rows, "active_placements": active, "boundary": "This is a descriptive pressure layer. It is not renamed as Mangal Dosha and is not reduced to a sex-weighted score.", } def compare_kuja(person_a: Mapping[str, Any], person_b: Mapping[str, Any], profile_id: str = DEFAULT_PROFILE) -> Dict[str, Any]: """Compare two Kuja results without sex-specific tolerance or scoring.""" a, b = analyze_kuja(person_a, profile_id), analyze_kuja(person_b, profile_id) if a["status"] != "executed" or b["status"] != "executed": outcome = "incomplete" else: difference = abs(a["active_reference_count"] - b["active_reference_count"]) outcome = "not_triggered" if not a["active_reference_count"] and not b["active_reference_count"] else ( "broadly_balanced" if difference <= 1 else "asymmetric" ) return { "comparison_method": "neutral_reference_count", "gender_asymmetry_used": False, "outcome": outcome, "person_a": a, "person_b": b, "boundary": "Both-active is a traditional balancing consideration, not an automatic approval or cancellation.", } def _moon_lon(chart: Mapping[str, Any]) -> Optional[float]: longitude = _planet_lon(chart, "Moon") return longitude if longitude is not None else _lon(chart.get("moon_lon")) def _are_natural_friends(lord_a: str, lord_b: str) -> bool: if lord_a == lord_b: return True return lord_b in FRIENDSHIP.get(lord_a, {}).get("friend", []) and lord_a in FRIENDSHIP.get(lord_b, {}).get("friend", []) def _ashtakoota_with_exceptions( person_a: Mapping[str, Any], person_b: Mapping[str, Any], ) -> Dict[str, Any]: moon_a, moon_b = _moon_lon(person_a), _moon_lon(person_b) if moon_a is None or moon_b is None: return {"status": "not_available", "reason": "Both Moon longitudes are required for Ashtakoota."} forward = calculate_ashtakoot(moon_a, moon_b) reverse = calculate_ashtakoot(moon_b, moon_a) # The historical Varna point encodes caste hierarchy. It cannot truthfully # be relabelled as a modern value judgement, so the international product # excludes it rather than deriving a replacement from lunar signs. score_names = ("Vashya", "Tara", "Yoni", "GrahaMaitri", "Gana", "Bhakoot", "Nadi") symmetric_scores = { name: round( (float(forward["scores"][name]) + float(reverse["scores"][name])) / 2.0, 2, ) for name in score_names } international_core = { "method": "International Ashtakoota core v1 (both input orders averaged; classical Varna excluded)", "scores": symmetric_scores, "total_score": round(sum(symmetric_scores.values()), 2), "max_score": 35.0, "match_percentage": round(sum(symmetric_scores.values()) / 35.0 * 100, 1), "person_a_details": { "moon_sign": SIGNS[int(moon_a // 30)], "nakshatra": get_nakshatra(moon_a)["name"], "nakshatra_pada": get_nakshatra(moon_a)["pada"], }, "person_b_details": { "moon_sign": SIGNS[int(moon_b // 30)], "nakshatra": get_nakshatra(moon_b)["name"], "nakshatra_pada": get_nakshatra(moon_b)["pada"], }, "directionality": { "status": "symmetric", "rule": "all included Kutas are averaged across person_a -> person_b and person_b -> person_a", }, "boundary": "This is a mutual-reference score, not a caste hierarchy or an automatic relationship verdict.", } sign_a, sign_b = SIGNS[int(moon_a // 30)], SIGNS[int(moon_b // 30)] nak_a, nak_b = get_nakshatra(moon_a), get_nakshatra(moon_b) bhakoot_exceptions = [] if symmetric_scores["Bhakoot"] == 0: lord_a, lord_b = SIGN_LORDS[sign_a], SIGN_LORDS[sign_b] if lord_a == lord_b: bhakoot_exceptions.append("same_sign_lord") elif _are_natural_friends(lord_a, lord_b): bhakoot_exceptions.append("mutual_natural_friend_sign_lords") nadi_exceptions = [] if symmetric_scores["Nadi"] == 0: if nak_a["index"] == nak_b["index"] and nak_a["pada"] != nak_b["pada"]: nadi_exceptions.append("same_nakshatra_different_pada") if sign_a == sign_b and nak_a["index"] != nak_b["index"]: nadi_exceptions.append("same_rashi_different_nakshatra") return { "status": "executed", "international_core": international_core, "shared_life_dialogue": { "status": "not_scored", "storage": "no_answers_collected", "prompts": [ "What does a committed relationship mean to each of you?", "How do you want to make decisions about money, home, and care responsibilities?", "What boundaries, privacy expectations, and conflict-repair habits matter most?", "How do you each imagine work, family, location, and personal autonomy over the next few years?", ], "boundary": "These are voluntary discussion prompts, not a compatibility test or a score.", }, "classical_varna": { "status": "excluded_from_default", "reason": "Historical caste hierarchy is not used in the international product or its default score.", }, "person_a": {"moon_sign": sign_a, "nakshatra": nak_a}, "person_b": {"moon_sign": sign_b, "nakshatra": nak_b}, "exception_review": { "bhakoot": { "raw_score": symmetric_scores["Bhakoot"], "candidate_cancellations": bhakoot_exceptions, "status": "candidate_mitigation" if bhakoot_exceptions else "none", }, "nadi": { "raw_score": symmetric_scores["Nadi"], "candidate_cancellations": nadi_exceptions, "status": "candidate_mitigation" if nadi_exceptions else "none", }, "yoni": { "raw_score": symmetric_scores["Yoni"], "status": "legacy_three_band_table", "boundary": "The legacy 4/2/0 table remains visible. Intermediate 3/1 grades require a separately versioned source table before becoming a default score.", }, }, } def _structure_status(chart: Mapping[str, Any]) -> Dict[str, Any]: keys = {"d1_relationship", "d9", "upapada", "darakaraka", "relationship_dasha"} supplied = [key for key in keys if chart.get(key) is not None] return { "status": "partial" if supplied else "not_available", "supplied_layers": sorted(supplied), "required_for_complete_reading": ["D1 relationship structure", "D9", "UL", "DK", "relationship timing / Dasha"], "boundary": "36-point matching and Dosha layers do not replace D1/D9/UL/DK or timing review.", } def build_marriage_match_dossier( person_a: Mapping[str, Any], person_b: Mapping[str, Any], *, relationship_type: str = "romantic_or_marriage", profile_id: str = DEFAULT_PROFILE, ) -> Dict[str, Any]: """Build the stable, user-facing love/marriage compatibility payload. Inputs are chart snapshots only. The caller is responsible for calculating D1/D9/timing snapshots; this function never stores birth details. """ if relationship_type not in VALID_RELATIONSHIP_TYPES: raise ValueError("This product supports romantic/marriage matching only.") profile = _profile(profile_id) kuja = compare_kuja(person_a, person_b, profile_id) ashtakoota = _ashtakoota_with_exceptions(person_a, person_b) return { "product": PRODUCT_VERSION, "relationship_type": relationship_type, "rule_profile": { "id": profile_id, "label": profile["label"], "notes": profile["notes"], "gender_asymmetry_used": False, }, "ashtakoota_36": ashtakoota, "kuja_mangal_dosha": kuja, "papasamya": { "person_a": analyze_papasamya(person_a, profile_id), "person_b": analyze_papasamya(person_b, profile_id), }, "relationship_structure": { "person_a": _structure_status(person_a), "person_b": _structure_status(person_b), }, "presentation": { "headline": "Relationship compatibility is presented as parallel traditional layers, not one approval score.", "decision_boundary": "Do not let the 36-point score, Kuja, or Papasamya alone determine a relationship conclusion.", "not_in_scope": ["business partnerships", "family matching", "age-based cancellation", "sex-asymmetric tolerance"], }, "audit": { "kuja": "executed: Mars-only, Lagna/Moon/Venus frames where data is supplied", "papasamya": "executed separately: five-planet pressure placement review", "ashtakoota": ashtakoota["status"], "d1_d9_ul_dk_timing": "input-dependent; no missing layer is fabricated", }, }