Texte verbessert

public/py/api.py

@@ -278,6 +278,24 @@ def action_comet_brightnesses(args: list[str]) -> dict:
         fail(str(exc), extra={"action": "comet_brightnesses"})
 
 
+def parse_comet_payload_arg(payload_arg: str, action_name: str) -> list[dict]:
+    try:
+        if payload_arg.startswith("@"):
+            payload_path = payload_arg[1:]
+            with open(payload_path, "r", encoding="utf-8") as handle:
+                payload_text = handle.read()
+        else:
+            payload_text = base64.urlsafe_b64decode(payload_arg.encode("ascii")).decode("utf-8")
+        payload = json.loads(payload_text)
+    except Exception as exc:
+        fail(f"Payload fuer {action_name} ist ungueltig.", extra={"details": str(exc)})
+
+    if not isinstance(payload, list):
+        fail(f"Payload fuer {action_name} muss eine JSON-Liste sein.")
+
+    return [item for item in payload if isinstance(item, dict)]
+
+
 def search_event(
     body: astronomy.Body,
     direction: astronomy.Direction,
@@ -824,6 +842,31 @@ def body_horizontal_coords(
     return float(hor.altitude), float(hor.azimuth)
 
 
+def comet_horizontal_coords(
+    comet: dict,
+    observer: astronomy.Observer,
+    dt_utc: datetime,
+) -> dict | None:
+    brightness = comets.calculate_brightness(comet, dt_utc)
+    ra_hours = brightness.get("ra_hours")
+    dec_deg = brightness.get("dec_deg")
+    if ra_hours is None or dec_deg is None:
+        return None
+
+    time_value = dt_to_time(dt_utc)
+    hor = astronomy.Horizon(time_value, observer, float(ra_hours), float(dec_deg), astronomy.Refraction.Normal)
+
+    return {
+        "altitude_deg": float(hor.altitude),
+        "azimuth_deg": float(hor.azimuth),
+        "estimated_magnitude": brightness.get("estimated_magnitude"),
+        "heliocentric_distance_au": brightness.get("heliocentric_distance_au"),
+        "geocentric_distance_au": brightness.get("geocentric_distance_au"),
+        "ra_hours": float(ra_hours),
+        "dec_deg": float(dec_deg),
+    }
+
+
 def body_constellation_info(
     body: astronomy.Body,
     dt_utc: datetime,
@@ -864,6 +907,301 @@ def serialize_body_position(
     return payload
 
 
+def action_favorite_comet_events_for_month(args: list[str]) -> dict:
+    if len(args) != 12:
+        fail(
+            "Aktion favorite_comet_events_for_month erwartet 12 Argumente: latitude longitude elevation year month timezone sample_offset_minutes min_comet_alt_deg max_sun_alt_deg max_comet_magnitude min_consecutive_days payloadBase64Oder@Datei",
+            extra={"argv": args},
+        )
+
+    latitude = parse_float(args[0], "Latitude")
+    longitude = parse_float(args[1], "Longitude")
+    elevation = parse_float(args[2], "Elevation")
+
+    try:
+        year = int(args[3])
+        month = int(args[4])
+        sample_offset_minutes = int(args[6])
+        min_consecutive_days = int(args[10])
+    except ValueError as exc:
+        fail("Jahr, Monat oder Kometenparameter sind ungueltig.", extra={"details": str(exc), "argv": args})
+
+    if month < 1 or month > 12:
+        fail("Monat muss zwischen 1 und 12 liegen.", extra={"month": month})
+
+    timezone_name = args[5]
+    min_comet_alt_deg = parse_float(args[7], "Minimale Kometenhoehe")
+    max_sun_alt_deg = parse_float(args[8], "Maximale Sonnenhoehe")
+    max_comet_magnitude = parse_float(args[9], "Maximale Kometenhelligkeit")
+    comet_payload = parse_comet_payload_arg(args[11], "favorite_comet_events_for_month")
+
+    try:
+        tz = ZoneInfo(timezone_name)
+    except Exception as exc:
+        fail("Zeitzone ist ungueltig oder auf dem Server nicht verfuegbar.", extra={"details": str(exc)})
+
+    observer = astronomy.Observer(latitude, longitude, elevation)
+    local_start = datetime(year, month, 1, 0, 0, 0, tzinfo=tz)
+    if month == 12:
+        local_end = datetime(year + 1, 1, 1, 0, 0, 0, tzinfo=tz)
+    else:
+        local_end = datetime(year, month + 1, 1, 0, 0, 0, tzinfo=tz)
+
+    utc_start = local_start.astimezone(timezone.utc)
+    utc_end = local_end.astimezone(timezone.utc)
+    utc_start_padded = utc_start - timedelta(days=1)
+    utc_end_padded = utc_end + timedelta(days=1)
+
+    def comet_name(comet: dict) -> str:
+        designation = str(comet.get("designation_and_name") or "").strip()
+        if designation != "":
+            return designation
+        packed = str(comet.get("designation_packed") or "").strip()
+        if packed != "":
+            return packed
+        comet_id = comet.get("id")
+        return f"Komet {comet_id}" if comet_id is not None else "Komet"
+
+    def append_perihelion_event(events: list[dict], comet: dict, name: str) -> None:
+        perihelion_utc = comets.build_perihelion_datetime(comet)
+        if perihelion_utc is None:
+            return
+
+        local_dt = perihelion_utc.astimezone(tz)
+        if not (local_start <= local_dt < local_end):
+            return
+
+        events.append({
+            "kind": "perihelion",
+            "comet_id": comets.parse_int(comet.get("id")),
+            "comet_name": name,
+            "label": f"Perihel von {name}",
+            "utc_iso": perihelion_utc.isoformat().replace("+00:00", "Z"),
+            "local_iso": local_dt.isoformat(),
+            "local_date": local_dt.strftime("%d.%m.%Y"),
+            "local_time": local_dt.strftime("%H:%M"),
+        })
+
+    def append_geocentric_minimum_event(events: list[dict], comet: dict, name: str) -> None:
+        sample_step = timedelta(hours=2)
+        sample_times: list[datetime] = []
+        sample_distances: list[float] = []
+
+        current_utc = utc_start_padded
+        while current_utc <= utc_end_padded:
+            brightness = comets.calculate_brightness(comet, current_utc)
+            distance_au = brightness.get("geocentric_distance_au")
+            if isinstance(distance_au, (int, float)):
+                sample_times.append(current_utc)
+                sample_distances.append(float(distance_au))
+            current_utc += sample_step
+
+        candidate_index = None
+        candidate_distance = None
+        for index in range(1, len(sample_distances) - 1):
+            current_distance = sample_distances[index]
+            if current_distance > sample_distances[index - 1] or current_distance > sample_distances[index + 1]:
+                continue
+
+            candidate_utc = sample_times[index]
+            if not (utc_start <= candidate_utc < utc_end):
+                continue
+
+            if candidate_distance is None or current_distance < candidate_distance:
+                candidate_index = index
+                candidate_distance = current_distance
+
+        if candidate_index is None:
+            return
+
+        rough_utc = sample_times[candidate_index]
+        best_utc = rough_utc
+        best_distance = candidate_distance if candidate_distance is not None else sample_distances[candidate_index]
+        refine_start = rough_utc - sample_step
+        refine_end = rough_utc + sample_step
+        refine_step = timedelta(minutes=5)
+        current_utc = refine_start
+        while current_utc <= refine_end:
+            brightness = comets.calculate_brightness(comet, current_utc)
+            distance_au = brightness.get("geocentric_distance_au")
+            if isinstance(distance_au, (int, float)) and float(distance_au) < best_distance:
+                best_distance = float(distance_au)
+                best_utc = current_utc
+            current_utc += refine_step
+
+        if not (utc_start <= best_utc < utc_end):
+            return
+
+        local_dt = best_utc.astimezone(tz)
+        events.append({
+            "kind": "geocentric_minimum",
+            "comet_id": comets.parse_int(comet.get("id")),
+            "comet_name": name,
+            "label": f"{name} in Erdnähe ({best_distance:.3f} AE)",
+            "distance_au": best_distance,
+            "utc_iso": best_utc.isoformat().replace("+00:00", "Z"),
+            "local_iso": local_dt.isoformat(),
+            "local_date": local_dt.strftime("%d.%m.%Y"),
+            "local_time": local_dt.strftime("%H:%M"),
+        })
+
+    def append_visibility_events(events: list[dict], comet: dict, name: str) -> None:
+        daily_checks: list[dict] = []
+        current_local = local_start
+
+        while current_local < local_end:
+            day_start_utc = current_local.astimezone(timezone.utc)
+            next_day_utc = (current_local + timedelta(days=1)).astimezone(timezone.utc)
+            start_time = dt_to_time(day_start_utc)
+
+            sunrise = search_event(astronomy.Body.Sun, astronomy.Direction.Rise, observer, start_time, next_day_utc)
+            sunset = search_event(astronomy.Body.Sun, astronomy.Direction.Set, observer, start_time, next_day_utc)
+
+            def build_period(sample_local: datetime | None) -> dict:
+                if sample_local is None:
+                    return {
+                        "ok": False,
+                        "sample_local": None,
+                        "comet_alt_deg": None,
+                        "sun_alt_deg": None,
+                        "estimated_magnitude": None,
+                    }
+
+                sample_utc = sample_local.astimezone(timezone.utc)
+                comet_coords = comet_horizontal_coords(comet, observer, sample_utc)
+                if comet_coords is None:
+                    return {
+                        "ok": False,
+                        "sample_local": sample_local,
+                        "comet_alt_deg": None,
+                        "sun_alt_deg": None,
+                        "estimated_magnitude": None,
+                    }
+
+                comet_alt_deg = float(comet_coords["altitude_deg"])
+                sun_alt_deg = body_altitude_deg(astronomy.Body.Sun, observer, sample_utc)
+                estimated_magnitude = comet_coords.get("estimated_magnitude")
+                is_ok = (
+                    comet_alt_deg >= min_comet_alt_deg
+                    and sun_alt_deg <= max_sun_alt_deg
+                    and isinstance(estimated_magnitude, (int, float))
+                    and float(estimated_magnitude) <= max_comet_magnitude
+                )
+
+                return {
+                    "ok": is_ok,
+                    "sample_local": sample_local,
+                    "comet_alt_deg": comet_alt_deg,
+                    "sun_alt_deg": sun_alt_deg,
+                    "estimated_magnitude": float(estimated_magnitude) if isinstance(estimated_magnitude, (int, float)) else None,
+                }
+
+            morning_local = None
+            if sunrise is not None:
+                sunrise_local = time_to_datetime(sunrise).astimezone(tz)
+                morning_local = sunrise_local - timedelta(minutes=sample_offset_minutes)
+
+            evening_local = None
+            if sunset is not None:
+                sunset_local = time_to_datetime(sunset).astimezone(tz)
+                evening_local = sunset_local + timedelta(minutes=sample_offset_minutes)
+
+            daily_checks.append({
+                "date": current_local.strftime("%d.%m.%Y"),
+                "morning": build_period(morning_local),
+                "evening": build_period(evening_local),
+            })
+
+            current_local += timedelta(days=1)
+
+        def build_windows(period_key: str, label_text: str) -> list[dict]:
+            comet_events = []
+            start_index = None
+
+            for index, day in enumerate(daily_checks):
+                period = day[period_key]
+                is_ok = bool(period["ok"])
+                if is_ok and start_index is None:
+                    start_index = index
+
+                is_last = index == len(daily_checks) - 1
+                if start_index is not None and (not is_ok or is_last):
+                    end_index = index if (is_ok and is_last) else index - 1
+                    duration_days = end_index - start_index + 1
+                    if duration_days >= min_consecutive_days:
+                        window_days = daily_checks[start_index:end_index + 1]
+                        start_day = window_days[0]
+                        end_day = window_days[-1]
+                        start_local = start_day[period_key]["sample_local"]
+                        if start_local is not None:
+                            range_text = (
+                                start_day["date"]
+                                if start_index == end_index
+                                else f'{start_day["date"]} bis {end_day["date"]}'
+                            )
+                            best_magnitude = None
+                            for window_day in window_days:
+                                magnitude = window_day[period_key]["estimated_magnitude"]
+                                if magnitude is None:
+                                    continue
+                                best_magnitude = magnitude if best_magnitude is None else min(best_magnitude, magnitude)
+
+                            label = f"Gute Sichtbarkeit von {name} am {label_text} ({range_text})"
+                            if best_magnitude is not None:
+                                label += f" bis ca. {best_magnitude:.1f} mag"
+
+                            end_local = end_day[period_key]["sample_local"] or start_local
+                            comet_events.append({
+                                "kind": "good_visibility",
+                                "period": period_key,
+                                "comet_id": comets.parse_int(comet.get("id")),
+                                "comet_name": name,
+                                "label": label,
+                                "best_magnitude": best_magnitude,
+                                "utc_iso": start_local.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
+                                "local_iso": start_local.isoformat(),
+                                "local_date": start_local.strftime("%d.%m.%Y"),
+                                "local_time": start_local.strftime("%H:%M"),
+                                "end_local_iso": end_local.isoformat(),
+                                "end_local_date": end_day["date"],
+                                "end_local_time": end_local.strftime("%H:%M"),
+                                "duration_minutes": max(1, int(round((end_local - start_local).total_seconds() / 60.0))),
+                                "duration_days": duration_days,
+                            })
+                    start_index = None
+
+            return comet_events
+
+        events.extend(build_windows("morning", "Morgen"))
+        events.extend(build_windows("evening", "Abend"))
+
+    events: list[dict] = []
+    for comet in comet_payload:
+        name = comet_name(comet)
+        append_perihelion_event(events, comet, name)
+        append_geocentric_minimum_event(events, comet, name)
+        append_visibility_events(events, comet, name)
+
+    events.sort(key=lambda item: item["local_iso"])
+
+    return {
+        "ok": True,
+        "action": "favorite_comet_events_for_month",
+        "selected": {
+            "year": year,
+            "month": month,
+            "timezone": timezone_name,
+            "sample_offset_minutes": sample_offset_minutes,
+            "min_comet_alt_deg": min_comet_alt_deg,
+            "max_sun_alt_deg": max_sun_alt_deg,
+            "max_comet_magnitude": max_comet_magnitude,
+            "min_consecutive_days": min_consecutive_days,
+            "comet_count": len(comet_payload),
+        },
+        "events": events,
+    }
+
+
 def action_mercury_good_visibility_for_month(args: list[str]) -> dict:
     if len(args) != 10:
         fail(
@@ -4554,7 +4892,7 @@ def action_satellite_passes(args: list[str]) -> dict:
 
 def main() -> None:
     if len(sys.argv) < 2:
-        fail("Es wurde keine Aktion uebergeben.", extra={"available_actions": ["sun_moon_rise_set", "solar_longitude_to_datetime", "current_solar_longitude", "astronomical_conversions", "comet_brightnesses", "moon_star_occultations", "moon_phase_details", "satellite_passes", "planet_rise_set", "month_sky_context", "moon_phases_for_month", "golden_handle_for_month", "season_changes_for_month", "time_changes_for_month", "moon_planet_approaches", "moon_deep_sky_approaches_for_month", "planet_bright_star_approaches_for_month", "planet_conjunctions_for_month", "eclipses_for_month", "moon_apsides_for_month", "sun_apsides_for_month", "inner_planet_elongations_for_month", "venus_peak_magnitude_for_month", "outer_planet_events_for_month", "outer_planet_stations_for_month", "jupiter_moons_one_side_for_month", "mercury_good_visibility_for_month", "planet_parades_for_month", "planet_constellation_changes_for_month", "moon_star_occultations_for_month"]})
+        fail("Es wurde keine Aktion uebergeben.", extra={"available_actions": ["sun_moon_rise_set", "solar_longitude_to_datetime", "current_solar_longitude", "astronomical_conversions", "comet_brightnesses", "favorite_comet_events_for_month", "moon_star_occultations", "moon_phase_details", "satellite_passes", "planet_rise_set", "month_sky_context", "moon_phases_for_month", "golden_handle_for_month", "season_changes_for_month", "time_changes_for_month", "moon_planet_approaches", "moon_deep_sky_approaches_for_month", "planet_bright_star_approaches_for_month", "planet_conjunctions_for_month", "eclipses_for_month", "moon_apsides_for_month", "sun_apsides_for_month", "inner_planet_elongations_for_month", "venus_peak_magnitude_for_month", "outer_planet_events_for_month", "outer_planet_stations_for_month", "jupiter_moons_one_side_for_month", "mercury_good_visibility_for_month", "planet_parades_for_month", "planet_constellation_changes_for_month", "moon_star_occultations_for_month"]})
 
     action = sys.argv[1]
     args = sys.argv[2:]
@@ -4584,6 +4922,11 @@ def main() -> None:
         print(json.dumps(result, ensure_ascii=True))
         return
 
+    if action == "favorite_comet_events_for_month":
+        result = action_favorite_comet_events_for_month(args)
+        print(json.dumps(result, ensure_ascii=True))
+        return
+
     if action == "moon_star_occultations":
         result = action_moon_star_occultations(args)
         print(json.dumps(result, ensure_ascii=True))
@@ -4714,7 +5057,7 @@ def main() -> None:
         print(json.dumps(result, ensure_ascii=True))
         return
 
-    fail("Unbekannte Aktion.", extra={"action": action, "available_actions": ["sun_moon_rise_set", "solar_longitude_to_datetime", "current_solar_longitude", "astronomical_conversions", "comet_brightnesses", "moon_star_occultations", "moon_phase_details", "satellite_passes", "planet_rise_set", "planet_visibility_chart", "month_sky_context", "moon_phases_for_month", "golden_handle_for_month", "season_changes_for_month", "time_changes_for_month", "moon_planet_approaches", "moon_deep_sky_approaches_for_month", "planet_bright_star_approaches_for_month", "planet_conjunctions_for_month", "eclipses_for_month", "moon_apsides_for_month", "sun_apsides_for_month", "inner_planet_elongations_for_month", "venus_peak_magnitude_for_month", "outer_planet_events_for_month", "outer_planet_stations_for_month", "jupiter_moons_one_side_for_month", "mercury_good_visibility_for_month", "planet_parades_for_month", "planet_constellation_changes_for_month", "moon_star_occultations_for_month"]})
+    fail("Unbekannte Aktion.", extra={"action": action, "available_actions": ["sun_moon_rise_set", "solar_longitude_to_datetime", "current_solar_longitude", "astronomical_conversions", "comet_brightnesses", "favorite_comet_events_for_month", "moon_star_occultations", "moon_phase_details", "satellite_passes", "planet_rise_set", "planet_visibility_chart", "month_sky_context", "moon_phases_for_month", "golden_handle_for_month", "season_changes_for_month", "time_changes_for_month", "moon_planet_approaches", "moon_deep_sky_approaches_for_month", "planet_bright_star_approaches_for_month", "planet_conjunctions_for_month", "eclipses_for_month", "moon_apsides_for_month", "sun_apsides_for_month", "inner_planet_elongations_for_month", "venus_peak_magnitude_for_month", "outer_planet_events_for_month", "outer_planet_stations_for_month", "jupiter_moons_one_side_for_month", "mercury_good_visibility_for_month", "planet_parades_for_month", "planet_constellation_changes_for_month", "moon_star_occultations_for_month"]})
 
 
 if __name__ == "__main__":