755 lines
29 KiB
Python
755 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
import base64
|
|
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
|
|
from skyfield import almanac as skyfield_almanac
|
|
from astronomical_conversions import skyfield_context, skyfield_wgs84
|
|
|
|
|
|
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),
|
|
}
|
|
|
|
# DE421 enthält die äußeren Planeten als Systemschwerpunkte, nicht als
|
|
# separate Planetenzentren. Für Ephemeriden ist das die vorgesehene
|
|
# Zielkörperdefinition dieser Datei.
|
|
SKYFIELD_EPHEMERIS_TARGETS = {
|
|
"Sun": "sun",
|
|
"Moon": "moon",
|
|
"Mercury": "mercury",
|
|
"Venus": "venus",
|
|
"Mars": "mars",
|
|
"Jupiter": 5,
|
|
"Saturn": 6,
|
|
"Uranus": 7,
|
|
"Neptune": 8,
|
|
}
|
|
|
|
J2000_OBLIQUITY_DEG = 23.439279444444445
|
|
GAUSSIAN_GRAVITATIONAL_CONSTANT = 0.01720209895
|
|
PARABOLIC_ECCENTRICITY_TOLERANCE = 1.0e-6
|
|
MAX_ROWS = 50000
|
|
CUSTOM_EVENT_STEP_MINUTES = 10
|
|
|
|
|
|
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 parse_payload_float(payload: dict, key: str) -> float:
|
|
value = payload.get(key)
|
|
if value is None:
|
|
raise ValueError(f"{key} fehlt.")
|
|
return float(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 skyfield_observer(latitude: float, longitude: float, elevation: float):
|
|
_timescale, planets = skyfield_context()
|
|
return planets["earth"] + skyfield_wgs84.latlon(latitude, longitude, elevation_m=elevation)
|
|
|
|
|
|
def skyfield_planet_state(body_name: str, observer, dt_utc: datetime, timescale=None, planets=None):
|
|
if timescale is None or planets is None:
|
|
timescale, planets = skyfield_context()
|
|
time_value = timescale.from_datetime(dt_utc.astimezone(timezone.utc))
|
|
apparent = observer.at(time_value).observe(planets[SKYFIELD_EPHEMERIS_TARGETS[body_name]]).apparent()
|
|
ra, dec, _distance = apparent.radec(epoch=timescale.J2000)
|
|
altitude, azimuth, _distance = apparent.altaz("standard")
|
|
return float(ra.hours), float(dec.degrees), float(azimuth.degrees), float(altitude.degrees)
|
|
|
|
|
|
def search_skyfield_rise_set(observer, body_name: str, local_day_start: datetime, timescale, planets) -> tuple[datetime | None, datetime | None]:
|
|
"""Sucht Auf- und Untergang gemeinsam über Skyfields diskrete Ereignissuche."""
|
|
target = planets[SKYFIELD_EPHEMERIS_TARGETS[body_name]]
|
|
event_function = skyfield_almanac.risings_and_settings(
|
|
planets,
|
|
target,
|
|
observer - planets["earth"],
|
|
)
|
|
local_day_end = local_day_start + timedelta(days=1)
|
|
start_time = timescale.from_datetime(local_day_start.astimezone(timezone.utc))
|
|
end_time = timescale.from_datetime(local_day_end.astimezone(timezone.utc))
|
|
event_times, event_states = skyfield_almanac.find_discrete(start_time, end_time, event_function)
|
|
|
|
rise = None
|
|
set_ = None
|
|
for event_time, state in zip(event_times.utc_datetime(), event_states):
|
|
event_dt = event_time.astimezone(timezone.utc)
|
|
if bool(state) and rise is None:
|
|
rise = event_dt
|
|
elif not bool(state) and set_ is None:
|
|
set_ = event_dt
|
|
return rise, set_
|
|
|
|
|
|
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 solve_elliptic_anomaly(mean_anomaly: float, eccentricity: float) -> float:
|
|
anomaly = mean_anomaly if eccentricity < 0.8 else (math.pi if mean_anomaly >= 0.0 else -math.pi)
|
|
for _ in range(30):
|
|
delta = (anomaly - eccentricity * math.sin(anomaly) - mean_anomaly) / (1.0 - eccentricity * math.cos(anomaly))
|
|
anomaly -= delta
|
|
if abs(delta) < 1.0e-12:
|
|
break
|
|
return anomaly
|
|
|
|
|
|
def solve_hyperbolic_anomaly(mean_anomaly: float, eccentricity: float) -> float:
|
|
anomaly = 0.0 if mean_anomaly == 0.0 else math.asinh(mean_anomaly / eccentricity)
|
|
for _ in range(40):
|
|
sinh_value = math.sinh(anomaly)
|
|
cosh_value = math.cosh(anomaly)
|
|
delta = (eccentricity * sinh_value - anomaly - mean_anomaly) / (eccentricity * cosh_value - 1.0)
|
|
anomaly -= delta
|
|
if abs(delta) < 1.0e-12:
|
|
break
|
|
return anomaly
|
|
|
|
|
|
def solve_parabolic_parameter(delta_days: float, perihelion_distance_au: float) -> float:
|
|
scale = GAUSSIAN_GRAVITATIONAL_CONSTANT * delta_days / math.sqrt(2.0 * perihelion_distance_au**3)
|
|
parameter = scale
|
|
for _ in range(40):
|
|
numerator = parameter + (parameter**3) / 3.0 - scale
|
|
denominator = 1.0 + parameter**2
|
|
delta = numerator / denominator
|
|
parameter -= delta
|
|
if abs(delta) < 1.0e-12:
|
|
break
|
|
return parameter
|
|
|
|
|
|
def true_anomaly_and_radius(delta_days: float, perihelion_distance_au: float, eccentricity: float) -> tuple[float, float]:
|
|
if perihelion_distance_au <= 0.0:
|
|
raise ValueError("Periheldistanz muss positiv sein.")
|
|
|
|
if eccentricity < 1.0 - PARABOLIC_ECCENTRICITY_TOLERANCE:
|
|
semi_major_axis = perihelion_distance_au / (1.0 - eccentricity)
|
|
mean_motion = GAUSSIAN_GRAVITATIONAL_CONSTANT / (semi_major_axis ** 1.5)
|
|
mean_anomaly = math.fmod(mean_motion * delta_days, 2.0 * math.pi)
|
|
eccentric_anomaly = solve_elliptic_anomaly(mean_anomaly, eccentricity)
|
|
radius = semi_major_axis * (1.0 - eccentricity * math.cos(eccentric_anomaly))
|
|
true_anomaly = 2.0 * math.atan2(
|
|
math.sqrt(1.0 + eccentricity) * math.sin(eccentric_anomaly / 2.0),
|
|
math.sqrt(1.0 - eccentricity) * math.cos(eccentric_anomaly / 2.0),
|
|
)
|
|
return true_anomaly, radius
|
|
|
|
if eccentricity > 1.0 + PARABOLIC_ECCENTRICITY_TOLERANCE:
|
|
semi_major_axis_abs = perihelion_distance_au / (eccentricity - 1.0)
|
|
mean_anomaly = GAUSSIAN_GRAVITATIONAL_CONSTANT * delta_days / (semi_major_axis_abs ** 1.5)
|
|
hyperbolic_anomaly = solve_hyperbolic_anomaly(mean_anomaly, eccentricity)
|
|
radius = semi_major_axis_abs * (eccentricity * math.cosh(hyperbolic_anomaly) - 1.0)
|
|
true_anomaly = 2.0 * math.atan2(
|
|
math.sqrt(eccentricity + 1.0) * math.sinh(hyperbolic_anomaly / 2.0),
|
|
math.sqrt(eccentricity - 1.0) * math.cosh(hyperbolic_anomaly / 2.0),
|
|
)
|
|
return true_anomaly, radius
|
|
|
|
parabolic_parameter = solve_parabolic_parameter(delta_days, perihelion_distance_au)
|
|
true_anomaly = 2.0 * math.atan(parabolic_parameter)
|
|
radius = perihelion_distance_au * (1.0 + parabolic_parameter**2)
|
|
return true_anomaly, radius
|
|
|
|
|
|
def ecliptic_to_equatorial(x_ecl: float, y_ecl: float, z_ecl: float) -> tuple[float, float, float]:
|
|
epsilon = math.radians(J2000_OBLIQUITY_DEG)
|
|
cos_epsilon = math.cos(epsilon)
|
|
sin_epsilon = math.sin(epsilon)
|
|
return (
|
|
x_ecl,
|
|
y_ecl * cos_epsilon - z_ecl * sin_epsilon,
|
|
y_ecl * sin_epsilon + z_ecl * cos_epsilon,
|
|
)
|
|
|
|
|
|
def minorplanet_heliocentric_vector(payload: dict, dt_utc: datetime) -> tuple[float, float, float]:
|
|
epoch_date_iso = str(payload.get("epochDateIso") or "").strip()
|
|
if epoch_date_iso == "":
|
|
raise ValueError("Epochendatum fehlt.")
|
|
|
|
epoch_dt = datetime.fromisoformat(f"{epoch_date_iso}T00:00:00+00:00").astimezone(timezone.utc)
|
|
days_since_epoch = (dt_utc - epoch_dt).total_seconds() / 86400.0
|
|
|
|
mean_anomaly_deg = parse_payload_float(payload, "meanAnomalyDeg")
|
|
mean_motion_deg_per_day = parse_payload_float(payload, "meanMotionDegPerDay")
|
|
eccentricity = max(0.0, min(0.999999, parse_payload_float(payload, "eccentricity")))
|
|
semimajor_axis_au = parse_payload_float(payload, "semimajorAxisAu")
|
|
inclination = math.radians(parse_payload_float(payload, "inclinationDeg"))
|
|
ascending_node = math.radians(parse_payload_float(payload, "ascendingNodeDeg"))
|
|
arg_perihelion = math.radians(parse_payload_float(payload, "argPerihelionDeg"))
|
|
|
|
mean_anomaly = math.radians(mean_anomaly_deg + (mean_motion_deg_per_day * days_since_epoch))
|
|
eccentric_anomaly = solve_elliptic_anomaly(math.fmod(mean_anomaly, 2.0 * math.pi), eccentricity)
|
|
true_anomaly = 2.0 * math.atan2(
|
|
math.sqrt(1.0 + eccentricity) * math.sin(eccentric_anomaly / 2.0),
|
|
math.sqrt(1.0 - eccentricity) * math.cos(eccentric_anomaly / 2.0),
|
|
)
|
|
radius_au = semimajor_axis_au * (1.0 - eccentricity * math.cos(eccentric_anomaly))
|
|
argument_of_latitude = true_anomaly + arg_perihelion
|
|
|
|
x_ecl = radius_au * (
|
|
(math.cos(ascending_node) * math.cos(argument_of_latitude))
|
|
- (math.sin(ascending_node) * math.sin(argument_of_latitude) * math.cos(inclination))
|
|
)
|
|
y_ecl = radius_au * (
|
|
(math.sin(ascending_node) * math.cos(argument_of_latitude))
|
|
+ (math.cos(ascending_node) * math.sin(argument_of_latitude) * math.cos(inclination))
|
|
)
|
|
z_ecl = radius_au * (math.sin(argument_of_latitude) * math.sin(inclination))
|
|
|
|
return ecliptic_to_equatorial(x_ecl, y_ecl, z_ecl)
|
|
|
|
|
|
def comet_heliocentric_vector(payload: dict, dt_utc: datetime) -> tuple[float, float, float]:
|
|
perihelion_iso = str(payload.get("perihelionIso") or "").strip()
|
|
if perihelion_iso == "":
|
|
raise ValueError("Perihelzeit fehlt.")
|
|
|
|
perihelion_dt = datetime.fromisoformat(perihelion_iso.replace("Z", "+00:00")).astimezone(timezone.utc)
|
|
delta_days = (dt_utc - perihelion_dt).total_seconds() / 86400.0
|
|
|
|
perihelion_distance_au = parse_payload_float(payload, "perihelionDistanceAu")
|
|
eccentricity = parse_payload_float(payload, "eccentricity")
|
|
inclination = math.radians(parse_payload_float(payload, "inclinationDeg"))
|
|
ascending_node = math.radians(parse_payload_float(payload, "ascendingNodeDeg"))
|
|
arg_perihelion = math.radians(parse_payload_float(payload, "argPerihelionDeg"))
|
|
|
|
true_anomaly, radius = true_anomaly_and_radius(delta_days, perihelion_distance_au, eccentricity)
|
|
argument_of_latitude = arg_perihelion + true_anomaly
|
|
|
|
x_ecl = radius * (math.cos(ascending_node) * math.cos(argument_of_latitude) - math.sin(ascending_node) * math.sin(argument_of_latitude) * math.cos(inclination))
|
|
y_ecl = radius * (math.sin(ascending_node) * math.cos(argument_of_latitude) + math.cos(ascending_node) * math.sin(argument_of_latitude) * math.cos(inclination))
|
|
z_ecl = radius * (math.sin(argument_of_latitude) * math.sin(inclination))
|
|
|
|
return ecliptic_to_equatorial(x_ecl, y_ecl, z_ecl)
|
|
|
|
|
|
def custom_geocentric_vector(body_type: str, payload: dict, time_value: astronomy.Time) -> astronomy.Vector:
|
|
dt_utc = time_to_datetime(time_value)
|
|
if body_type == "minorplanet":
|
|
helio = minorplanet_heliocentric_vector(payload, dt_utc)
|
|
elif body_type == "comet":
|
|
helio = comet_heliocentric_vector(payload, dt_utc)
|
|
else:
|
|
raise ValueError("Objekttyp ist ungueltig.")
|
|
|
|
earth_vector = astronomy.HelioVector(astronomy.Body.Earth, time_value)
|
|
return astronomy.Vector(
|
|
helio[0] - earth_vector.x,
|
|
helio[1] - earth_vector.y,
|
|
helio[2] - earth_vector.z,
|
|
time_value,
|
|
)
|
|
|
|
|
|
def custom_topocentric_vector(body_type: str, payload: dict, observer: astronomy.Observer, time_value: astronomy.Time) -> astronomy.Vector:
|
|
geocentric_vector = custom_geocentric_vector(body_type, payload, time_value)
|
|
observer_vector = astronomy.ObserverVector(time_value, observer, False)
|
|
return astronomy.Vector(
|
|
geocentric_vector.x - observer_vector.x,
|
|
geocentric_vector.y - observer_vector.y,
|
|
geocentric_vector.z - observer_vector.z,
|
|
time_value,
|
|
)
|
|
|
|
|
|
def custom_altitude_deg(body_type: str, payload: dict, observer: astronomy.Observer, time_value: astronomy.Time) -> float:
|
|
topocentric_vector = custom_topocentric_vector(body_type, payload, observer, time_value)
|
|
rotation = astronomy.Rotation_EQJ_HOR(time_value, observer)
|
|
horizontal_vector = astronomy.RotateVector(rotation, topocentric_vector)
|
|
horizontal = astronomy.HorizonFromVector(horizontal_vector, astronomy.Refraction.Normal)
|
|
return float(horizontal.lat)
|
|
|
|
|
|
def custom_horizontal_coordinates(body_type: str, payload: dict, observer: astronomy.Observer, time_value: astronomy.Time) -> tuple[float, float]:
|
|
topocentric_vector = custom_topocentric_vector(body_type, payload, observer, time_value)
|
|
rotation = astronomy.Rotation_EQJ_HOR(time_value, observer)
|
|
horizontal_vector = astronomy.RotateVector(rotation, topocentric_vector)
|
|
horizontal = astronomy.HorizonFromVector(horizontal_vector, astronomy.Refraction.Normal)
|
|
return float(horizontal.lon), float(horizontal.lat)
|
|
|
|
|
|
def refine_custom_event(
|
|
body_type: str,
|
|
payload: dict,
|
|
observer: astronomy.Observer,
|
|
left_dt: datetime,
|
|
right_dt: datetime,
|
|
*,
|
|
rising: bool,
|
|
) -> datetime:
|
|
for _ in range(24):
|
|
midpoint = left_dt + (right_dt - left_dt) / 2
|
|
altitude = custom_altitude_deg(body_type, payload, observer, dt_to_time(midpoint))
|
|
if (altitude >= 0.0) == rising:
|
|
right_dt = midpoint
|
|
else:
|
|
left_dt = midpoint
|
|
return right_dt
|
|
|
|
|
|
def search_custom_events_for_day(
|
|
body_type: str,
|
|
payload: dict,
|
|
observer: astronomy.Observer,
|
|
local_day_start: datetime,
|
|
tz: ZoneInfo,
|
|
) -> tuple[str | None, str | None]:
|
|
step = timedelta(minutes=CUSTOM_EVENT_STEP_MINUTES)
|
|
day_end = local_day_start + timedelta(days=1)
|
|
|
|
previous_dt = local_day_start.astimezone(timezone.utc)
|
|
previous_altitude = custom_altitude_deg(body_type, payload, observer, dt_to_time(previous_dt))
|
|
rise_time = None
|
|
set_time = None
|
|
|
|
current_dt = previous_dt + step
|
|
while current_dt <= day_end.astimezone(timezone.utc):
|
|
current_altitude = custom_altitude_deg(body_type, payload, observer, dt_to_time(current_dt))
|
|
|
|
if rise_time is None and previous_altitude < 0.0 <= current_altitude:
|
|
rise_time = refine_custom_event(body_type, payload, observer, previous_dt, current_dt, rising=True)
|
|
if set_time is None and previous_altitude >= 0.0 > current_altitude:
|
|
set_time = refine_custom_event(body_type, payload, observer, previous_dt, current_dt, rising=False)
|
|
|
|
previous_dt = current_dt
|
|
previous_altitude = current_altitude
|
|
current_dt += step
|
|
|
|
rise_label = rise_time.astimezone(tz).strftime("%H:%M") if rise_time is not None else None
|
|
set_label = set_time.astimezone(tz).strftime("%H:%M") if set_time is not None else None
|
|
return rise_label, set_label
|
|
|
|
|
|
def build_rows_response(
|
|
*,
|
|
observer: astronomy.Observer,
|
|
timezone_name: str,
|
|
object_key: str,
|
|
object_label: str,
|
|
interval_minutes: int,
|
|
range_value: int,
|
|
range_unit: str,
|
|
row_builder,
|
|
rise_set_builder,
|
|
action_name: str,
|
|
start_dt_str: str = "",
|
|
) -> dict:
|
|
try:
|
|
tz = ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
fail("Zeitzone ist ungueltig.", extra={"details": str(exc)})
|
|
|
|
if start_dt_str:
|
|
try:
|
|
local_start = datetime.strptime(start_dt_str, "%Y-%m-%dT%H:%M").replace(tzinfo=tz)
|
|
except ValueError as exc:
|
|
fail("Startdatum ist ungueltig.", extra={"details": str(exc), "start_dt_str": start_dt_str})
|
|
else:
|
|
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]] = {}
|
|
|
|
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)
|
|
|
|
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)
|
|
rise_set_cache[day_key] = rise_set_builder(local_day_start)
|
|
|
|
rise_label, set_label = rise_set_cache[day_key]
|
|
row = row_builder(current_local, time_value, rise_label, set_label)
|
|
rows.append(row)
|
|
current_local += step
|
|
|
|
return {
|
|
"ok": True,
|
|
"action": action_name,
|
|
"observer": {
|
|
"latitude": observer.latitude,
|
|
"longitude": observer.longitude,
|
|
"elevation": observer.height,
|
|
"timezone": timezone_name,
|
|
},
|
|
"object": {
|
|
"key": object_key,
|
|
"label": object_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 action_planet_ephemeris(args: list[str]) -> dict:
|
|
if len(args) not in {8, 9}:
|
|
fail(
|
|
"Aktion planet_ephemeris erwartet 8 oder 9 Argumente: latitude longitude elevation timezone body intervalMinutes rangeValue rangeUnit [startDt]",
|
|
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()
|
|
start_dt_str = str(args[8]).strip() if len(args) == 9 else ""
|
|
|
|
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})
|
|
|
|
label, body = EPHEMERIS_BODIES[body_name]
|
|
observer = astronomy.Observer(latitude, longitude, elevation)
|
|
skyfield_timescale, skyfield_planets = skyfield_context()
|
|
skyfield_topos = skyfield_observer(latitude, longitude, elevation)
|
|
|
|
def row_builder(current_local: datetime, time_value: astronomy.Time, rise_label: str | None, set_label: str | None) -> dict:
|
|
ra_hours, dec_deg, azimuth_deg, altitude_deg = skyfield_planet_state(
|
|
body_name,
|
|
skyfield_topos,
|
|
current_local.astimezone(timezone.utc),
|
|
skyfield_timescale,
|
|
skyfield_planets,
|
|
)
|
|
return {
|
|
"object_name": label,
|
|
"date_local": current_local.strftime("%d.%m.%Y"),
|
|
"time_local": current_local.strftime("%H:%M"),
|
|
"local_iso": current_local.isoformat(),
|
|
"ra": format_ra_hours(ra_hours),
|
|
"ra_decimal_hours": round(ra_hours, 8),
|
|
"dec": format_dec_deg(dec_deg),
|
|
"dec_decimal_deg": round(dec_deg, 8),
|
|
"azimuth_deg": round(azimuth_deg, 6),
|
|
"altitude_deg": round(altitude_deg, 6),
|
|
"rise": rise_label,
|
|
"set": set_label,
|
|
}
|
|
|
|
def rise_set_builder(local_day_start: datetime) -> tuple[str | None, str | None]:
|
|
try:
|
|
tz = ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
fail("Zeitzone ist ungueltig.", extra={"details": str(exc)})
|
|
|
|
rise, set_ = search_skyfield_rise_set(
|
|
skyfield_topos,
|
|
body_name,
|
|
local_day_start,
|
|
skyfield_timescale,
|
|
skyfield_planets,
|
|
)
|
|
rise_label = rise.astimezone(tz).strftime("%H:%M") if rise is not None else None
|
|
set_label = set_.astimezone(tz).strftime("%H:%M") if set_ is not None else None
|
|
return rise_label, set_label
|
|
|
|
return build_rows_response(
|
|
observer=observer,
|
|
timezone_name=timezone_name,
|
|
object_key=body_name,
|
|
object_label=label,
|
|
interval_minutes=interval_minutes,
|
|
range_value=range_value,
|
|
range_unit=range_unit,
|
|
row_builder=row_builder,
|
|
rise_set_builder=rise_set_builder,
|
|
action_name="planet_ephemeris",
|
|
start_dt_str=start_dt_str,
|
|
)
|
|
|
|
|
|
def action_small_body_ephemeris(args: list[str]) -> dict:
|
|
if len(args) not in {9, 10}:
|
|
fail(
|
|
"Aktion small_body_ephemeris erwartet 9 oder 10 Argumente: latitude longitude elevation timezone bodyType payload intervalMinutes rangeValue rangeUnit [startDt]",
|
|
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_type = str(args[4]).strip().lower()
|
|
|
|
try:
|
|
payload_json = base64.b64decode(args[5]).decode("utf-8")
|
|
payload = json.loads(payload_json)
|
|
except json.JSONDecodeError as exc:
|
|
fail("Objektdaten sind ungueltig.", extra={"details": str(exc)})
|
|
|
|
try:
|
|
interval_minutes = int(args[6])
|
|
range_value = int(args[7])
|
|
except ValueError as exc:
|
|
fail("Intervall oder Zeitraum ist ungueltig.", extra={"details": str(exc), "argv": args})
|
|
|
|
range_unit = str(args[8]).strip()
|
|
start_dt_str = str(args[9]).strip() if len(args) == 10 else ""
|
|
|
|
if body_type not in {"minorplanet", "comet"}:
|
|
fail("Objekttyp ist ungueltig.", extra={"body_type": body_type})
|
|
if not isinstance(payload, dict):
|
|
fail("Objektdaten sind ungueltig.")
|
|
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 range_unit not in {"minutes", "hours", "days", "weeks", "months", "years"}:
|
|
fail("Zeitraum-Einheit ist ungueltig.", extra={"range_unit": range_unit})
|
|
|
|
observer = astronomy.Observer(latitude, longitude, elevation)
|
|
object_label = str(payload.get("designation") or payload.get("label") or "Objekt").strip() or "Objekt"
|
|
object_key = f"{body_type}:{payload.get('id', '')}"
|
|
|
|
def row_builder(current_local: datetime, time_value: astronomy.Time, rise_label: str | None, set_label: str | None) -> dict:
|
|
try:
|
|
topocentric_vector = custom_topocentric_vector(body_type, payload, observer, time_value)
|
|
azimuth_deg, altitude_deg = custom_horizontal_coordinates(body_type, payload, observer, time_value)
|
|
except Exception as exc:
|
|
fail("Die Ephemeriden konnten fuer dieses Objekt nicht berechnet werden.", extra={"details": str(exc), "object": object_label})
|
|
|
|
eq = astronomy.EquatorFromVector(topocentric_vector)
|
|
return {
|
|
"object_name": object_label,
|
|
"date_local": current_local.strftime("%d.%m.%Y"),
|
|
"time_local": current_local.strftime("%H:%M"),
|
|
"local_iso": current_local.isoformat(),
|
|
"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),
|
|
"azimuth_deg": round(float(azimuth_deg), 6),
|
|
"altitude_deg": round(float(altitude_deg), 6),
|
|
"rise": rise_label,
|
|
"set": set_label,
|
|
}
|
|
|
|
def rise_set_builder(local_day_start: datetime) -> tuple[str | None, str | None]:
|
|
try:
|
|
tz = ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
fail("Zeitzone ist ungueltig.", extra={"details": str(exc)})
|
|
|
|
try:
|
|
return search_custom_events_for_day(body_type, payload, observer, local_day_start, tz)
|
|
except Exception as exc:
|
|
fail("Auf- und Untergang konnten fuer dieses Objekt nicht berechnet werden.", extra={"details": str(exc), "object": object_label})
|
|
|
|
return build_rows_response(
|
|
observer=observer,
|
|
timezone_name=timezone_name,
|
|
object_key=object_key,
|
|
object_label=object_label,
|
|
interval_minutes=interval_minutes,
|
|
range_value=range_value,
|
|
range_unit=range_unit,
|
|
row_builder=row_builder,
|
|
rise_set_builder=rise_set_builder,
|
|
action_name="small_body_ephemeris",
|
|
start_dt_str=start_dt_str,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
available_actions = ["planet_ephemeris", "small_body_ephemeris"]
|
|
|
|
if len(sys.argv) < 2:
|
|
fail("Es wurde keine Aktion uebergeben.", extra={"available_actions": available_actions})
|
|
|
|
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
|
|
|
|
if action == "small_body_ephemeris":
|
|
result = action_small_body_ephemeris(args)
|
|
print(json.dumps(result, ensure_ascii=True))
|
|
return
|
|
|
|
fail("Unbekannte Aktion.", extra={"action": action, "available_actions": available_actions})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|