#!/usr/bin/env python3 import json import math import os import sys from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) if SCRIPT_DIR not in sys.path: sys.path.insert(0, SCRIPT_DIR) import astronomy EPHEMERIS_BODIES = { "Sun": ("Sonne", astronomy.Body.Sun), "Moon": ("Mond", astronomy.Body.Moon), "Mercury": ("Merkur", astronomy.Body.Mercury), "Venus": ("Venus", astronomy.Body.Venus), "Mars": ("Mars", astronomy.Body.Mars), "Jupiter": ("Jupiter", astronomy.Body.Jupiter), "Saturn": ("Saturn", astronomy.Body.Saturn), "Uranus": ("Uranus", astronomy.Body.Uranus), "Neptune": ("Neptun", astronomy.Body.Neptune), } def fail(message: str, *, extra: dict | None = None, code: int = 1) -> None: payload = {"ok": False, "error": message} if extra: payload.update(extra) print(json.dumps(payload, ensure_ascii=True)) raise SystemExit(code) def parse_float(value: str, label: str) -> float: try: return float(value) except ValueError as exc: fail(f"{label} ist ungueltig.", extra={"details": str(exc), "value": value}) def dt_to_time(dt_utc: datetime) -> astronomy.Time: dt_utc = dt_utc.astimezone(timezone.utc) return astronomy.Time.Make( dt_utc.year, dt_utc.month, dt_utc.day, dt_utc.hour, dt_utc.minute, dt_utc.second + (dt_utc.microsecond / 1_000_000.0), ) def time_to_datetime(time_value: astronomy.Time) -> datetime: year, month, day, hour, minute, second = time_value.Calendar() second_int = int(second) microsecond = int(round((second - second_int) * 1_000_000)) if microsecond >= 1_000_000: second_int += 1 microsecond -= 1_000_000 return datetime(year, month, day, hour, minute, second_int, microsecond, tzinfo=timezone.utc) def serialize_event(label: str, event_time: astronomy.Time | None, tz: ZoneInfo) -> dict: if event_time is None: return {"label": label, "found": False} utc_dt = time_to_datetime(event_time) local_dt = utc_dt.astimezone(tz) return { "label": label, "found": True, "utc_iso": utc_dt.isoformat().replace("+00:00", "Z"), "local_iso": local_dt.isoformat(), "local_time": local_dt.strftime("%H:%M"), "local_date": local_dt.strftime("%Y-%m-%d"), } def search_event( body: astronomy.Body, direction: astronomy.Direction, observer: astronomy.Observer, start_time: astronomy.Time, end_utc: datetime, ) -> astronomy.Time | None: search_limit_days = (end_utc - time_to_datetime(start_time)).total_seconds() / 86400.0 if search_limit_days <= 0: return None result = astronomy.SearchRiseSet(body, observer, direction, start_time, search_limit_days) if result is None: return None result_dt = time_to_datetime(result) if result_dt >= end_utc: return None return result def format_ra_hours(ra_hours: float) -> str: total_seconds = int(round(float(ra_hours) * 3600.0)) total_seconds %= 24 * 3600 hours = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 seconds = total_seconds % 60 return f"{hours:02d}:{minutes:02d}:{seconds:02d}" def format_dec_deg(dec_deg: float) -> str: sign = "+" if dec_deg >= 0 else "-" total_seconds = int(round(abs(float(dec_deg)) * 3600.0)) degrees = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 seconds = total_seconds % 60 return f"{sign}{degrees:02d}:{minutes:02d}:{seconds:02d}" def days_in_month(year: int, month: int) -> int: if month == 12: next_month = datetime(year + 1, 1, 1) else: next_month = datetime(year, month + 1, 1) this_month = datetime(year, month, 1) return (next_month - this_month).days def add_calendar_unit(base: datetime, amount: int, unit: str) -> datetime: if unit == "minutes": return base + timedelta(minutes=amount) if unit == "hours": return base + timedelta(hours=amount) if unit == "days": return base + timedelta(days=amount) if unit == "weeks": return base + timedelta(weeks=amount) if unit == "months": month_index = (base.month - 1) + amount year = base.year + (month_index // 12) month = (month_index % 12) + 1 day = min(base.day, days_in_month(year, month)) return base.replace(year=year, month=month, day=day) if unit == "years": year = base.year + amount day = base.day if base.month == 2 and base.day == 29: day = min(day, days_in_month(year, base.month)) return base.replace(year=year, day=day) fail("Zeitraum-Einheit ist ungueltig.", extra={"unit": unit}) def action_planet_ephemeris(args: list[str]) -> dict: if len(args) != 8: fail( "Aktion planet_ephemeris erwartet 8 Argumente: latitude longitude elevation timezone body intervalMinutes rangeValue rangeUnit", extra={"argv": args}, ) latitude = parse_float(args[0], "Latitude") longitude = parse_float(args[1], "Longitude") elevation = parse_float(args[2], "Elevation") timezone_name = args[3] body_name = str(args[4]).strip() try: interval_minutes = int(args[5]) range_value = int(args[6]) except ValueError as exc: fail("Intervall oder Zeitraum ist ungueltig.", extra={"details": str(exc), "argv": args}) range_unit = str(args[7]).strip() if interval_minutes <= 0: fail("Intervall muss groesser als 0 sein.", extra={"interval_minutes": interval_minutes}) if range_value <= 0: fail("Zeitraum muss groesser als 0 sein.", extra={"range_value": range_value}) if body_name not in EPHEMERIS_BODIES: fail("Objekt ist ungueltig.", extra={"body": body_name}) if range_unit not in {"minutes", "hours", "days", "weeks", "months", "years"}: fail("Zeitraum-Einheit ist ungueltig.", extra={"range_unit": range_unit}) try: tz = ZoneInfo(timezone_name) except Exception as exc: fail("Zeitzone ist ungueltig.", extra={"details": str(exc)}) label, body = EPHEMERIS_BODIES[body_name] observer = astronomy.Observer(latitude, longitude, elevation) local_start = datetime.now(tz).replace(second=0, microsecond=0) local_end = add_calendar_unit(local_start, range_value, range_unit) rows = [] current_local = local_start step = timedelta(minutes=interval_minutes) rise_set_cache: dict[str, tuple[str | None, str | None]] = {} max_rows = 50000 while current_local <= local_end: if len(rows) >= max_rows: fail( "Die Anfrage erzeugt zu viele Tabellenzeilen. Bitte Zeitraum verkuerzen oder groesseres Intervall waehlen.", extra={ "max_rows": max_rows, "interval_minutes": interval_minutes, "range_value": range_value, "range_unit": range_unit, }, ) current_utc = current_local.astimezone(timezone.utc) time_value = dt_to_time(current_utc) eq = astronomy.Equator(body, time_value, observer, False, True) day_key = current_local.strftime("%Y-%m-%d") if day_key not in rise_set_cache: local_day_start = current_local.replace(hour=0, minute=0, second=0, microsecond=0) local_day_end = local_day_start + timedelta(days=1) day_start_time = dt_to_time(local_day_start.astimezone(timezone.utc)) day_end_utc = local_day_end.astimezone(timezone.utc) rise = search_event(body, astronomy.Direction.Rise, observer, day_start_time, day_end_utc) set_ = search_event(body, astronomy.Direction.Set, observer, day_start_time, day_end_utc) rise_label = serialize_event("Aufgang", rise, tz).get("local_time") if rise is not None else None set_label = serialize_event("Untergang", set_, tz).get("local_time") if set_ is not None else None rise_set_cache[day_key] = (rise_label, set_label) rise_label, set_label = rise_set_cache[day_key] rows.append({ "object_name": label, "date_local": current_local.strftime("%d.%m.%Y"), "time_local": current_local.strftime("%H:%M"), "ra": format_ra_hours(float(eq.ra)), "ra_decimal_hours": round(float(eq.ra), 8), "dec": format_dec_deg(float(eq.dec)), "dec_decimal_deg": round(float(eq.dec), 8), "rise": rise_label, "set": set_label, }) current_local += step return { "ok": True, "action": "planet_ephemeris", "observer": { "latitude": latitude, "longitude": longitude, "elevation": elevation, "timezone": timezone_name, }, "object": { "key": body_name, "label": label, }, "window": { "local_start": local_start.isoformat(), "local_end": local_end.isoformat(), "interval_minutes": interval_minutes, "range_value": range_value, "range_unit": range_unit, }, "rows": rows, } def main() -> None: if len(sys.argv) < 2: fail("Es wurde keine Aktion uebergeben.", extra={"available_actions": ["planet_ephemeris"]}) action = sys.argv[1] args = sys.argv[2:] if action == "planet_ephemeris": result = action_planet_ephemeris(args) print(json.dumps(result, ensure_ascii=True)) return fail("Unbekannte Aktion.", extra={"action": action, "available_actions": ["planet_ephemeris"]}) if __name__ == "__main__": main()