import os import uuid from datetime import datetime, timedelta from typing import Optional, Dict, Any, List import pytz import swisseph as swe from fastapi import APIRouter, HTTPException from pydantic import BaseModel from kerykeion import AstrologicalSubject, KerykeionChartSVG from chart_utils import extract_solar_return_wheel router = APIRouter(prefix="/calculators", tags=["Saturn Return"]) swe.set_ephe_path("/usr/share/ephe") swe.set_topo(0, 0) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CHARTS_DIR = os.path.join(BASE_DIR, "charts") SATURN_RETURN_DIR = os.path.join(CHARTS_DIR, "saturn_return") EPHE_PATH = os.path.join(BASE_DIR, "ephe") os.makedirs(SATURN_RETURN_DIR, exist_ok=True) swe.set_ephe_path(EPHE_PATH) SATURN_RETURN_RESULTS: Dict[str, Dict[str, Any]] = {} SIGNS = [ "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces" ] PLANET_SYMBOLS = { "Sun": "☉", "Moon": "☽", "Mercury": "☿", "Venus": "♀", "Mars": "♂", "Jupiter": "♃", "Saturn": "♄", "Uranus": "♅", "Neptune": "♆", "Pluto": "♇", "North Node": "☊", "South Node": "☋", "Black Moon Lilith": "⚸", "Ceres": "⚳", "Pallas": "⚴", "Juno": "⚵", "Vesta": "⚶", "Chiron": "⚷", "Pholus": "φ", "White Moon Selena": "W", "Ascendant": "AS", "Descendant": "DS", "Midheaven": "MC", "Imum Coeli": "IC", "Vertex": "Vx", "Part of Fortune": "X", } BODY_IDS = { "Sun": swe.SUN, "Moon": swe.MOON, "Mercury": swe.MERCURY, "Venus": swe.VENUS, "Mars": swe.MARS, "Jupiter": swe.JUPITER, "Saturn": swe.SATURN, "Uranus": swe.URANUS, "Neptune": swe.NEPTUNE, "Pluto": swe.PLUTO, "North Node": swe.TRUE_NODE, "South Node": 999, "Black Moon Lilith": swe.MEAN_APOG, "White Moon Selena": 999, "Ceres": swe.CERES, "Pallas": swe.PALLAS, "Juno": swe.JUNO, "Vesta": swe.VESTA, "Chiron": 2060, "Pholus": 5145, } ASPECT_SYMBOLS = { "conjunction": "☌", "opposition": "☍", "trine": "△", "square": "□", "sextile": "⚹", "semisextile": "⚺", "semisquare": "∠", "quintile": "⬠", "sesquiquadrate": "⚼", "biquintile": "⫲", "quincunx": "⚻", } MAJOR_ASPECTS = { "conjunction": {"angle": 0, "orb": 8}, "sextile": {"angle": 60, "orb": 6}, "square": {"angle": 90, "orb": 8}, "trine": {"angle": 120, "orb": 8}, "opposition": {"angle": 180, "orb": 8}, } MINOR_ASPECTS = { "semisextile": {"angle": 30, "orb": 2}, "semisquare": {"angle": 45, "orb": 2}, "quintile": {"angle": 72, "orb": 2}, "sesquiquadrate": {"angle": 135, "orb": 2}, "biquintile": {"angle": 144, "orb": 2}, "quincunx": {"angle": 150, "orb": 3}, } class SaturnReturnRequest(BaseModel): year: int month: int day: int hour: int minute: int second: int = 0 latitude: float longitude: float timezone: float = 5.5 birth_place: Optional[str] = None return_latitude: Optional[float] = None return_longitude: Optional[float] = None return_place: Optional[str] = None cycle: str = "first" house_system: str = "P" def get_timezone_str(timezone: float) -> str: if timezone == 5.5: return "Asia/Kolkata" elif timezone == 0: return "UTC" else: return f"Etc/GMT{-timezone:+d}" def generate_saturn_return_chart( birth_year: int, birth_month: int, birth_day: int, birth_hour: int, birth_minute: int, birth_lat: float, birth_lon: float, timezone: float, birth_place: Optional[str], return_dt: datetime, return_lat: float, return_lon: float, return_place: Optional[str], name: str = "User", ): try: timezone_str = get_timezone_str(timezone) city = birth_place or "Birth Place" nation = "IN" natal_subject = AstrologicalSubject( name=f"{name} Natal", year=birth_year, month=birth_month, day=birth_day, hour=birth_hour, minute=birth_minute, city=city, nation=nation, lat=birth_lat, lng=birth_lon, tz_str=timezone_str, ) return_city = return_place or return_place or "Return Place" return_subject = AstrologicalSubject( name=f"{name} Saturn Return", year=return_dt.year, month=return_dt.month, day=return_dt.day, hour=return_dt.hour, minute=return_dt.minute, city=return_city, nation=nation, lat=return_lat, lng=return_lon, tz_str=timezone_str, ) safe_name = "".join(c for c in name if c.isalnum()) or "chart" chart_name = f"{safe_name}_Saturn_Return_{datetime.now().strftime('%Y%m%d%H%M%S')}" before_files = set(os.listdir(SATURN_RETURN_DIR)) chart = KerykeionChartSVG( natal_subject, chart_type="Transit", second_obj=return_subject, new_output_directory=SATURN_RETURN_DIR, ) chart.makeSVG() after_files = set(os.listdir(SATURN_RETURN_DIR)) new_files = list(after_files - before_files) svg_files = [f for f in new_files if f.lower().endswith(".svg")] if not svg_files: svg_files = [ f for f in os.listdir(SATURN_RETURN_DIR) if f.lower().endswith(".svg") ] if not svg_files: return None svg_files.sort( key=lambda f: os.path.getmtime(os.path.join(SATURN_RETURN_DIR, f)), reverse=True, ) latest_svg = svg_files[0] latest_path = os.path.join(SATURN_RETURN_DIR, latest_svg) final_name = f"{chart_name}.svg" final_path = os.path.join(SATURN_RETURN_DIR, final_name) extract_solar_return_wheel(latest_path, final_path) return f"/charts/saturn_return/{final_name}" except Exception as e: print("Saturn Return chart generation error:", e) return None def normalize_deg(value: float) -> float: return value % 360 def angle_diff(a: float, b: float) -> float: diff = abs(a - b) % 360 return min(diff, 360 - diff) def local_to_utc(year, month, day, hour, minute, second, timezone): local_dt = datetime(year, month, day, hour, minute, second) tz_hours = int(timezone) tz_minutes = int(round((timezone - tz_hours) * 60)) return local_dt - timedelta(hours=tz_hours, minutes=tz_minutes) def jd_from_datetime(dt: datetime): return swe.julday( dt.year, dt.month, dt.day, dt.hour + dt.minute / 60 + dt.second / 3600, ) def get_sign(longitude: float) -> str: return SIGNS[int(normalize_deg(longitude) // 30)] def format_degree(longitude: float) -> str: longitude = normalize_deg(longitude) deg_in_sign = longitude % 30 deg = int(deg_in_sign) minute = (deg_in_sign - deg) * 60 minute_rounded = int(round(minute)) if minute_rounded == 60: deg += 1 minute_rounded = 0 return f"{deg}° {minute_rounded}'" def format_orb(orb_float: float) -> str: deg = int(orb_float) minutes = int(round((orb_float - deg) * 60)) if minutes == 60: deg += 1 minutes = 0 return f"{deg}° {minutes}'" def decimal_to_dms(value: float, lat=True) -> str: if lat: direction = "N" if value >= 0 else "S" else: direction = "E" if value >= 0 else "W" v = abs(value) deg = int(v) minutes = int(round((v - deg) * 60)) if minutes == 60: deg += 1 minutes = 0 return f"{deg}° {minutes}' {direction}" def calc_chiron_pholus(jd, body_id): try: if body_id in [2060, 5145]: result = swe.calc_ut(jd, body_id) return {"longitude": normalize_deg(result[0][0]), "speed": 0, "retrograde": False} return calc_body(jd, body_id) except Exception: return calc_body(jd, body_id) def calc_body(jd: float, body_id: int): try: if body_id in [2060, 5145]: data = swe.calc_ut(jd, body_id)[0] return { "longitude": normalize_deg(data[0]), "speed": data[3], "retrograde": data[3] < 0, } except Exception: pass data = swe.calc_ut(jd, body_id, swe.FLG_MOSEPH | swe.FLG_SPEED)[0] return { "longitude": normalize_deg(data[0]), "speed": data[3], "retrograde": data[3] < 0, } def get_chart_points(jd: float, lat: float, lon: float, house_system="P"): cusps, ascmc = swe.houses(jd, lat, lon, house_system.encode()) asc = normalize_deg(ascmc[0]) mc = normalize_deg(ascmc[1]) desc = normalize_deg(asc + 180) ic = normalize_deg(mc + 180) return { "cusps": list(cusps), "Ascendant": asc, "Descendant": desc, "Midheaven": mc, "Imum Coeli": ic, } def house_for_longitude(longitude: float, cusps: List[float]) -> int: lon = normalize_deg(longitude) for i in range(12): start = normalize_deg(cusps[i]) end = normalize_deg(cusps[(i + 1) % 12]) if start <= end: if start <= lon < end: return i + 1 else: if lon >= start or lon < end: return i + 1 return 12 def format_planet(name: str, longitude: float, speed: float, house: int): return { "name": name, "symbol": PLANET_SYMBOLS.get(name, ""), "longitude": round(normalize_deg(longitude), 4), "sign": get_sign(longitude), "degree": format_degree(longitude), "house": house, "retrograde": speed < 0, } def part_of_fortune(sun_lon, moon_lon, asc_lon): return normalize_deg(asc_lon + moon_lon - sun_lon) def vertex_approx(asc_lon, mc_lon): return normalize_deg(mc_lon + 90) def calculate_planets(jd: float, lat: float, lon: float, house_system="P"): points = get_chart_points(jd, lat, lon, house_system) cusps = points["cusps"] planets = [] calculated = {} for name, body_id in BODY_IDS.items(): if name in ["Chiron", "Pholus"]: continue try: body = calc_body(jd, body_id) calculated[name] = body house = house_for_longitude(body["longitude"], cusps) planets.append(format_planet(name, body["longitude"], body["speed"], house)) except Exception: continue for name, body_id in [("Chiron", 2060), ("Pholus", 5145)]: try: result = swe.calc_ut(jd, body_id) chiron_lon = normalize_deg(result[0][0]) calculated[name] = {"longitude": chiron_lon, "speed": 0} planets.append(format_planet(name, chiron_lon, 0, house_for_longitude(chiron_lon, cusps))) except Exception: pass if "North Node" in calculated: south_lon = normalize_deg(calculated["North Node"]["longitude"] + 180) planets.append(format_planet("South Node", south_lon, calculated["North Node"]["speed"], house_for_longitude(south_lon, cusps))) if "Black Moon Lilith" in calculated: selena_lon = normalize_deg(calculated["Black Moon Lilith"]["longitude"] + 180) planets.append(format_planet("White Moon Selena", selena_lon, 0, house_for_longitude(selena_lon, cusps))) for point_name in ["Ascendant", "Descendant", "Midheaven", "Imum Coeli"]: point_lon = points[point_name] planets.append(format_planet(point_name, point_lon, 0, house_for_longitude(point_lon, cusps))) if "Sun" in calculated and "Moon" in calculated: pof_lon = part_of_fortune(calculated["Sun"]["longitude"], calculated["Moon"]["longitude"], points["Ascendant"]) planets.append(format_planet("Part of Fortune", pof_lon, 0, house_for_longitude(pof_lon, cusps))) vx_lon = vertex_approx(points["Ascendant"], points["Midheaven"]) planets.append(format_planet("Vertex", vx_lon, 0, house_for_longitude(vx_lon, cusps))) return planets def get_cycle_range(cycle: str): if cycle == "second": return 56, 60, "Second Saturn Return" if cycle == "third": return 84, 90, "Third Saturn Return" return 27, 31, "First Saturn Return" def find_saturn_return_passes(birth_utc: datetime, natal_saturn_lon: float, cycle: str): start_age, end_age, cycle_label = get_cycle_range(cycle) search_start = birth_utc + timedelta(days=int(start_age * 365.25)) search_end = birth_utc + timedelta(days=int(end_age * 365.25)) candidates = [] current = search_start while current <= search_end: jd = jd_from_datetime(current) saturn = calc_body(jd, swe.SATURN) orb = angle_diff(saturn["longitude"], natal_saturn_lon) if orb <= 1.0: candidates.append( { "datetime": current, "orb": orb, } ) current += timedelta(days=1) grouped = [] for candidate in candidates: if not grouped: grouped.append(candidate) continue last = grouped[-1] if (candidate["datetime"] - last["datetime"]).days > 25: grouped.append(candidate) else: if candidate["orb"] < last["orb"]: grouped[-1] = candidate exact_passes = [] for item in grouped: center = item["datetime"] refine_start = center - timedelta(days=3) refine_end = center + timedelta(days=3) best_dt = center best_orb = 999 t = refine_start while t <= refine_end: jd = jd_from_datetime(t) saturn = calc_body(jd, swe.SATURN) orb = angle_diff(saturn["longitude"], natal_saturn_lon) if orb < best_orb: best_orb = orb best_dt = t t += timedelta(minutes=30) jd = jd_from_datetime(best_dt) saturn = calc_body(jd, swe.SATURN) exact_passes.append( { "utc_datetime": best_dt, "saturn_longitude": saturn["longitude"], "orb": best_orb, "motion": "Retrograde" if saturn["speed"] < 0 else "Direct", "age": round((best_dt - birth_utc).days / 365.25, 1), } ) return exact_passes, cycle_label, start_age, end_age def detect_overlay_aspects(natal_planets, return_planets): planet_order = [ "Sun", "Moon", "Mercury", "Venus", "Mars", "Jupiter", "Saturn", "Uranus", "Neptune", "Pluto", "North Node", "South Node", "Black Moon Lilith", "White Moon Selena", "Ceres", "Pallas", "Juno", "Vesta", "Chiron", "Pholus", "Ascendant", "Descendant", "Midheaven", "Imum Coeli", "Vertex", "Part of Fortune" ] aspects = [] natal_lookup = {p["name"]: p for p in natal_planets} return_lookup = {p["name"]: p for p in return_planets} for np_name in planet_order: if np_name not in natal_lookup: continue np_data = natal_lookup[np_name] np_order = planet_order.index(np_name) for rp_name in planet_order: if rp_name not in return_lookup: continue rp_data = return_lookup[rp_name] diff = angle_diff(rp_data["longitude"], np_data["longitude"]) for aspect_type in ["Major", "Minor"]: cfg_map = MAJOR_ASPECTS if aspect_type == "Major" else MINOR_ASPECTS for aspect_name, cfg in cfg_map.items(): orb = abs(diff - cfg["angle"]) if orb <= cfg["orb"]: aspects.append({ "return_planet": rp_name, "return_symbol": rp_data["symbol"], "aspect": aspect_name, "aspect_symbol": ASPECT_SYMBOLS[aspect_name], "aspect_type": aspect_type, "natal_planet": np_name, "natal_symbol": np_data["symbol"], "orb": format_orb(orb), "orb_value": round(orb, 4), "sort_order": np_order, }) aspects.sort(key=lambda x: (x["sort_order"], x["orb_value"])) return aspects def calculate_saturn_return_result(payload: SaturnReturnRequest): timezone = payload.timezone birth_utc = local_to_utc( payload.year, payload.month, payload.day, payload.hour, payload.minute, payload.second, timezone, ) birth_jd = jd_from_datetime(birth_utc) birth_lat = payload.latitude birth_lon = payload.longitude return_lat = payload.return_latitude if payload.return_latitude is not None else birth_lat return_lon = payload.return_longitude if payload.return_longitude is not None else birth_lon house_system = payload.house_system natal_saturn = calc_body(birth_jd, swe.SATURN) natal_saturn_lon = natal_saturn["longitude"] passes, cycle_label, start_age, end_age = find_saturn_return_passes( birth_utc, natal_saturn_lon, payload.cycle, ) if not passes: raise HTTPException(status_code=404, detail="No Saturn return pass found.") selected_pass = passes[0] return_utc = selected_pass["utc_datetime"] return_jd = jd_from_datetime(return_utc) local_birth_dt = datetime( payload.year, payload.month, payload.day, payload.hour, payload.minute, payload.second, ) local_return_dt = return_utc + timedelta(hours=timezone) natal_planets = calculate_planets( birth_jd, birth_lat, birth_lon, house_system, ) return_planets = calculate_planets( return_jd, return_lat, return_lon, house_system, ) overlay_aspects = detect_overlay_aspects(natal_planets, return_planets) major_overlay_aspects = [ a for a in overlay_aspects if a["aspect_type"] == "Major" ] minor_overlay_aspects = [ a for a in overlay_aspects if a["aspect_type"] == "Minor" ] exact_passes = [] for index, p in enumerate(passes): pass_local_dt = p["utc_datetime"] + timedelta(hours=timezone) exact_passes.append( { "pass_number": index + 1, "date": pass_local_dt.strftime("%B %d, %Y").replace(" 0", " "), "time": pass_local_dt.strftime("%H:%M"), "timezone": "Asia/Kolkata" if timezone == 5.5 else f"UTC{timezone:+}", "motion": p["motion"], "age": p["age"], "used": index == 0, "saturn_longitude": round(p["saturn_longitude"], 4), "sign": get_sign(p["saturn_longitude"]), "degree": format_degree(p["saturn_longitude"]), "orb": format_orb(p["orb"]), } ) chart_url = generate_saturn_return_chart( birth_year=payload.year, birth_month=payload.month, birth_day=payload.day, birth_hour=payload.hour, birth_minute=payload.minute, birth_lat=birth_lat, birth_lon=birth_lon, timezone=timezone, birth_place=payload.birth_place, return_dt=local_return_dt, return_lat=return_lat, return_lon=return_lon, return_place=payload.return_place, ) return { "success": True, "details": { "birthdate": local_birth_dt.strftime("%B %d, %Y at %H:%M").replace(" 0", " "), "birth_place": payload.birth_place or "Selected location", "birth_coordinates": f"{decimal_to_dms(birth_lat, True)}, {decimal_to_dms(birth_lon, False)}", "house_system": "Placidus" if house_system == "P" else house_system, "cycle_label": cycle_label, "age_range": f"Age {start_age}–{end_age}", "return_date": local_return_dt.strftime("%B %d, %Y").replace(" 0", " "), "return_time": local_return_dt.strftime("%H:%M"), "timezone": "Asia/Kolkata" if timezone == 5.5 else f"UTC{timezone:+}", "return_cycle": cycle_label, "exact_pass": f"1 of {len(passes)}", "pass_phase": selected_pass["motion"] + " pass", "age_at_exact_return": selected_pass["age"], "return_place": payload.return_place or payload.birth_place or "Selected location", "return_coordinates": f"{decimal_to_dms(return_lat, True)}, {decimal_to_dms(return_lon, False)}", }, "natal_saturn": { "longitude": round(natal_saturn_lon, 4), "sign": get_sign(natal_saturn_lon), "degree": format_degree(natal_saturn_lon), "retrograde": natal_saturn["retrograde"], }, "exact_passes": exact_passes, "natal_planets": natal_planets, "return_planets": return_planets, "overlay_aspects": overlay_aspects, "major_overlay_aspects": major_overlay_aspects, "minor_overlay_aspects": minor_overlay_aspects, "chart_url": chart_url, } @router.post("/saturn-return-chart") def create_saturn_return_chart(payload: SaturnReturnRequest): result = calculate_saturn_return_result(payload) result_id = str(uuid.uuid4().int)[:7] SATURN_RETURN_RESULTS[result_id] = result return { "success": True, "result_id": result_id, "redirect_url": f"/calculators/saturn-return-chart/{result_id}", "data": result, } @router.get("/saturn-return-chart/{result_id}") def read_saturn_return_chart(result_id: str): result = SATURN_RETURN_RESULTS.get(result_id) if not result: raise HTTPException(status_code=404, detail="Saturn return result not found.") return result