refactor strict evidence and extend parity layers
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Shared consultation workflow boundary for API and MCP callers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPTS_DIR = ROOT / "scripts"
|
||||
for path in (ROOT, SCRIPTS_DIR):
|
||||
if str(path) not in sys.path:
|
||||
sys.path.insert(0, str(path))
|
||||
|
||||
|
||||
def execute_consultation_workflow(body: dict[str, Any], *, surface: str = "api") -> dict[str, Any]:
|
||||
from jyotish_api_server import JyotishAPIHandler, execute_consultation_workflow as _execute
|
||||
|
||||
handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
|
||||
return _execute(handler, body=body, surface=surface)
|
||||
|
||||
|
||||
def build_runtime_evidence_helpers(chart: dict[str, Any]) -> dict[str, Any]:
|
||||
from jyotish_api_server import JyotishAPIHandler
|
||||
|
||||
handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
|
||||
return {
|
||||
"vedastro_official": handler._high_rigor_vedastro_official_summary(chart),
|
||||
"vedastro_archive_manifest": handler._compute_vedastro_gateway_archives(),
|
||||
"interpretation_coverage": handler._interpretation_source_runtime_coverage(chart),
|
||||
}
|
||||
@@ -13,7 +13,7 @@ ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from mcp_server import _existing_interpretation_source_pack # noqa: E402
|
||||
from scripts.strict_evidence_service import existing_interpretation_source_pack # noqa: E402
|
||||
|
||||
|
||||
REQUIRED_LAYERS = [
|
||||
@@ -100,7 +100,7 @@ CANDIDATE_KEYWORDS = [
|
||||
|
||||
|
||||
def build_report() -> dict[str, Any]:
|
||||
source_pack = _existing_interpretation_source_pack()
|
||||
source_pack = existing_interpretation_source_pack()
|
||||
inventory = source_pack.get("interpretation_source_inventory") if isinstance(source_pack, dict) else {}
|
||||
if not isinstance(inventory, dict):
|
||||
inventory = {}
|
||||
|
||||
@@ -204,6 +204,13 @@ def _western_evidence_packet_from_body(
|
||||
solar_return_year=timing_request.get('solar_return_year'),
|
||||
secondary_progression_date=timing_request.get('secondary_progression_date'),
|
||||
solar_arc_date=timing_request.get('solar_arc_date'),
|
||||
converse_secondary_progression_date=timing_request.get('converse_secondary_progression_date'),
|
||||
converse_solar_arc_date=timing_request.get('converse_solar_arc_date'),
|
||||
midpoint_date=timing_request.get('midpoint_date'),
|
||||
lunar_return_start_date=timing_request.get('lunar_return_start_date'),
|
||||
duration_scan_start_date=timing_request.get('duration_scan_start_date'),
|
||||
duration_scan_end_date=timing_request.get('duration_scan_end_date'),
|
||||
parans_date=timing_request.get('parans_date'),
|
||||
)
|
||||
if timing:
|
||||
packet['timing_techniques'] = timing
|
||||
@@ -211,7 +218,8 @@ def _western_evidence_packet_from_body(
|
||||
packet['missing_sections'] = [item for item in packet['missing_sections'] if item != 'timing_techniques']
|
||||
packet['boundary'] = (
|
||||
'Native calculations include only explicitly requested transit, solar-return, secondary-progression, '
|
||||
'and solar-arc layers; they do not infer duration, outcomes, or interpretation.'
|
||||
'solar-arc, midpoint, lunar-return and daily duration-scan layers; parans remain blocked until a '
|
||||
'dedicated latitude-aware event solver is implemented. Outputs do not infer outcomes or interpretation.'
|
||||
)
|
||||
return packet
|
||||
except Exception as exc: # pragma: no cover - defensive boundary
|
||||
|
||||
@@ -1200,9 +1200,13 @@ def _base_strict_narrative_payload(route_label, strict, *, fallback_headline, st
|
||||
monthly_frame = strict.get('monthly_adjudication_summary') if isinstance(strict, dict) else {}
|
||||
monthly_frame = monthly_frame if isinstance(monthly_frame, dict) else {}
|
||||
event_judgement = strict.get('event_judgement') if isinstance(strict, dict) else {}
|
||||
event_judgement = event_judgement if isinstance(event_judgement, dict) else {}
|
||||
adjudication = strict.get('adjudication_stages') if isinstance(strict, dict) else {}
|
||||
adjudication = adjudication if isinstance(adjudication, dict) else {}
|
||||
boundary_contract = strict.get('prediction_boundary_contract') if isinstance(strict, dict) else {}
|
||||
boundary_contract = boundary_contract if isinstance(boundary_contract, dict) else {}
|
||||
confidence_boundary = boundary_contract.get('confidence_boundary') if isinstance(boundary_contract, dict) else {}
|
||||
confidence_boundary = confidence_boundary if isinstance(confidence_boundary, dict) else {}
|
||||
confidence_cap = strict.get('confidence_cap') or event_judgement.get('confidence_cap') or 'unknown'
|
||||
dominant_label = event_judgement.get('dominant_label') if isinstance(event_judgement, dict) else None
|
||||
|
||||
@@ -1657,11 +1661,21 @@ def _build_ai_prompt_pack(report):
|
||||
if isinstance(primary_strict_contract, dict)
|
||||
else {}
|
||||
)
|
||||
try:
|
||||
from strict_evidence_service import existing_interpretation_source_pack
|
||||
fallback_source_pack = existing_interpretation_source_pack()
|
||||
except Exception:
|
||||
fallback_source_pack = {}
|
||||
interpretation_source_audit = (
|
||||
primary_audit.get('interpretation_source_pack')
|
||||
if isinstance(primary_audit, dict) and isinstance(primary_audit.get('interpretation_source_pack'), dict)
|
||||
else {}
|
||||
)
|
||||
fallback_domain_layers = (
|
||||
fallback_source_pack.get('domain_invocation_layers')
|
||||
if isinstance(fallback_source_pack, dict) and isinstance(fallback_source_pack.get('domain_invocation_layers'), dict)
|
||||
else {}
|
||||
)
|
||||
guided_topics = modules.get('guided_topics') if isinstance(modules.get('guided_topics'), list) else build_guided_topics(report)
|
||||
capability_evidence_pool = build_capability_evidence_pool_summary()
|
||||
|
||||
@@ -1743,7 +1757,7 @@ def _build_ai_prompt_pack(report):
|
||||
'missing_refs': interpretation_source_audit.get('missing_refs') or [],
|
||||
},
|
||||
'prediction_boundary_contract': primary_prediction_boundary_contract or {},
|
||||
'domain_invocation_layers': primary_domain_invocation_layers or {},
|
||||
'domain_invocation_layers': primary_domain_invocation_layers or fallback_domain_layers or {},
|
||||
'output_template_contract': primary_output_template_contract or {},
|
||||
'mevg_collection_queue': primary_mevg_collection_queue or {},
|
||||
'real_case_calibration_layer': primary_real_case_calibration_layer or {},
|
||||
@@ -2038,20 +2052,20 @@ def _attach_vedastro_official_full_snapshot(report, args):
|
||||
|
||||
def _load_strict_evidence_collector():
|
||||
try:
|
||||
from mcp_server import _collect_strict_evidence as collector
|
||||
from strict_evidence_service import collect_strict_evidence as collector
|
||||
return collector
|
||||
except Exception:
|
||||
mcp_path = os.path.join(ROOT_DIR, 'mcp_server.py')
|
||||
if not os.path.exists(mcp_path):
|
||||
service_path = os.path.join(SCRIPT_DIR, 'strict_evidence_service.py')
|
||||
if not os.path.exists(service_path):
|
||||
raise
|
||||
spec = importlib.util.spec_from_file_location("jyotish_root_mcp_server", mcp_path)
|
||||
spec = importlib.util.spec_from_file_location("jyotish_strict_evidence_service", service_path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise ImportError(f"Unable to load mcp_server from {mcp_path}")
|
||||
raise ImportError(f"Unable to load strict_evidence_service from {service_path}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
collector = getattr(module, "_collect_strict_evidence", None)
|
||||
collector = getattr(module, "collect_strict_evidence", None)
|
||||
if collector is None:
|
||||
raise ImportError("mcp_server._collect_strict_evidence not found")
|
||||
raise ImportError("strict_evidence_service.collect_strict_evidence not found")
|
||||
return collector
|
||||
|
||||
|
||||
|
||||
@@ -200,19 +200,23 @@ def build_status(oracle_file: str) -> dict[str, Any]:
|
||||
"summary": {
|
||||
"shadbala_task_count": len(shadbala_tasks),
|
||||
"external_verified_shadbala_tasks": len(external_verified),
|
||||
"can_claim_shadbala_absolute_closure": True,
|
||||
"external_packet_fields_complete": True,
|
||||
"same_chart_parity_status": "blocked",
|
||||
"same_chart_parity_reason": "PyJHora same-chart replay still mismatches local Shadbala total virupas.",
|
||||
"can_claim_shadbala_absolute_closure": False,
|
||||
"production_tuning_allowed": False,
|
||||
"required_planets": REQUIRED_PLANETS,
|
||||
"required_components": REQUIRED_COMPONENTS,
|
||||
},
|
||||
"first_priority": None,
|
||||
"next_actions": [
|
||||
"Shadbala external absolute-value closure is complete for the current target set.",
|
||||
"Reconcile Shadbala component-level formulas against PyJHora/JHora same-chart raw values.",
|
||||
"Keep global calibration blocked until Tajika/Sahams and other oracle fronts pass validation.",
|
||||
],
|
||||
"boundary": (
|
||||
"This board isolates Shadbala absolute values. Dasha boundary dates are a separate closure task. "
|
||||
"Production tuning remains forbidden until external component-level evidence is complete."
|
||||
"Packet fields are complete for the current target set, but absolute closure remains blocked "
|
||||
"until same-chart parity passes."
|
||||
),
|
||||
}
|
||||
|
||||
@@ -283,15 +287,17 @@ def render_markdown(report: dict[str, Any]) -> str:
|
||||
"",
|
||||
f"- shadbala_task_count: `{summary['shadbala_task_count']}`",
|
||||
f"- external_verified_shadbala_tasks: `{summary['external_verified_shadbala_tasks']}`",
|
||||
f"- external_packet_fields_complete: `{str(summary.get('external_packet_fields_complete', False)).lower()}`",
|
||||
f"- same_chart_parity_status: `{summary.get('same_chart_parity_status', 'not_checked')}`",
|
||||
f"- can_claim_shadbala_absolute_closure: `{str(summary['can_claim_shadbala_absolute_closure']).lower()}`",
|
||||
f"- production_tuning_allowed: `{str(summary['production_tuning_allowed']).lower()}`",
|
||||
"",
|
||||
]
|
||||
if first is None:
|
||||
lines.extend([
|
||||
"## Closure Complete",
|
||||
"## Packet Complete; Parity Blocked",
|
||||
"",
|
||||
"Shadbala external absolute-value closure is complete for the current target set.",
|
||||
"Shadbala external packet fields are complete for the current target set, but same-chart parity is still blocked.",
|
||||
"",
|
||||
"## Next Actions",
|
||||
"",
|
||||
|
||||
@@ -48,7 +48,7 @@ Do not add private birth data, API keys, or desktop oracle screenshots to this p
|
||||
如果没有 VedAstro official_raw_response,请标记 official_blocked 或 local_fallback。
|
||||
如果我提供西方占星导出,请作为 western_oracle_payload 进入统一主链,不要把单边西占信号说成双系统互证。
|
||||
如果我没有西占导出,请自动计算热带本命证据包(ASC/MC、宫位、主要相位、容许度),并明确它只完成本命层;流年、次限、太阳弧、日返仍须单独计算或导入。
|
||||
如需西占时间技术,请传 western_timing:`{"transit_date":"YYYY-MM-DD","solar_return_year":YYYY,"secondary_progression_date":"YYYY-MM-DD","solar_arc_date":"YYYY-MM-DD"}`;当前支持指定日 transit、精确太阳回归、次限行星与真实太阳弧;未实现的角度推进、converse、paran/midpoint 不得标成已用。
|
||||
如需西占时间技术,请传 western_timing:`{"transit_date":"YYYY-MM-DD","solar_return_year":YYYY,"secondary_progression_date":"YYYY-MM-DD","solar_arc_date":"YYYY-MM-DD","converse_secondary_progression_date":"YYYY-MM-DD","converse_solar_arc_date":"YYYY-MM-DD","midpoint_date":"YYYY-MM-DD","lunar_return_start_date":"YYYY-MM-DD","duration_scan_start_date":"YYYY-MM-DD","duration_scan_end_date":"YYYY-MM-DD","parans_date":"YYYY-MM-DD"}`;当前支持指定日 transit、精确太阳回归、次限行星、真实太阳弧、converse 次限/太阳弧、midpoints、月返和每日过境持续窗口;parans 与高级次限角度仍返回 blocked,不得标成已用。
|
||||
|
||||
## Highest Quality Mode
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stable strict-evidence service boundary.
|
||||
|
||||
This module is the import target for engine/API code. The current implementation
|
||||
delegates to the legacy MCP implementation while the large helper stack is being
|
||||
extracted out of `mcp_server.py`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
|
||||
def collect_strict_evidence(route: str, result: dict[str, Any]) -> dict[str, Any]:
|
||||
from mcp_server import _collect_strict_evidence
|
||||
|
||||
return _collect_strict_evidence(route, result)
|
||||
|
||||
|
||||
def existing_interpretation_source_pack() -> dict[str, Any]:
|
||||
from mcp_server import _existing_interpretation_source_pack
|
||||
|
||||
return _existing_interpretation_source_pack()
|
||||
+1
-1
@@ -62,7 +62,7 @@ def varga_map(si, pi, div):
|
||||
o = _odd(si)
|
||||
if div==2: return (4 if o else 3) if pi==0 else (3 if o else 4)
|
||||
if div==3: return (si+pi*4)%12 # Drekkana: same → +4 → +8, no odd/even distinction
|
||||
if div==4: return (si+pi)%12 if o else (si+8+pi)%12
|
||||
if div==4: return (si+pi*3)%12
|
||||
if div==7: return (si+pi)%12 if o else (si+6+pi)%12
|
||||
if div==9:
|
||||
# BPHS Navamsa: movable=same, fixed=9th from sign (+8), dual=5th from sign (+4)
|
||||
|
||||
@@ -187,12 +187,231 @@ def calculate_solar_arc_directions(*, target_date: str, **birth: Any) -> dict[st
|
||||
}
|
||||
|
||||
|
||||
def calculate_converse_secondary_progressions(*, target_date: str, **birth: Any) -> dict[str, Any]:
|
||||
"""Calculate converse progressed planets using one ephemeris day per tropical year backward."""
|
||||
natal_chart = build_tropical_natal_chart(**birth)
|
||||
target_jd, local = _target_jd(target_date, birth["timezone"])
|
||||
birth_jd = _birth_jd(**birth)
|
||||
elapsed_years = (target_jd - birth_jd) / 365.242189
|
||||
progressed_jd = birth_jd - elapsed_years
|
||||
planets = _progressed_planets(progressed_jd)
|
||||
natal_points = {
|
||||
**natal_chart["natal"]["planets"],
|
||||
"ascendant": natal_chart["natal"]["angles"]["ascendant"],
|
||||
"mc": natal_chart["natal"]["angles"]["mc"],
|
||||
}
|
||||
return {
|
||||
"technique": "converse_secondary_progressions",
|
||||
"status": "partial",
|
||||
"method": "one_ephemeris_day_per_tropical_year_backward",
|
||||
"target_date": target_date,
|
||||
"target_local_time": local.isoformat(),
|
||||
"elapsed_tropical_years": round(elapsed_years, 8),
|
||||
"progressed_julian_day_ut": round(progressed_jd, 8),
|
||||
"progressed_planets": planets,
|
||||
"aspects": _cross_aspects(planets, natal_points),
|
||||
"progressed_angles": {
|
||||
"status": "blocked",
|
||||
"reason": "Progressed angle method is not selected; quotidian/solar-arc/Naibod variants are not interchangeable.",
|
||||
},
|
||||
"boundary": "Converse progressed planets only; progressed angles and interpretation remain blocked until a method is selected.",
|
||||
}
|
||||
|
||||
|
||||
def calculate_converse_solar_arc_directions(*, target_date: str, **birth: Any) -> dict[str, Any]:
|
||||
"""Direct natal points backward by the converse secondary-progressed Sun arc."""
|
||||
natal_chart = build_tropical_natal_chart(**birth)
|
||||
progressions = calculate_converse_secondary_progressions(target_date=target_date, **birth)
|
||||
natal_sun = natal_chart["natal"]["planets"]["sun"]["longitude"]
|
||||
progressed_sun = progressions["progressed_planets"]["sun"]["longitude"]
|
||||
arc = _longitude(natal_sun - progressed_sun)
|
||||
natal_points = {
|
||||
**natal_chart["natal"]["planets"],
|
||||
"ascendant": natal_chart["natal"]["angles"]["ascendant"],
|
||||
"mc": natal_chart["natal"]["angles"]["mc"],
|
||||
}
|
||||
directed = {name: _point(point["longitude"] - arc) for name, point in natal_points.items()}
|
||||
return {
|
||||
"technique": "converse_solar_arc_directions",
|
||||
"status": "partial",
|
||||
"method": "converse_secondary_progressed_sun_arc",
|
||||
"target_date": target_date,
|
||||
"natal_sun_longitude": natal_sun,
|
||||
"converse_progressed_sun_longitude": progressed_sun,
|
||||
"converse_solar_arc_degrees": round(arc, 8),
|
||||
"directed_points": directed,
|
||||
"aspects": _cross_aspects(directed, natal_points),
|
||||
"boundary": "Backward solar arc applied to natal planets/ASC/MC. Interpretation and parans remain separate audited layers.",
|
||||
}
|
||||
|
||||
|
||||
def _midpoint_longitude(first: float, second: float) -> float:
|
||||
diff = _longitude(second - first)
|
||||
if diff > 180.0:
|
||||
diff -= 360.0
|
||||
return _longitude(first + diff / 2.0)
|
||||
|
||||
|
||||
def calculate_midpoints(*, target_date: str | None = None, orb: float = 1.5, **birth: Any) -> dict[str, Any]:
|
||||
"""Calculate natal midpoint tree and optional transit conjunction/opposition hits."""
|
||||
natal_chart = build_tropical_natal_chart(**birth)
|
||||
natal_points = {
|
||||
**natal_chart["natal"]["planets"],
|
||||
"ascendant": natal_chart["natal"]["angles"]["ascendant"],
|
||||
"mc": natal_chart["natal"]["angles"]["mc"],
|
||||
}
|
||||
names = [name for name in [*_PLANETS.keys(), "ascendant", "mc"] if name in natal_points]
|
||||
midpoints: dict[str, dict[str, Any]] = {}
|
||||
for index, first_name in enumerate(names):
|
||||
for second_name in names[index + 1:]:
|
||||
key = f"{first_name}/{second_name}"
|
||||
lon = _midpoint_longitude(natal_points[first_name]["longitude"], natal_points[second_name]["longitude"])
|
||||
midpoints[key] = _point(lon)
|
||||
result: dict[str, Any] = {
|
||||
"technique": "midpoints",
|
||||
"status": "used",
|
||||
"method": "shortest_arc_direct_midpoints",
|
||||
"orb_degrees": float(orb),
|
||||
"natal_midpoints": midpoints,
|
||||
"boundary": "Midpoint geometry only; hits are conjunction/opposition contacts, not interpretations.",
|
||||
}
|
||||
if target_date:
|
||||
transit = calculate_transit_to_natal(target_date=target_date, **birth)
|
||||
hits: list[dict[str, Any]] = []
|
||||
for transit_name, transit_point in transit["transit_planets"].items():
|
||||
for midpoint_name, midpoint in midpoints.items():
|
||||
separation = abs(transit_point["longitude"] - midpoint["longitude"])
|
||||
separation = min(separation, 360.0 - separation)
|
||||
for aspect, exact in {"conjunction": 0.0, "opposition": 180.0}.items():
|
||||
hit_orb = abs(separation - exact)
|
||||
if hit_orb <= orb:
|
||||
hits.append({
|
||||
"transit_planet": transit_name,
|
||||
"midpoint": midpoint_name,
|
||||
"aspect": aspect,
|
||||
"orb": round(hit_orb, 6),
|
||||
"separation": round(separation, 6),
|
||||
})
|
||||
result["target_date"] = target_date
|
||||
result["transit_midpoint_hits"] = sorted(hits, key=lambda row: (row["orb"], row["transit_planet"], row["midpoint"]))
|
||||
return result
|
||||
|
||||
|
||||
def calculate_lunar_return(*, start_date: str, **birth: Any) -> dict[str, Any]:
|
||||
"""Find the next exact tropical lunar return after a local start date."""
|
||||
natal_chart = build_tropical_natal_chart(**birth)
|
||||
natal_moon = natal_chart["natal"]["planets"]["moon"]["longitude"]
|
||||
start_jd, _ = _target_jd(start_date, birth["timezone"])
|
||||
return_jd = swe.mooncross_ut(natal_moon, start_jd, swe.FLG_SWIEPH)
|
||||
return_local = _jd_to_local(return_jd, birth["timezone"])
|
||||
return_birth = {
|
||||
**birth,
|
||||
"year": return_local.year,
|
||||
"month": return_local.month,
|
||||
"day": return_local.day,
|
||||
"hour": return_local.hour,
|
||||
"minute": return_local.minute,
|
||||
"second": return_local.second,
|
||||
}
|
||||
return_chart = build_tropical_natal_chart(**return_birth)
|
||||
returned_moon = return_chart["natal"]["planets"]["moon"]["longitude"]
|
||||
delta = abs(_longitude(returned_moon - natal_moon))
|
||||
delta = min(delta, 360.0 - delta)
|
||||
return {
|
||||
"technique": "lunar_return",
|
||||
"status": "used",
|
||||
"method": "Swiss Ephemeris mooncross_ut tropical longitude",
|
||||
"start_date": start_date,
|
||||
"return_julian_day_ut": round(return_jd, 8),
|
||||
"return_local_time": return_local.isoformat(),
|
||||
"natal_moon_longitude": natal_moon,
|
||||
"return_moon_longitude": returned_moon,
|
||||
"moon_longitude_delta": round(delta, 8),
|
||||
"return_chart": return_chart,
|
||||
"boundary": "Exact lunar return time and chart only; monthly topics require separate audited interpretation.",
|
||||
}
|
||||
|
||||
|
||||
def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days: int = 370, **birth: Any) -> dict[str, Any]:
|
||||
"""Scan daily transit-to-natal aspect activity and group consecutive windows."""
|
||||
start = datetime.fromisoformat(start_date)
|
||||
end = datetime.fromisoformat(end_date)
|
||||
if end < start:
|
||||
raise ValueError("end_date must be on or after start_date")
|
||||
days = (end.date() - start.date()).days + 1
|
||||
if days > max_days:
|
||||
raise ValueError(f"duration scan range exceeds max_days={max_days}")
|
||||
daily_hits: list[dict[str, Any]] = []
|
||||
active: dict[tuple[str, str, str], dict[str, Any]] = {}
|
||||
windows: list[dict[str, Any]] = []
|
||||
for offset in range(days):
|
||||
current = (start + timedelta(days=offset)).date().isoformat()
|
||||
transit = calculate_transit_to_natal(target_date=current, **birth)
|
||||
keys = set()
|
||||
for aspect in transit["aspects"]:
|
||||
key = (aspect["transit_planet"], aspect["natal_point"], aspect["aspect"])
|
||||
keys.add(key)
|
||||
if key not in active:
|
||||
active[key] = {"start_date": current, "min_orb": aspect["orb"]}
|
||||
else:
|
||||
active[key]["min_orb"] = min(active[key]["min_orb"], aspect["orb"])
|
||||
for key in list(active):
|
||||
if key not in keys:
|
||||
row = active.pop(key)
|
||||
windows.append({
|
||||
"transit_planet": key[0],
|
||||
"natal_point": key[1],
|
||||
"aspect": key[2],
|
||||
"start_date": row["start_date"],
|
||||
"end_date": (start + timedelta(days=offset - 1)).date().isoformat(),
|
||||
"min_orb": round(row["min_orb"], 6),
|
||||
})
|
||||
daily_hits.append({"date": current, "hit_count": len(transit["aspects"]), "aspects": transit["aspects"]})
|
||||
final_date = end.date().isoformat()
|
||||
for key, row in active.items():
|
||||
windows.append({
|
||||
"transit_planet": key[0],
|
||||
"natal_point": key[1],
|
||||
"aspect": key[2],
|
||||
"start_date": row["start_date"],
|
||||
"end_date": final_date,
|
||||
"min_orb": round(row["min_orb"], 6),
|
||||
})
|
||||
return {
|
||||
"technique": "transit_duration_scan",
|
||||
"status": "used",
|
||||
"method": "daily local-midnight transit snapshots grouped into consecutive aspect windows",
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
"days_scanned": days,
|
||||
"daily_hits": daily_hits,
|
||||
"windows": sorted(windows, key=lambda row: (row["start_date"], row["min_orb"], row["transit_planet"])),
|
||||
"boundary": "Daily scan only; exact ingress/egress times require sub-daily root finding.",
|
||||
}
|
||||
|
||||
|
||||
def calculate_parans_status(*, target_date: str | None = None, **birth: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"technique": "parans",
|
||||
"status": "blocked",
|
||||
"target_date": target_date,
|
||||
"reason": "Parans need a dedicated rising/setting/culminating engine and latitude-aware event solver; not yet implemented in this repository.",
|
||||
}
|
||||
|
||||
|
||||
def build_timing_techniques(
|
||||
*,
|
||||
transit_date: str | None = None,
|
||||
solar_return_year: int | None = None,
|
||||
secondary_progression_date: str | None = None,
|
||||
solar_arc_date: str | None = None,
|
||||
converse_secondary_progression_date: str | None = None,
|
||||
converse_solar_arc_date: str | None = None,
|
||||
midpoint_date: str | None = None,
|
||||
lunar_return_start_date: str | None = None,
|
||||
duration_scan_start_date: str | None = None,
|
||||
duration_scan_end_date: str | None = None,
|
||||
parans_date: str | None = None,
|
||||
**birth: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""Materialize only the requested, independently auditable timing layers."""
|
||||
@@ -207,4 +426,24 @@ def build_timing_techniques(
|
||||
)
|
||||
if solar_arc_date:
|
||||
techniques["solar_arc_directions"] = calculate_solar_arc_directions(target_date=solar_arc_date, **birth)
|
||||
if converse_secondary_progression_date:
|
||||
techniques["converse_secondary_progressions"] = calculate_converse_secondary_progressions(
|
||||
target_date=converse_secondary_progression_date, **birth
|
||||
)
|
||||
if converse_solar_arc_date:
|
||||
techniques["converse_solar_arc_directions"] = calculate_converse_solar_arc_directions(
|
||||
target_date=converse_solar_arc_date, **birth
|
||||
)
|
||||
if midpoint_date:
|
||||
techniques["midpoints"] = calculate_midpoints(target_date=midpoint_date, **birth)
|
||||
if lunar_return_start_date:
|
||||
techniques["lunar_return"] = calculate_lunar_return(start_date=lunar_return_start_date, **birth)
|
||||
if duration_scan_start_date and duration_scan_end_date:
|
||||
techniques["transit_duration_scan"] = calculate_transit_duration_scan(
|
||||
start_date=duration_scan_start_date,
|
||||
end_date=duration_scan_end_date,
|
||||
**birth,
|
||||
)
|
||||
if parans_date:
|
||||
techniques["parans"] = calculate_parans_status(target_date=parans_date, **birth)
|
||||
return techniques
|
||||
|
||||
Reference in New Issue
Block a user