#!/usr/bin/env python3 """Shared canonical calculation-profile contract for the Jyotish runtime. Single source of truth for calculation-profile identity. Both ``scripts/domain_calculation_service.py`` (CLI/REST/MCP canonical chart) and the CLI wrappers in ``scripts/jyotish_engine.py`` delegate here, so one (birth input, effective settings, observed ephemeris) tuple yields one profile. Contract rules -------------- 1. Deterministic: identical normalized input gives identical ``input_hash`` / ``profile_id`` / ``profile_hash`` across runs, processes and machines. 2. Privacy (AGENTS 6.6): the profile never stores birth data. Birth date/time/place/coordinates fold into the irreversible ``input_hash``; only calculation settings, ``timezone``, ``engine`` and ``algorithm`` remain. 3. One hash algorithm: canonical JSON ``ensure_ascii=True, sort_keys=True, separators=(",",":"), default=str`` — identical to ``domain_calculation_service._canonical_hash``. 4. Observed-only ephemeris: ``engine`` carries the five observed fields (source/flags_verified/flags/provider/policy); ``ephemeris_path`` is never part of the profile or any hash, so profiles are stable across machine paths. No provider is hard-coded (ERR-042/ERR-121). 5. ``profile_id == profile_hash`` (self hash); ``input_hash`` covers the normalized input contract only (no engine observation). 6. Structured errors: unparseable input raises ``CalculationProfileError`` (a ``ValueError``); IANA timezone names are retained as-is. Stdlib-only so it imports identically under direct-script execution and as ``scripts.calculation_profile_contract``. """ from __future__ import annotations import hashlib import json import math from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError SCHEMA = "jyotish.calculation_profile.v1" PROFILE_VERSION = "1.0" DEFAULT_ALGORITHM = "sidereal_natal_chart" DEFAULTS = { "ayanamsa": "lahiri", "node_mode": "mean", "position_mode": "legacy", "house_system": "whole_sign", "dasha_year_days": 365.25, "solar_return_location_mode": "birth_place", "annual_year_policy": "solar_return_exact", "coordinate_precision": "coordinates", } # single truth source: every setting the profile exposes SETTINGS_KEYS = ( "ayanamsa", "node_mode", "position_mode", "house_system", "dasha_year_days", "solar_return_location_mode", "annual_year_policy", ) EPHEMERIS_FIELDS = ( "ephemeris_source", "ephemeris_flags_verified", "ephemeris_flags", "ephemeris_provider", "ephemeris_policy", ) POSITION_MODES = ("legacy", "mean", "apparent") NODE_MODES = ("mean", "true") class CalculationProfileError(ValueError): """Raised for inputs that cannot be normalized without data loss.""" # --------------------------------------------------------------------------- # canonical hashing (unified with domain_calculation_service) # --------------------------------------------------------------------------- def _canonical_hash(value: Any) -> str: encoded = json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _as_float(value: Any, label: str) -> float | None: if value is None or value == "": return None try: result = float(value) except (TypeError, ValueError): raise CalculationProfileError(f"{label} must be numeric, got {value!r}") from None if not math.isfinite(result): raise CalculationProfileError(f"{label} must be finite, got {value!r}") return result def _as_int(value: Any, label: str) -> int: if isinstance(value, bool): raise CalculationProfileError(f"{label} must be an integer, got {value!r}") try: result = int(value) except (TypeError, ValueError): raise CalculationProfileError(f"{label} must be an integer, got {value!r}") from None if isinstance(value, float) and not value.is_integer(): raise CalculationProfileError(f"{label} must be an integer, got {value!r}") return result # --------------------------------------------------------------------------- # input normalization (root and nested birth containers) # --------------------------------------------------------------------------- def _root_or_nested(payload: dict[str, Any], key: str) -> Any: """Read a key from the payload root, then ``birth``, then ``location``.""" for container in (payload, payload.get("birth"), payload.get("location")): if isinstance(container, dict) and container.get(key) is not None: return container[key] return None def _first_not_none(*values: Any) -> Any: return next((value for value in values if value is not None), None) def _normalize_date(payload: dict[str, Any]) -> str | None: value = _root_or_nested(payload, "date") if value is not None: parts = [part for part in str(value).strip().split("-") if part != ""] if len(parts) != 3: raise CalculationProfileError(f"invalid date {value!r}, expected YYYY-MM-DD") year, month, day = (_as_int(part, f"date part {part!r}") for part in parts) if not (1 <= year <= 9999 and 1 <= month <= 12 and 1 <= day <= 31): raise CalculationProfileError(f"invalid date {value!r}, parts out of range") return f"{year:04d}-{month:02d}-{day:02d}" year = _root_or_nested(payload, "year") month = _root_or_nested(payload, "month") day = _root_or_nested(payload, "day") if year is None and month is None and day is None: return None if year is None or month is None or day is None: raise CalculationProfileError("incomplete birth date: year/month/day must be given together") return _normalize_date({"date": f"{_as_int(year, 'year')}-{_as_int(month, 'month')}-{_as_int(day, 'day')}"}) def _normalize_time(payload: dict[str, Any]) -> list[float] | None: value = _root_or_nested(payload, "time") if value is not None: parts = [part for part in str(value).strip().split(":") if part != ""] if len(parts) not in (2, 3): raise CalculationProfileError(f"invalid time {value!r}, expected HH:MM[:SS]") hour = _as_int(parts[0], f"time hour {parts[0]!r}") minute = _as_int(parts[1], f"time minute {parts[1]!r}") second = _as_float(parts[2], f"time second {parts[2]!r}") if len(parts) == 3 else 0.0 if not (0 <= hour <= 23 and 0 <= minute <= 59 and 0 <= second < 60): raise CalculationProfileError(f"invalid time {value!r}, parts out of range") return [float(hour), float(minute), float(second)] hour = _root_or_nested(payload, "hour") minute = _root_or_nested(payload, "minute") second = _root_or_nested(payload, "second") if hour is None and minute is None and second is None: return None if hour is None or minute is None: raise CalculationProfileError("incomplete birth time: hour/minute must be given together") return _normalize_time({"time": f"{_as_int(hour, 'hour')}:{_as_int(minute, 'minute')}:{_as_float(second, 'second') if second is not None else 0.0}"}) def _format_utc_offset(value: Any) -> str: """Normalize a numeric UTC offset to ``+HH:MM`` (or ``-HH:MM``).""" if isinstance(value, str): text = value.strip() if not text: raise CalculationProfileError("empty UTC offset") if ":" in text: sign = -1 if text.startswith("-") else 1 parts = text.lstrip("+-").split(":") hours = _as_float(parts[0], f"offset hour {parts[0]!r}") minutes = _as_float(parts[1], f"offset minute {parts[1]!r}") if len(parts) > 1 else 0.0 total = sign * (hours + minutes / 60.0) else: total = _as_float(text, f"UTC offset {text!r}") elif isinstance(value, bool): raise CalculationProfileError(f"invalid UTC offset {value!r}") else: total = _as_float(value, "UTC offset") if total is None or not -14 <= total <= 14: raise CalculationProfileError(f"invalid UTC offset {value!r}, must be within -14..+14") total_minutes = int(round(total * 60.0)) sign = "+" if total_minutes >= 0 else "-" total_minutes = abs(total_minutes) return f"{sign}{total_minutes // 60:02d}:{total_minutes % 60:02d}" def _normalize_timezone(payload: dict[str, Any]) -> tuple[str | None, str | None]: """Return ``(utc_offset, timezone_name)``. A numeric ``tz``/``utc_offset`` becomes canonical ``+HH:MM``. A non-numeric ``tz`` that looks like an IANA zone (contains ``/``) is retained as the timezone name; an explicit ``timezone`` field wins as the name. """ raw_offset = _first_not_none(_root_or_nested(payload, "tz"), _root_or_nested(payload, "utc_offset")) name = None for key in ("timezone", "timezone_name", "tz_name"): candidate = _root_or_nested(payload, key) if isinstance(candidate, str) and candidate.strip(): name = candidate.strip() break if raw_offset is None or raw_offset == "": return None, name try: return _format_utc_offset(raw_offset), name except CalculationProfileError: if isinstance(raw_offset, str) and "/" in raw_offset: return None, raw_offset.strip() raise def _normalize_birth(payload: dict[str, Any]) -> dict[str, Any]: offset, tz_name = _normalize_timezone(payload) return { "date": _normalize_date(payload), "time": _normalize_time(payload), "utc_offset": offset, "timezone_name": tz_name, "latitude": _as_float( _first_not_none(_root_or_nested(payload, "latitude"), _root_or_nested(payload, "lat")), "latitude", ), "longitude": _as_float( _first_not_none(_root_or_nested(payload, "longitude"), _root_or_nested(payload, "lon")), "longitude", ), "place": _root_or_nested(payload, "place"), "coordinate_precision": _root_or_nested(payload, "coordinate_precision") or DEFAULTS["coordinate_precision"], } def _normalize_settings(payload: dict[str, Any]) -> dict[str, Any]: settings_block = payload.get("settings") settings_block = settings_block if isinstance(settings_block, dict) else {} def _camel(key: str) -> str: head, _, tail = key.partition("_") return head + tail[:1].upper() + tail[1:] def _setting(key: str) -> Any: if key in settings_block and settings_block[key] is not None: return settings_block[key] return _first_not_none(_root_or_nested(payload, key), _root_or_nested(payload, _camel(key))) ayanamsa = str(_setting("ayanamsa") or DEFAULTS["ayanamsa"]).strip().lower() if not ayanamsa: raise CalculationProfileError("ayanamsa must be a non-empty string") node_mode = str(_setting("node_mode") or DEFAULTS["node_mode"]).strip().lower() if node_mode not in NODE_MODES: raise CalculationProfileError(f"node_mode must be one of {NODE_MODES}, got {node_mode!r}") position_mode = str(_setting("position_mode") or DEFAULTS["position_mode"]).strip().lower() if position_mode not in POSITION_MODES: raise CalculationProfileError(f"position_mode must be one of {POSITION_MODES}, got {position_mode!r}") house_system = str(_setting("house_system") or DEFAULTS["house_system"]).strip() if not house_system: raise CalculationProfileError("house_system must be a non-empty string") dasha_year_days_value = _setting("dasha_year_days") if dasha_year_days_value is None: dasha_year_days_value = DEFAULTS["dasha_year_days"] dasha_year_days = _as_float(dasha_year_days_value, "dasha_year_days") if dasha_year_days <= 0: raise CalculationProfileError(f"dasha_year_days must be positive, got {dasha_year_days!r}") solar_return_location_mode = str( _setting("solar_return_location_mode") or DEFAULTS["solar_return_location_mode"] ).strip() or DEFAULTS["solar_return_location_mode"] annual_year_policy = str( _setting("annual_year_policy") or DEFAULTS["annual_year_policy"] ).strip() or DEFAULTS["annual_year_policy"] return { "ayanamsa": ayanamsa, "node_mode": node_mode, "position_mode": position_mode, "house_system": house_system, "dasha_year_days": dasha_year_days, "solar_return_location_mode": solar_return_location_mode, "annual_year_policy": annual_year_policy, } def _effective_ephemeris(payload: dict[str, Any]) -> dict[str, Any]: """Observed ephemeris provenance only; never fabricate a provider. ``ephemeris_path`` is excluded: absolute machine paths must not participate in profile identity or hashing. """ observed: dict[str, Any] = {} for container in (payload.get("engine"), payload.get("ephemeris"), payload.get("meta")): if not isinstance(container, dict): continue for field in EPHEMERIS_FIELDS: if field in container and container[field] is not None: observed[field] = container[field] if observed: observed.setdefault("ephemeris_policy", "observed_provider_recorded") return observed return { "ephemeris_provider": "not_observed", "ephemeris_source": None, "ephemeris_flags": None, "ephemeris_flags_verified": False, "ephemeris_policy": "no_hardcoded_provider_observed_provider_required", } # --------------------------------------------------------------------------- # public builders # --------------------------------------------------------------------------- def build_calculation_profile(payload: dict[str, Any]) -> dict[str, Any]: """Build the canonical profile for any payload (dict or Namespace-shaped). Birth fields may be nested under ``birth``/``location`` or at the root (``year``/``month``/``day``/``hour``/``minute``/``second``/``lat``/``lon``/ ``tz``). They are only hashed into ``input_hash``; the returned profile carries no birth data. Settings are computed once (single truth) and mirrored flat and under ``effective_settings``. """ birth = _normalize_birth(payload) settings = _normalize_settings(payload) offset = birth["utc_offset"] timezone = { "name": birth["timezone_name"] or (f"UTC{offset}" if offset is not None else None), "utc_offset": offset, } algorithm = str(payload.get("algorithm") or DEFAULT_ALGORITHM).strip() or DEFAULT_ALGORITHM input_hash = _canonical_hash({"birth": birth, "settings": settings}) profile: dict[str, Any] = { "schema": SCHEMA, "profile_version": PROFILE_VERSION, "algorithm": algorithm, "effective_settings": {**settings, "timezone_offset": offset}, "timezone": timezone, "engine": _effective_ephemeris(payload), "coordinate_precision": birth["coordinate_precision"], **{key: settings[key] for key in SETTINGS_KEYS}, "input_hash": input_hash, "profile_id": None, "profile_hash": None, } profile_hash = _canonical_hash(profile) profile["profile_id"] = profile_hash profile["profile_hash"] = profile_hash return profile def _payload_from_args(args: Any) -> dict[str, Any]: """Convert an argparse Namespace (or a plain dict) into a root-style payload.""" if isinstance(args, dict): return dict(args) if not all(hasattr(args, name) for name in ("year", "month", "day", "hour", "minute")): raise TypeError( "attach_calculation_profile expects an argparse Namespace or dict, " f"got {type(args).__name__}" ) def _attr(name: str) -> Any: return getattr(args, name, None) return { "year": _attr("year"), "month": _attr("month"), "day": _attr("day"), "hour": _attr("hour"), "minute": _attr("minute"), "second": _attr("second"), "lat": _attr("lat"), "lon": _attr("lon"), "tz": _attr("tz"), "place": _attr("place") or _attr("location_name"), "timezone": _attr("timezone") or _attr("timezone_name"), "settings": { key: _first_not_none(_attr(key), DEFAULTS[key]) for key in SETTINGS_KEYS }, } def _observed_ephemeris_from_result(result: dict[str, Any]) -> dict[str, Any] | None: """Lift observed ephemeris provenance from a computed result. Only the five contract fields are carried; ``ephemeris_path`` (an absolute machine path) is deliberately excluded from profile identity. """ meta = result.get("meta") if isinstance(meta, dict) and meta.get("ephemeris_provider"): return {key: value for key, value in meta.items() if key in EPHEMERIS_FIELDS} contract = result.get("calculation_contract") if isinstance(contract, dict): effective = contract.get("effective") if isinstance(effective, dict) and effective.get("ephemeris_provider"): return {key: value for key, value in effective.items() if key in EPHEMERIS_FIELDS} return None def attach_calculation_profile(result: dict[str, Any], args: Any) -> dict[str, Any]: """Attach the canonical profile to a result in place. Profile metadata and deterministic result-binding metadata are written; every pre-existing business key keeps its value. A canonical profile already present (e.g. from ``domain_calculation_service.compute_chart``) is kept unchanged; otherwise the profile is built from ``args`` plus any observed ephemeris provenance in ``result``. """ existing = result.get("calculation_profile") if isinstance(result, dict) else None if ( isinstance(existing, dict) and isinstance(existing.get("profile_id"), str) and isinstance(existing.get("profile_hash"), str) ): result["calculation_profile_id"] = existing["profile_id"] return result payload = _payload_from_args(args) observed = _observed_ephemeris_from_result(result) if observed: payload["engine"] = observed profile = build_calculation_profile(payload) result["calculation_profile"] = profile result["calculation_profile_id"] = profile["profile_id"] return bind_result_to_profile(result, profile) def bind_result_to_profile(result: dict[str, Any], profile: dict[str, Any]) -> dict[str, Any]: """Bind one concrete calculation result to its normalized input profile.""" if not isinstance(result, dict) or not isinstance(profile, dict): raise TypeError("result and calculation profile must be dictionaries") input_hash = profile.get("input_hash") if not isinstance(input_hash, str) or len(input_hash) != 64: raise ValueError("calculation profile is missing a valid input_hash") result_payload = { key: value for key, value in result.items() if key not in {"result_hash", "result_binding", "calculation_profile", "calculation_profile_id"} } encoded = json.dumps( canonicalize_result_payload({"input_hash": input_hash, "result": result_payload}), ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str, ) result_hash = hashlib.sha256(encoded.encode("utf-8")).hexdigest() result["result_hash"] = result_hash result["result_binding"] = {"input_hash": input_hash, "result_hash": result_hash} return result def canonicalize_result_payload(value: Any) -> Any: """Convert producer mapping keys to deterministic JSON keys before hashing.""" if isinstance(value, dict): normalized: dict[str, Any] = {} for key, child in value.items(): normalized_key = key if isinstance(key, str) else f"__{type(key).__name__}__:{key}" normalized[normalized_key] = canonicalize_result_payload(child) return normalized if isinstance(value, (list, tuple)): return [canonicalize_result_payload(item) for item in value] return value if __name__ == "__main__": import sys sample = { "birth": { "date": "1990-01-01", "time": "12:00:00", "utc_offset": "+08:00", "latitude": 39.9, "longitude": 116.4, }, "settings": {"ayanamsa": "lahiri", "node_mode": "mean"}, } print(json.dumps(build_calculation_profile(sample), ensure_ascii=False, indent=2, default=str)) sys.exit(0)