236 lines
9.4 KiB
Python
236 lines
9.4 KiB
Python
"""
|
|
Planetary Events Service - Calculate astronomical events using Skyfield
|
|
Computes conjunctions, oppositions, and other events for major solar system bodies
|
|
"""
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime, timedelta
|
|
from skyfield.api import load, wgs84
|
|
from skyfield import almanac
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PlanetaryEventsService:
|
|
"""Service for calculating planetary astronomical events"""
|
|
|
|
def __init__(self):
|
|
"""Initialize Skyfield ephemeris and timescale"""
|
|
self.ts = None
|
|
self.eph = None
|
|
self._initialized = False
|
|
|
|
def _ensure_initialized(self):
|
|
"""Lazy load ephemeris data (downloads ~30MB on first run)"""
|
|
if not self._initialized:
|
|
logger.info("Loading Skyfield ephemeris (DE421)...")
|
|
self.ts = load.timescale()
|
|
self.eph = load('de421.bsp') # Covers 1900-2050
|
|
self._initialized = True
|
|
logger.info("Skyfield ephemeris loaded successfully")
|
|
|
|
def get_planet_mapping(self) -> Dict[str, Dict[str, str]]:
|
|
"""
|
|
Map database body IDs to Skyfield names
|
|
|
|
Returns:
|
|
Dictionary mapping body_id to Skyfield ephemeris names
|
|
"""
|
|
return {
|
|
'10': {'skyfield': 'sun', 'name': 'Sun', 'name_zh': '太阳'},
|
|
'199': {'skyfield': 'mercury', 'name': 'Mercury', 'name_zh': '水星'},
|
|
'299': {'skyfield': 'venus', 'name': 'Venus', 'name_zh': '金星'},
|
|
'399': {'skyfield': 'earth', 'name': 'Earth', 'name_zh': '地球'},
|
|
'301': {'skyfield': 'moon', 'name': 'Moon', 'name_zh': '月球'},
|
|
'499': {'skyfield': 'mars', 'name': 'Mars', 'name_zh': '火星'},
|
|
'599': {'skyfield': 'jupiter barycenter', 'name': 'Jupiter', 'name_zh': '木星'},
|
|
'699': {'skyfield': 'saturn barycenter', 'name': 'Saturn', 'name_zh': '土星'},
|
|
'799': {'skyfield': 'uranus barycenter', 'name': 'Uranus', 'name_zh': '天王星'},
|
|
'899': {'skyfield': 'neptune barycenter', 'name': 'Neptune', 'name_zh': '海王星'},
|
|
}
|
|
|
|
def calculate_oppositions_conjunctions(
|
|
self,
|
|
body_ids: Optional[List[str]] = None,
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None,
|
|
days_ahead: int = 365
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Calculate oppositions and conjunctions for specified bodies
|
|
|
|
Args:
|
|
body_ids: List of body IDs to calculate (default: all major planets)
|
|
start_date: Start date (default: today)
|
|
end_date: End date (default: start_date + days_ahead)
|
|
days_ahead: Days to look ahead if end_date not specified
|
|
|
|
Returns:
|
|
List of event dictionaries
|
|
"""
|
|
self._ensure_initialized()
|
|
|
|
# Set time range
|
|
if start_date is None:
|
|
start_date = datetime.utcnow()
|
|
if end_date is None:
|
|
end_date = start_date + timedelta(days=days_ahead)
|
|
|
|
t0 = self.ts.utc(start_date.year, start_date.month, start_date.day)
|
|
t1 = self.ts.utc(end_date.year, end_date.month, end_date.day)
|
|
|
|
logger.info(f"Calculating planetary events from {start_date.date()} to {end_date.date()}")
|
|
|
|
# Get planet mapping
|
|
planet_map = self.get_planet_mapping()
|
|
|
|
# Default to major planets (exclude Sun, Moon)
|
|
if body_ids is None:
|
|
body_ids = ['199', '299', '499', '599', '699', '799', '899']
|
|
|
|
# Earth as reference point
|
|
earth = self.eph['earth']
|
|
|
|
events = []
|
|
|
|
for body_id in body_ids:
|
|
if body_id not in planet_map:
|
|
logger.warning(f"Body ID {body_id} not in planet mapping, skipping")
|
|
continue
|
|
|
|
planet_info = planet_map[body_id]
|
|
skyfield_name = planet_info['skyfield']
|
|
|
|
try:
|
|
planet = self.eph[skyfield_name]
|
|
|
|
# Calculate oppositions and conjunctions
|
|
f = almanac.oppositions_conjunctions(self.eph, planet)
|
|
times, event_types = almanac.find_discrete(t0, t1, f)
|
|
|
|
for ti, event_type in zip(times, event_types):
|
|
event_time = ti.utc_datetime()
|
|
|
|
# Convert timezone-aware datetime to naive UTC for database
|
|
# Database expects TIMESTAMP (timezone-naive)
|
|
event_time = event_time.replace(tzinfo=None)
|
|
|
|
# event_type: 0 = conjunction, 1 = opposition
|
|
is_conjunction = (event_type == 0)
|
|
|
|
event_name = '合' if is_conjunction else '冲'
|
|
event_type_en = 'conjunction' if is_conjunction else 'opposition'
|
|
|
|
# Calculate separation angle
|
|
earth_pos = earth.at(ti)
|
|
planet_pos = planet.at(ti)
|
|
separation = earth_pos.separation_from(planet_pos)
|
|
|
|
# Create event data
|
|
event = {
|
|
'body_id': body_id,
|
|
'title': f"{planet_info['name_zh']} {event_name} ({planet_info['name']} {event_type_en.capitalize()})",
|
|
'event_type': event_type_en,
|
|
'event_time': event_time,
|
|
'description': f"{planet_info['name_zh']}将发生{event_name}现象。" +
|
|
(f"与地球的角距离约{separation.degrees:.2f}°。" if is_conjunction else "处于冲日位置，是观测的最佳时机。"),
|
|
'details': {
|
|
'event_subtype': event_name,
|
|
'separation_degrees': round(separation.degrees, 4),
|
|
'planet_name': planet_info['name'],
|
|
'planet_name_zh': planet_info['name_zh'],
|
|
},
|
|
'source': 'skyfield_calculation'
|
|
}
|
|
|
|
events.append(event)
|
|
logger.debug(f"Found {event_type_en}: {planet_info['name']} at {event_time}")
|
|
|
|
except KeyError:
|
|
logger.error(f"Planet {skyfield_name} not found in ephemeris")
|
|
except Exception as e:
|
|
logger.error(f"Error calculating events for {body_id}: {e}")
|
|
|
|
logger.info(f"Calculated {len(events)} planetary events")
|
|
return events
|
|
|
|
def calculate_planetary_distances(
|
|
self,
|
|
body_pairs: List[tuple],
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None,
|
|
days_ahead: int = 365,
|
|
threshold_degrees: float = 5.0
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Calculate close approaches between planet pairs
|
|
|
|
Args:
|
|
body_pairs: List of (body_id1, body_id2) tuples to check
|
|
start_date: Start date
|
|
end_date: End date
|
|
days_ahead: Days to look ahead
|
|
threshold_degrees: Only report if closer than this angle
|
|
|
|
Returns:
|
|
List of close approach events
|
|
"""
|
|
self._ensure_initialized()
|
|
|
|
if start_date is None:
|
|
start_date = datetime.utcnow()
|
|
if end_date is None:
|
|
end_date = start_date + timedelta(days=days_ahead)
|
|
|
|
planet_map = self.get_planet_mapping()
|
|
events = []
|
|
|
|
# Sample every day
|
|
current = start_date
|
|
while current <= end_date:
|
|
t = self.ts.utc(current.year, current.month, current.day)
|
|
|
|
for body_id1, body_id2 in body_pairs:
|
|
if body_id1 not in planet_map or body_id2 not in planet_map:
|
|
continue
|
|
|
|
try:
|
|
planet1 = self.eph[planet_map[body_id1]['skyfield']]
|
|
planet2 = self.eph[planet_map[body_id2]['skyfield']]
|
|
|
|
pos1 = planet1.at(t)
|
|
pos2 = planet2.at(t)
|
|
separation = pos1.separation_from(pos2)
|
|
|
|
if separation.degrees < threshold_degrees:
|
|
# Use naive UTC datetime for database
|
|
event_time = current.replace(tzinfo=None) if hasattr(current, 'tzinfo') else current
|
|
|
|
event = {
|
|
'body_id': body_id1, # Primary body
|
|
'title': f"{planet_map[body_id1]['name_zh']}与{planet_map[body_id2]['name_zh']}接近",
|
|
'event_type': 'close_approach',
|
|
'event_time': event_time,
|
|
'description': f"{planet_map[body_id1]['name_zh']}与{planet_map[body_id2]['name_zh']}的角距离约{separation.degrees:.2f}°，这是较为罕见的天象。",
|
|
'details': {
|
|
'body_id_secondary': body_id2,
|
|
'separation_degrees': round(separation.degrees, 4),
|
|
'planet1_name': planet_map[body_id1]['name'],
|
|
'planet2_name': planet_map[body_id2]['name'],
|
|
},
|
|
'source': 'skyfield_calculation'
|
|
}
|
|
events.append(event)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error calculating distance for {body_id1}-{body_id2}: {e}")
|
|
|
|
current += timedelta(days=1)
|
|
|
|
logger.info(f"Found {len(events)} close approach events")
|
|
return events
|
|
|
|
|
|
# Singleton instance
|
|
planetary_events_service = PlanetaryEventsService()
|