Introduction
The convergence of Large Language Models and astronomical data presents unprecedented opportunities for creating intelligent astronomy assistants. An LLM-based astronomy agentic AI represents a sophisticated system that combines natural language understanding with real-time astronomical calculations, weather monitoring, and celestial object tracking. This article explores the comprehensive development of such a system, focusing on practical implementation details, architectural considerations, and the integration of multiple data sources to create a powerful astronomical companion.
The primary objective of our astronomy agent is to serve as an intelligent intermediary between users and the vast repository of astronomical data available through various APIs and computational libraries. The agent must demonstrate several key capabilities including accurate location determination through multiple methods, real-time weather condition assessment for observing suitability, precise calculation of sunrise and sunset times, identification and tracking of visible celestial objects with their exact coordinates, and comprehensive summarization of astronomical information in an accessible format.
The significance of such a system extends beyond mere data aggregation. By leveraging the reasoning capabilities of modern LLMs, the agent can provide contextual recommendations, explain complex astronomical phenomena, and adapt its responses based on user expertise levels and observing conditions. This creates a more intuitive and educational experience for both amateur astronomers and seasoned observers.
Architectural Overview
The architecture of an LLM-based astronomy agentic AI follows a modular design pattern that emphasizes separation of concerns, scalability, and maintainability. The system comprises several interconnected components that work together to provide comprehensive astronomical assistance.
At the core of the architecture lies the LLM agent framework, which serves as the central orchestrator for all system operations. This framework is responsible for interpreting user queries, determining the appropriate sequence of actions, coordinating with various data sources, and synthesizing responses that combine multiple information streams into coherent and useful outputs.
The location management subsystem handles the critical task of determining or obtaining the user's geographical position. This component supports multiple input methods including direct coordinate input, address-based geocoding, IP-based geolocation, and integration with GPS-enabled devices. The flexibility in location determination ensures that the system can operate effectively across various usage scenarios and device capabilities.
Weather integration forms another crucial component, providing real-time atmospheric conditions that directly impact astronomical observations. This subsystem interfaces with meteorological APIs to retrieve current weather data, cloud coverage information, atmospheric transparency metrics, and short-term forecasting data. The weather component also includes specialized algorithms for calculating astronomical seeing conditions and visibility predictions.
The astronomical calculation engine represents the most specialized component of the system. This engine integrates with multiple astronomical databases and calculation libraries to provide accurate information about celestial object positions, visibility windows, magnitude calculations, and coordinate transformations. The engine maintains up-to-date orbital elements for planets, asteroid ephemeris data, and stellar catalogs to ensure precision in all calculations.
The data synthesis and presentation layer handles the complex task of combining information from multiple sources into coherent responses. This component applies natural language generation techniques to create explanations that are appropriate for the user's knowledge level while maintaining scientific accuracy.
Core Components Implementation
Location Detection and Management
The location detection subsystem represents a critical foundation for all astronomical calculations. Accurate geographical positioning directly affects the precision of celestial object visibility predictions, sunrise and sunset calculations, and local weather conditions.
import requests
import geocoder
from typing import Tuple, Optional, Dict, Any
from dataclasses import dataclass
from abc import ABC, abstractmethod
@dataclass
class LocationData:
"""Comprehensive location information container."""
latitude: float
longitude: float
elevation: float
city: str
country: str
timezone: str
accuracy_radius: float
class LocationProvider(ABC):
"""Abstract base class for location determination strategies."""
@abstractmethod
def get_location(self, query: str = None) -> Optional[LocationData]:
"""Retrieve location information based on the provided query or method."""
pass
class IPLocationProvider(LocationProvider):
"""Determines location based on IP address geolocation."""
def __init__(self, api_key: str = None):
self.api_key = api_key
self.base_url = "http://ip-api.com/json/"
def get_location(self, query: str = None) -> Optional[LocationData]:
"""Retrieve location based on current IP address."""
try:
response = requests.get(f"{self.base_url}?fields=status,message,country,city,lat,lon,timezone,query")
response.raise_for_status()
data = response.json()
if data.get('status') == 'success':
return LocationData(
latitude=data['lat'],
longitude=data['lon'],
elevation=0.0, # IP geolocation doesn't provide elevation
city=data['city'],
country=data['country'],
timezone=data['timezone'],
accuracy_radius=10000.0 # Approximate accuracy for IP geolocation
)
except Exception as e:
print(f"IP location detection failed: {e}")
return None
class GeocodeLocationProvider(LocationProvider):
"""Determines location based on address or place name geocoding."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.opencagedata.com/geocode/v1/json"
def get_location(self, query: str) -> Optional[LocationData]:
"""Geocode an address or place name to coordinates."""
if not query:
return None
try:
params = {
'q': query,
'key': self.api_key,
'limit': 1,
'no_annotations': 0
}
response = requests.get(self.base_url, params=params)
response.raise_for_status()
data = response.json()
if data['results']:
result = data['results'][0]
geometry = result['geometry']
components = result['components']
annotations = result.get('annotations', {})
return LocationData(
latitude=geometry['lat'],
longitude=geometry['lng'],
elevation=annotations.get('elevation', {}).get('apparent', 0.0),
city=components.get('city', components.get('town', components.get('village', ''))),
country=components.get('country', ''),
timezone=annotations.get('timezone', {}).get('name', 'UTC'),
accuracy_radius=result.get('confidence', 1) * 1000 # Convert to meters
)
except Exception as e:
print(f"Geocoding failed: {e}")
return None
class CoordinateLocationProvider(LocationProvider):
"""Handles direct coordinate input with reverse geocoding for context."""
def __init__(self, api_key: str):
self.api_key = api_key
self.geocoder = GeocodeLocationProvider(api_key)
def get_location(self, query: str) -> Optional[LocationData]:
"""Parse coordinates and perform reverse geocoding for context."""
try:
# Parse various coordinate formats
coords = self._parse_coordinates(query)
if not coords:
return None
lat, lon = coords
# Perform reverse geocoding to get location context
reverse_query = f"{lat},{lon}"
location_data = self.geocoder.get_location(reverse_query)
if location_data:
# Update with the exact coordinates provided
location_data.latitude = lat
location_data.longitude = lon
location_data.accuracy_radius = 10.0 # High accuracy for direct coordinates
return location_data
else:
# Fallback to basic location data without context
return LocationData(
latitude=lat,
longitude=lon,
elevation=0.0,
city="Unknown",
country="Unknown",
timezone="UTC",
accuracy_radius=10.0
)
except Exception as e:
print(f"Coordinate parsing failed: {e}")
return None
def _parse_coordinates(self, coord_string: str) -> Optional[Tuple[float, float]]:
"""Parse various coordinate formats into decimal degrees."""
import re
# Remove extra whitespace and normalize
coord_string = re.sub(r'\s+', ' ', coord_string.strip())
# Pattern for decimal degrees (e.g., "40.7128, -74.0060")
decimal_pattern = r'^(-?\d+\.?\d*),?\s*(-?\d+\.?\d*)$'
match = re.match(decimal_pattern, coord_string)
if match:
return float(match.group(1)), float(match.group(2))
# Pattern for degrees, minutes, seconds
dms_pattern = r'(\d+)[°\s]+(\d+)[\'′\s]*(\d*\.?\d*)[\"″\s]*([NSEW])\s*,?\s*(\d+)[°\s]+(\d+)[\'′\s]*(\d*\.?\d*)[\"″\s]*([NSEW])'
match = re.match(dms_pattern, coord_string.upper())
if match:
lat_deg, lat_min, lat_sec, lat_dir, lon_deg, lon_min, lon_sec, lon_dir = match.groups()
lat = float(lat_deg) + float(lat_min)/60 + float(lat_sec or 0)/3600
if lat_dir in ['S']:
lat = -lat
lon = float(lon_deg) + float(lon_min)/60 + float(lon_sec or 0)/3600
if lon_dir in ['W']:
lon = -lon
return lat, lon
return None
class LocationManager:
"""Manages multiple location providers and determines the best location data."""
def __init__(self, geocoding_api_key: str):
self.providers = {
'ip': IPLocationProvider(),
'geocode': GeocodeLocationProvider(geocoding_api_key),
'coordinates': CoordinateLocationProvider(geocoding_api_key)
}
self.cached_location = None
def get_location(self, user_input: str = None) -> Optional[LocationData]:
"""Determine location using the most appropriate method."""
if user_input:
# Try to determine if input is coordinates
if self._looks_like_coordinates(user_input):
location = self.providers['coordinates'].get_location(user_input)
if location:
self.cached_location = location
return location
# Try geocoding for address/place names
location = self.providers['geocode'].get_location(user_input)
if location:
self.cached_location = location
return location
# Fallback to IP-based location
if not self.cached_location:
self.cached_location = self.providers['ip'].get_location()
return self.cached_location
def _looks_like_coordinates(self, text: str) -> bool:
"""Heuristic to determine if text contains coordinates."""
import re
# Check for patterns that suggest coordinates
patterns = [
r'-?\d+\.?\d*,\s*-?\d+\.?\d*', # Decimal degrees
r'\d+[°\s]+\d+[\'′\s]*\d*\.?\d*[\"″\s]*[NSEW]', # DMS format
]
for pattern in patterns:
if re.search(pattern, text.upper()):
return True
return False
The location management system implements a strategy pattern that allows for flexible location determination based on available information and user preferences. The system prioritizes accuracy and provides fallback mechanisms to ensure reliable operation even when primary location sources are unavailable.
Weather Integration System
Weather conditions play a crucial role in astronomical observations, affecting visibility, atmospheric stability, and equipment performance. The weather integration system provides comprehensive meteorological data specifically relevant to astronomical activities.
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
class ObservingCondition(Enum):
"""Enumeration of observing condition quality levels."""
EXCELLENT = "excellent"
GOOD = "good"
FAIR = "fair"
POOR = "poor"
IMPOSSIBLE = "impossible"
@dataclass
class WeatherData:
"""Comprehensive weather information for astronomical observations."""
temperature_celsius: float
humidity_percent: float
pressure_hpa: float
wind_speed_ms: float
wind_direction_degrees: float
cloud_cover_percent: float
visibility_km: float
dew_point_celsius: float
uv_index: float
condition_description: str
timestamp: datetime
# Derived astronomical metrics
seeing_arcsec: Optional[float] = None
transparency_percent: Optional[float] = None
observing_condition: Optional[ObservingCondition] = None
@dataclass
class WeatherForecast:
"""Weather forecast data for planning observations."""
forecast_time: datetime
temperature_celsius: float
humidity_percent: float
cloud_cover_percent: float
wind_speed_ms: float
precipitation_probability: float
condition_description: str
observing_condition: ObservingCondition
class WeatherProvider:
"""Integrates with OpenWeatherMap API for comprehensive weather data."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.openweathermap.org/data/2.5"
self.onecall_url = "https://api.openweathermap.org/data/3.0/onecall"
def get_current_weather(self, latitude: float, longitude: float) -> Optional[WeatherData]:
"""Retrieve current weather conditions for the specified location."""
try:
params = {
'lat': latitude,
'lon': longitude,
'appid': self.api_key,
'units': 'metric'
}
response = requests.get(f"{self.base_url}/weather", params=params)
response.raise_for_status()
data = response.json()
weather_data = WeatherData(
temperature_celsius=data['main']['temp'],
humidity_percent=data['main']['humidity'],
pressure_hpa=data['main']['pressure'],
wind_speed_ms=data.get('wind', {}).get('speed', 0),
wind_direction_degrees=data.get('wind', {}).get('deg', 0),
cloud_cover_percent=data.get('clouds', {}).get('all', 0),
visibility_km=data.get('visibility', 10000) / 1000,
dew_point_celsius=self._calculate_dew_point(
data['main']['temp'],
data['main']['humidity']
),
uv_index=0, # Not available in current weather endpoint
condition_description=data['weather'][0]['description'],
timestamp=datetime.fromtimestamp(data['dt'])
)
# Calculate astronomical observing metrics
weather_data.seeing_arcsec = self._estimate_seeing(weather_data)
weather_data.transparency_percent = self._estimate_transparency(weather_data)
weather_data.observing_condition = self._assess_observing_conditions(weather_data)
return weather_data
except Exception as e:
print(f"Weather data retrieval failed: {e}")
return None
def get_forecast(self, latitude: float, longitude: float, hours: int = 24) -> List[WeatherForecast]:
"""Retrieve weather forecast for astronomical planning."""
try:
params = {
'lat': latitude,
'lon': longitude,
'appid': self.api_key,
'units': 'metric',
'exclude': 'minutely,alerts'
}
response = requests.get(self.onecall_url, params=params)
response.raise_for_status()
data = response.json()
forecasts = []
for hour_data in data.get('hourly', [])[:hours]:
forecast = WeatherForecast(
forecast_time=datetime.fromtimestamp(hour_data['dt']),
temperature_celsius=hour_data['temp'],
humidity_percent=hour_data['humidity'],
cloud_cover_percent=hour_data['clouds'],
wind_speed_ms=hour_data['wind_speed'],
precipitation_probability=hour_data.get('pop', 0) * 100,
condition_description=hour_data['weather'][0]['description'],
observing_condition=self._assess_forecast_conditions(hour_data)
)
forecasts.append(forecast)
return forecasts
except Exception as e:
print(f"Weather forecast retrieval failed: {e}")
return []
def _calculate_dew_point(self, temperature: float, humidity: float) -> float:
"""Calculate dew point using Magnus formula."""
a = 17.27
b = 237.7
alpha = ((a * temperature) / (b + temperature)) + math.log(humidity / 100.0)
return (b * alpha) / (a - alpha)
def _estimate_seeing(self, weather: WeatherData) -> float:
"""Estimate astronomical seeing based on weather conditions."""
base_seeing = 2.0 # Base seeing in arcseconds
# Wind effect on seeing
wind_factor = 1.0 + (weather.wind_speed_ms / 10.0) * 0.5
# Temperature gradient effect (simplified)
temp_factor = 1.0 + abs(weather.temperature_celsius - weather.dew_point_celsius) / 20.0
# Humidity effect
humidity_factor = 1.0 + (weather.humidity_percent - 50) / 100.0 * 0.3
estimated_seeing = base_seeing * wind_factor * temp_factor * humidity_factor
return min(max(estimated_seeing, 0.5), 10.0) # Clamp between 0.5 and 10 arcsec
def _estimate_transparency(self, weather: WeatherData) -> float:
"""Estimate atmospheric transparency percentage."""
base_transparency = 85.0
# Cloud cover impact
cloud_impact = weather.cloud_cover_percent * 0.8
# Humidity impact
humidity_impact = max(0, weather.humidity_percent - 60) * 0.3
# Visibility impact
visibility_impact = max(0, (20 - weather.visibility_km) * 2)
transparency = base_transparency - cloud_impact - humidity_impact - visibility_impact
return max(min(transparency, 100.0), 0.0)
def _assess_observing_conditions(self, weather: WeatherData) -> ObservingCondition:
"""Assess overall observing conditions based on weather data."""
if weather.cloud_cover_percent > 80:
return ObservingCondition.IMPOSSIBLE
elif weather.cloud_cover_percent > 60:
return ObservingCondition.POOR
elif weather.cloud_cover_percent > 30:
return ObservingCondition.FAIR
elif weather.cloud_cover_percent > 10:
return ObservingCondition.GOOD
else:
return ObservingCondition.EXCELLENT
def _assess_forecast_conditions(self, hour_data: Dict[str, Any]) -> ObservingCondition:
"""Assess observing conditions for forecast data."""
cloud_cover = hour_data['clouds']
precipitation = hour_data.get('pop', 0) * 100
if cloud_cover > 80 or precipitation > 50:
return ObservingCondition.IMPOSSIBLE
elif cloud_cover > 60 or precipitation > 30:
return ObservingCondition.POOR
elif cloud_cover > 30 or precipitation > 10:
return ObservingCondition.FAIR
elif cloud_cover > 10:
return ObservingCondition.GOOD
else:
return ObservingCondition.EXCELLENT
import math
class AstronomicalWeatherAnalyzer:
"""Provides specialized analysis of weather data for astronomical purposes."""
@staticmethod
def calculate_limiting_magnitude(weather: WeatherData, light_pollution_class: str = "suburban") -> float:
"""Estimate limiting magnitude based on weather and light pollution."""
base_magnitudes = {
"rural": 6.5,
"suburban": 5.5,
"urban": 4.5,
"city": 3.5
}
base_mag = base_magnitudes.get(light_pollution_class, 5.5)
# Transparency effect
transparency_factor = weather.transparency_percent / 100.0
# Humidity effect on extinction
humidity_extinction = weather.humidity_percent / 100.0 * 0.5
estimated_magnitude = base_mag * transparency_factor - humidity_extinction
return max(min(estimated_magnitude, 7.0), 2.0)
@staticmethod
def recommend_observation_targets(weather: WeatherData) -> List[str]:
"""Recommend observation targets based on current conditions."""
recommendations = []
if weather.observing_condition == ObservingCondition.EXCELLENT:
recommendations.extend([
"Deep sky objects (galaxies, nebulae)",
"Double stars",
"Planetary details",
"Lunar surface features"
])
elif weather.observing_condition == ObservingCondition.GOOD:
recommendations.extend([
"Bright deep sky objects",
"Planets",
"Moon",
"Bright double stars"
])
elif weather.observing_condition == ObservingCondition.FAIR:
recommendations.extend([
"Planets",
"Moon",
"Bright stars"
])
elif weather.observing_condition == ObservingCondition.POOR:
recommendations.extend([
"Moon (if visible)",
"Bright planets"
])
return recommendations
The weather integration system provides not only basic meteorological data but also specialized astronomical metrics such as seeing estimates, transparency calculations, and observing condition assessments. These derived metrics help users make informed decisions about observation planning and target selection.
Astronomical Calculation Engine
The astronomical calculation engine forms the scientific core of the system, providing accurate calculations for celestial object positions, visibility predictions, and temporal astronomical events.
import ephem
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple, Any
from dataclasses import dataclass
from enum import Enum
import math
class ObjectType(Enum):
"""Classification of astronomical objects."""
PLANET = "planet"
MOON = "moon"
STAR = "star"
DEEP_SKY = "deep_sky"
SATELLITE = "satellite"
COMET = "comet"
ASTEROID = "asteroid"
@dataclass
class CelestialObject:
"""Comprehensive information about a celestial object."""
name: str
object_type: ObjectType
right_ascension: float # Hours
declination: float # Degrees
altitude: float # Degrees above horizon
azimuth: float # Degrees from north
magnitude: float
distance_au: Optional[float] = None
constellation: Optional[str] = None
rise_time: Optional[datetime] = None
set_time: Optional[datetime] = None
transit_time: Optional[datetime] = None
is_visible: bool = False
visibility_quality: str = "unknown"
description: Optional[str] = None
@dataclass
class SunMoonData:
"""Solar and lunar information for a specific date and location."""
sunrise: datetime
sunset: datetime
solar_noon: datetime
moonrise: Optional[datetime]
moonset: Optional[datetime]
moon_phase: float # 0-1, where 0 is new moon, 0.5 is full moon
moon_illumination: float # Percentage
moon_age_days: float
civil_twilight_begin: datetime
civil_twilight_end: datetime
nautical_twilight_begin: datetime
nautical_twilight_end: datetime
astronomical_twilight_begin: datetime
astronomical_twilight_end: datetime
class AstronomicalCalculator:
"""Provides comprehensive astronomical calculations using PyEphem."""
def __init__(self):
self.observer = ephem.Observer()
self.planet_catalog = self._initialize_planet_catalog()
self.star_catalog = self._initialize_star_catalog()
def set_location(self, latitude: float, longitude: float, elevation: float = 0):
"""Configure observer location for calculations."""
self.observer.lat = str(latitude)
self.observer.lon = str(longitude)
self.observer.elevation = elevation
def set_time(self, observation_time: datetime = None):
"""Set observation time, defaults to current time."""
if observation_time is None:
observation_time = datetime.utcnow()
self.observer.date = observation_time
def get_sun_moon_data(self, date: datetime = None) -> SunMoonData:
"""Calculate comprehensive sun and moon data for the specified date."""
if date is None:
date = datetime.utcnow()
# Set observer date
self.observer.date = date
sun = ephem.Sun()
moon = ephem.Moon()
# Calculate sun times
try:
sunrise = self.observer.next_rising(sun).datetime()
sunset = self.observer.next_setting(sun).datetime()
solar_noon = self.observer.next_transit(sun).datetime()
except ephem.CircumpolarError:
# Handle polar regions
sunrise = sunset = solar_noon = date
# Calculate moon times
try:
moonrise = self.observer.next_rising(moon).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
moonrise = None
try:
moonset = self.observer.next_setting(moon).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
moonset = None
# Calculate twilight times
self.observer.horizon = '-6' # Civil twilight
try:
civil_begin = self.observer.previous_rising(sun, use_center=True).datetime()
civil_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
civil_begin = civil_end = date
self.observer.horizon = '-12' # Nautical twilight
try:
nautical_begin = self.observer.previous_rising(sun, use_center=True).datetime()
nautical_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
nautical_begin = nautical_end = date
self.observer.horizon = '-18' # Astronomical twilight
try:
astro_begin = self.observer.previous_rising(sun, use_center=True).datetime()
astro_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
astro_begin = astro_end = date
# Reset horizon
self.observer.horizon = '0'
# Calculate moon phase information
moon.compute(self.observer)
moon_phase = moon.moon_phase
moon_illumination = moon.phase
# Calculate moon age
previous_new_moon = ephem.previous_new_moon(date)
moon_age = (date - previous_new_moon.datetime()).days
return SunMoonData(
sunrise=sunrise,
sunset=sunset,
solar_noon=solar_noon,
moonrise=moonrise,
moonset=moonset,
moon_phase=moon_phase,
moon_illumination=moon_illumination,
moon_age_days=moon_age,
civil_twilight_begin=civil_begin,
civil_twilight_end=civil_end,
nautical_twilight_begin=nautical_begin,
nautical_twilight_end=nautical_end,
astronomical_twilight_begin=astro_begin,
astronomical_twilight_end=astro_end
)
def get_visible_planets(self) -> List[CelestialObject]:
"""Calculate positions and visibility of all planets."""
visible_planets = []
for planet_name, planet_obj in self.planet_catalog.items():
planet_obj.compute(self.observer)
# Convert coordinates
ra_hours = float(planet_obj.ra) * 12 / math.pi
dec_degrees = float(planet_obj.dec) * 180 / math.pi
alt_degrees = float(planet_obj.alt) * 180 / math.pi
az_degrees = float(planet_obj.az) * 180 / math.pi
# Determine visibility
is_visible = alt_degrees > 0
visibility_quality = self._assess_visibility(alt_degrees, planet_obj.mag)
# Calculate rise/set times
try:
rise_time = self.observer.next_rising(planet_obj).datetime()
set_time = self.observer.next_setting(planet_obj).datetime()
transit_time = self.observer.next_transit(planet_obj).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
rise_time = set_time = transit_time = None
planet = CelestialObject(
name=planet_name,
object_type=ObjectType.PLANET,
right_ascension=ra_hours,
declination=dec_degrees,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=planet_obj.mag,
distance_au=planet_obj.earth_distance,
rise_time=rise_time,
set_time=set_time,
transit_time=transit_time,
is_visible=is_visible,
visibility_quality=visibility_quality,
description=self._get_planet_description(planet_name, planet_obj)
)
visible_planets.append(planet)
return visible_planets
def get_bright_stars(self, magnitude_limit: float = 3.0) -> List[CelestialObject]:
"""Get bright stars visible from current location."""
visible_stars = []
for star_name, star_data in self.star_catalog.items():
star = ephem.FixedBody()
star._ra = star_data['ra']
star._dec = star_data['dec']
star._epoch = '2000'
star.compute(self.observer)
if star.mag <= magnitude_limit:
alt_degrees = float(star.alt) * 180 / math.pi
az_degrees = float(star.az) * 180 / math.pi
is_visible = alt_degrees > 0
visibility_quality = self._assess_visibility(alt_degrees, star.mag)
star_obj = CelestialObject(
name=star_name,
object_type=ObjectType.STAR,
right_ascension=float(star.ra) * 12 / math.pi,
declination=float(star.dec) * 180 / math.pi,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=star.mag,
constellation=star_data.get('constellation'),
is_visible=is_visible,
visibility_quality=visibility_quality,
description=f"Bright star in constellation {star_data.get('constellation', 'Unknown')}"
)
visible_stars.append(star_obj)
return [star for star in visible_stars if star.is_visible]
def get_deep_sky_objects(self, magnitude_limit: float = 8.0) -> List[CelestialObject]:
"""Get visible deep sky objects (simplified catalog)."""
# This is a simplified implementation. In practice, you would use
# a comprehensive catalog like the Messier catalog or NGC catalog
messier_objects = {
'M31': {'ra': '00:42:44', 'dec': '+41:16:09', 'mag': 3.4, 'type': 'Galaxy'},
'M42': {'ra': '05:35:17', 'dec': '-05:23:14', 'mag': 4.0, 'type': 'Nebula'},
'M45': {'ra': '03:47:29', 'dec': '+24:07:00', 'mag': 1.6, 'type': 'Star Cluster'},
'M13': {'ra': '16:41:41', 'dec': '+36:27:37', 'mag': 5.8, 'type': 'Globular Cluster'},
'M57': {'ra': '18:53:35', 'dec': '+33:01:45', 'mag': 8.8, 'type': 'Planetary Nebula'},
}
visible_objects = []
for obj_name, obj_data in messier_objects.items():
if obj_data['mag'] <= magnitude_limit:
obj = ephem.FixedBody()
obj._ra = obj_data['ra']
obj._dec = obj_data['dec']
obj._epoch = '2000'
obj.compute(self.observer)
alt_degrees = float(obj.alt) * 180 / math.pi
az_degrees = float(obj.az) * 180 / math.pi
is_visible = alt_degrees > 10 # Require at least 10 degrees altitude
visibility_quality = self._assess_visibility(alt_degrees, obj_data['mag'])
if is_visible:
deep_sky_obj = CelestialObject(
name=obj_name,
object_type=ObjectType.DEEP_SKY,
right_ascension=float(obj.ra) * 12 / math.pi,
declination=float(obj.dec) * 180 / math.pi,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=obj_data['mag'],
is_visible=is_visible,
visibility_quality=visibility_quality,
description=f"{obj_data['type']} - {obj_name}"
)
visible_objects.append(deep_sky_obj)
return visible_objects
def _initialize_planet_catalog(self) -> Dict[str, Any]:
"""Initialize catalog of planets for calculations."""
return {
'Mercury': ephem.Mercury(),
'Venus': ephem.Venus(),
'Mars': ephem.Mars(),
'Jupiter': ephem.Jupiter(),
'Saturn': ephem.Saturn(),
'Uranus': ephem.Uranus(),
'Neptune': ephem.Neptune()
}
def _initialize_star_catalog(self) -> Dict[str, Dict[str, Any]]:
"""Initialize catalog of bright stars."""
return {
'Sirius': {'ra': '06:45:08.9', 'dec': '-16:42:58', 'mag': -1.46, 'constellation': 'Canis Major'},
'Canopus': {'ra': '06:23:57.1', 'dec': '-52:41:44', 'mag': -0.74, 'constellation': 'Carina'},
'Arcturus': {'ra': '14:15:39.7', 'dec': '+19:10:56', 'mag': -0.05, 'constellation': 'Boötes'},
'Vega': {'ra': '18:36:56.3', 'dec': '+38:47:01', 'mag': 0.03, 'constellation': 'Lyra'},
'Capella': {'ra': '05:16:41.4', 'dec': '+45:59:53', 'mag': 0.08, 'constellation': 'Auriga'},
'Rigel': {'ra': '05:14:32.3', 'dec': '-08:12:06', 'mag': 0.13, 'constellation': 'Orion'},
'Procyon': {'ra': '07:39:18.1', 'dec': '+05:13:30', 'mag': 0.34, 'constellation': 'Canis Minor'},
'Betelgeuse': {'ra': '05:55:10.3', 'dec': '+07:24:25', 'mag': 0.50, 'constellation': 'Orion'},
'Aldebaran': {'ra': '04:35:55.2', 'dec': '+16:30:33', 'mag': 0.85, 'constellation': 'Taurus'},
'Antares': {'ra': '16:29:24.5', 'dec': '-26:25:55', 'mag': 1.09, 'constellation': 'Scorpius'}
}
def _assess_visibility(self, altitude: float, magnitude: float) -> str:
"""Assess visibility quality based on altitude and magnitude."""
if altitude < 0:
return "not visible"
elif altitude < 10:
return "very low"
elif altitude < 30:
return "low"
elif altitude < 60:
return "good"
else:
return "excellent"
def _get_planet_description(self, planet_name: str, planet_obj: Any) -> str:
"""Generate descriptive information about a planet."""
descriptions = {
'Mercury': f"Closest planet to the Sun, currently {planet_obj.earth_distance:.2f} AU away",
'Venus': f"Brightest planet, currently {planet_obj.earth_distance:.2f} AU away",
'Mars': f"The Red Planet, currently {planet_obj.earth_distance:.2f} AU away",
'Jupiter': f"Largest planet, currently {planet_obj.earth_distance:.2f} AU away",
'Saturn': f"Ringed planet, currently {planet_obj.earth_distance:.2f} AU away",
'Uranus': f"Ice giant, currently {planet_obj.earth_distance:.2f} AU away",
'Neptune': f"Outermost planet, currently {planet_obj.earth_distance:.2f} AU away"
}
return descriptions.get(planet_name, f"Planet {planet_name}")
class SkyObjectSummarizer:
"""Generates comprehensive summaries of celestial objects."""
@staticmethod
def summarize_object(obj: CelestialObject, weather: Optional[WeatherData] = None) -> str:
"""Generate a detailed summary of a celestial object."""
summary_parts = []
# Basic identification
summary_parts.append(f"{obj.name} ({obj.object_type.value})")
# Position information
ra_hours = int(obj.right_ascension)
ra_minutes = int((obj.right_ascension - ra_hours) * 60)
ra_seconds = ((obj.right_ascension - ra_hours) * 60 - ra_minutes) * 60
dec_sign = "+" if obj.declination >= 0 else "-"
dec_degrees = int(abs(obj.declination))
dec_minutes = int((abs(obj.declination) - dec_degrees) * 60)
dec_seconds = ((abs(obj.declination) - dec_degrees) * 60 - dec_minutes) * 60
summary_parts.append(f"Position: RA {ra_hours:02d}h {ra_minutes:02d}m {ra_seconds:04.1f}s, "
f"Dec {dec_sign}{dec_degrees:02d}° {dec_minutes:02d}' {dec_seconds:04.1f}\"")
# Current sky position
summary_parts.append(f"Current position: {obj.altitude:.1f}° altitude, {obj.azimuth:.1f}° azimuth")
# Magnitude and visibility
summary_parts.append(f"Magnitude: {obj.magnitude:.1f}, Visibility: {obj.visibility_quality}")
# Timing information
if obj.rise_time:
summary_parts.append(f"Rises: {obj.rise_time.strftime('%H:%M')}")
if obj.set_time:
summary_parts.append(f"Sets: {obj.set_time.strftime('%H:%M')}")
if obj.transit_time:
summary_parts.append(f"Transits: {obj.transit_time.strftime('%H:%M')}")
# Additional information
if obj.distance_au:
summary_parts.append(f"Distance: {obj.distance_au:.2f} AU")
if obj.constellation:
summary_parts.append(f"Constellation: {obj.constellation}")
if obj.description:
summary_parts.append(f"Description: {obj.description}")
# Weather-based observing advice
if weather and obj.is_visible:
advice = SkyObjectSummarizer._generate_observing_advice(obj, weather)
if advice:
summary_parts.append(f"Observing advice: {advice}")
return ". ".join(summary_parts) + "."
@staticmethod
def _generate_observing_advice(obj: CelestialObject, weather: WeatherData) -> str:
"""Generate observing advice based on object type and weather."""
advice_parts = []
if obj.object_type == ObjectType.PLANET:
if weather.seeing_arcsec and weather.seeing_arcsec < 2.0:
advice_parts.append("excellent seeing for planetary details")
elif weather.seeing_arcsec and weather.seeing_arcsec > 4.0:
advice_parts.append("poor seeing may limit planetary detail visibility")
elif obj.object_type == ObjectType.DEEP_SKY:
if weather.transparency_percent and weather.transparency_percent > 80:
advice_parts.append("excellent transparency for deep sky observation")
elif weather.transparency_percent and weather.transparency_percent < 60:
advice_parts.append("reduced transparency may affect faint object visibility")
if obj.altitude < 30:
advice_parts.append("low altitude may cause atmospheric distortion")
if weather.moon_illumination and weather.moon_illumination > 75:
advice_parts.append("bright moonlight may wash out faint objects")
return ", ".join(advice_parts) if advice_parts else ""
The astronomical calculation engine provides comprehensive celestial mechanics calculations using the PyEphem library, which offers high precision ephemeris calculations based on the same algorithms used by professional observatories. The engine handles coordinate transformations, visibility predictions, and generates detailed object summaries.
LLM Agent Framework Integration
The LLM agent framework serves as the intelligent coordinator that interprets user queries, orchestrates data collection from various sources, and synthesizes comprehensive responses that combine astronomical calculations with contextual information.
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass
from abc import ABC, abstractmethod
import json
import re
from datetime import datetime
@dataclass
class AgentAction:
"""Represents an action the agent can take."""
name: str
description: str
parameters: Dict[str, Any]
function: Callable
@dataclass
class AgentResponse:
"""Structured response from the agent."""
content: str
actions_taken: List[str]
data_sources: List[str]
confidence: float
suggestions: List[str]
class AstronomyTool(ABC):
"""Abstract base class for astronomy tools."""
@abstractmethod
def get_name(self) -> str:
"""Return the tool name."""
pass
@abstractmethod
def get_description(self) -> str:
"""Return the tool description for the LLM."""
pass
@abstractmethod
def execute(self, **kwargs) -> Dict[str, Any]:
"""Execute the tool with given parameters."""
pass
class LocationTool(AstronomyTool):
"""Tool for location detection and management."""
def __init__(self, location_manager: LocationManager):
self.location_manager = location_manager
def get_name(self) -> str:
return "get_location"
def get_description(self) -> str:
return """Determines the user's location from various inputs including:
- IP-based geolocation (automatic)
- Address or place name (e.g., "New York", "London Observatory")
- Coordinates (e.g., "40.7128, -74.0060" or "40°42'46\"N 74°00'22\"W")
Returns latitude, longitude, elevation, city, country, and timezone."""
def execute(self, user_input: str = None) -> Dict[str, Any]:
"""Execute location detection."""
location = self.location_manager.get_location(user_input)
if location:
return {
"success": True,
"latitude": location.latitude,
"longitude": location.longitude,
"elevation": location.elevation,
"city": location.city,
"country": location.country,
"timezone": location.timezone,
"accuracy": location.accuracy_radius
}
else:
return {
"success": False,
"error": "Could not determine location"
}
class WeatherTool(AstronomyTool):
"""Tool for weather data retrieval and analysis."""
def __init__(self, weather_provider: WeatherProvider):
self.weather_provider = weather_provider
def get_name(self) -> str:
return "get_weather"
def get_description(self) -> str:
return """Retrieves current weather conditions and astronomical observing metrics including:
- Temperature, humidity, pressure, wind conditions
- Cloud cover percentage and visibility
- Estimated seeing conditions and atmospheric transparency
- Overall observing condition assessment
- Recommended observation targets based on conditions"""
def execute(self, latitude: float, longitude: float) -> Dict[str, Any]:
"""Execute weather data retrieval."""
weather = self.weather_provider.get_current_weather(latitude, longitude)
if weather:
recommendations = AstronomicalWeatherAnalyzer.recommend_observation_targets(weather)
limiting_mag = AstronomicalWeatherAnalyzer.calculate_limiting_magnitude(weather)
return {
"success": True,
"temperature_celsius": weather.temperature_celsius,
"humidity_percent": weather.humidity_percent,
"cloud_cover_percent": weather.cloud_cover_percent,
"wind_speed_ms": weather.wind_speed_ms,
"visibility_km": weather.visibility_km,
"seeing_arcsec": weather.seeing_arcsec,
"transparency_percent": weather.transparency_percent,
"observing_condition": weather.observing_condition.value,
"limiting_magnitude": limiting_mag,
"recommendations": recommendations,
"condition_description": weather.condition_description
}
else:
return {
"success": False,
"error": "Could not retrieve weather data"
}
class SunMoonTool(AstronomyTool):
"""Tool for sun and moon calculations."""
def __init__(self, calculator: AstronomicalCalculator):
self.calculator = calculator
def get_name(self) -> str:
return "get_sun_moon_times"
def get_description(self) -> str:
return """Calculates sun and moon times and information including:
- Sunrise, sunset, and solar noon times
- Moonrise and moonset times
- Moon phase, illumination percentage, and age
- Civil, nautical, and astronomical twilight times
- Best times for astronomical observations"""
def execute(self, latitude: float, longitude: float, elevation: float = 0, date: str = None) -> Dict[str, Any]:
"""Execute sun and moon calculations."""
self.calculator.set_location(latitude, longitude, elevation)
if date:
try:
target_date = datetime.fromisoformat(date)
except ValueError:
target_date = datetime.utcnow()
else:
target_date = datetime.utcnow()
sun_moon_data = self.calculator.get_sun_moon_data(target_date)
return {
"success": True,
"sunrise": sun_moon_data.sunrise.strftime("%H:%M"),
"sunset": sun_moon_data.sunset.strftime("%H:%M"),
"solar_noon": sun_moon_data.solar_noon.strftime("%H:%M"),
"moonrise": sun_moon_data.moonrise.strftime("%H:%M") if sun_moon_data.moonrise else "No moonrise",
"moonset": sun_moon_data.moonset.strftime("%H:%M") if sun_moon_data.moonset else "No moonset",
"moon_phase": sun_moon_data.moon_phase,
"moon_illumination": sun_moon_data.moon_illumination,
"moon_age_days": sun_moon_data.moon_age_days,
"civil_twilight_begin": sun_moon_data.civil_twilight_begin.strftime("%H:%M"),
"civil_twilight_end": sun_moon_data.civil_twilight_end.strftime("%H:%M"),
"nautical_twilight_begin": sun_moon_data.nautical_twilight_begin.strftime("%H:%M"),
"nautical_twilight_end": sun_moon_data.nautical_twilight_end.strftime("%H:%M"),
"astronomical_twilight_begin": sun_moon_data.astronomical_twilight_begin.strftime("%H:%M"),
"astronomical_twilight_end": sun_moon_data.astronomical_twilight_end.strftime("%H:%M")
}
class SkyObjectsTool(AstronomyTool):
"""Tool for celestial object visibility and information."""
def __init__(self, calculator: AstronomicalCalculator):
self.calculator = calculator
def get_name(self) -> str:
return "get_visible_objects"
def get_description(self) -> str:
return """Identifies currently visible celestial objects including:
- Planets with positions, magnitudes, and rise/set times
- Bright stars and their constellations
- Deep sky objects (galaxies, nebulae, clusters)
- Detailed coordinates and visibility information
- Observing recommendations for each object"""
def execute(self, latitude: float, longitude: float, elevation: float = 0,
object_types: List[str] = None, magnitude_limit: float = 6.0) -> Dict[str, Any]:
"""Execute sky object visibility calculations."""
self.calculator.set_location(latitude, longitude, elevation)
self.calculator.set_time()
if object_types is None:
object_types = ["planets", "stars", "deep_sky"]
results = {"success": True, "objects": []}
if "planets" in object_types:
planets = self.calculator.get_visible_planets()
for planet in planets:
if planet.is_visible:
results["objects"].append({
"name": planet.name,
"type": planet.object_type.value,
"altitude": planet.altitude,
"azimuth": planet.azimuth,
"magnitude": planet.magnitude,
"right_ascension": planet.right_ascension,
"declination": planet.declination,
"distance_au": planet.distance_au,
"rise_time": planet.rise_time.strftime("%H:%M") if planet.rise_time else None,
"set_time": planet.set_time.strftime("%H:%M") if planet.set_time else None,
"transit_time": planet.transit_time.strftime("%H:%M") if planet.transit_time else None,
"visibility_quality": planet.visibility_quality,
"description": planet.description
})
if "stars" in object_types:
stars = self.calculator.get_bright_stars(magnitude_limit)
for star in stars:
results["objects"].append({
"name": star.name,
"type": star.object_type.value,
"altitude": star.altitude,
"azimuth": star.azimuth,
"magnitude": star.magnitude,
"right_ascension": star.right_ascension,
"declination": star.declination,
"constellation": star.constellation,
"visibility_quality": star.visibility_quality,
"description": star.description
})
if "deep_sky" in object_types:
deep_sky = self.calculator.get_deep_sky_objects(magnitude_limit)
for obj in deep_sky:
results["objects"].append({
"name": obj.name,
"type": obj.object_type.value,
"altitude": obj.altitude,
"azimuth": obj.azimuth,
"magnitude": obj.magnitude,
"right_ascension": obj.right_ascension,
"declination": obj.declination,
"visibility_quality": obj.visibility_quality,
"description": obj.description
})
return results
class AstronomyAgent:
"""Main LLM-based astronomy agent that coordinates all tools and provides intelligent responses."""
def __init__(self, llm_client, location_manager: LocationManager,
weather_provider: WeatherProvider, calculator: AstronomicalCalculator):
self.llm_client = llm_client
self.tools = {
"location": LocationTool(location_manager),
"weather": WeatherTool(weather_provider),
"sun_moon": SunMoonTool(calculator),
"sky_objects": SkyObjectsTool(calculator)
}
self.conversation_history = []
self.current_location = None
self.current_weather = None
def process_query(self, user_query: str) -> AgentResponse:
"""Process a user query and return a comprehensive response."""
# Analyze the query to determine required actions
required_actions = self._analyze_query(user_query)
actions_taken = []
data_sources = []
collected_data = {}
# Execute required actions
for action in required_actions:
try:
if action == "location" and not self.current_location:
location_result = self.tools["location"].execute()
if location_result["success"]:
self.current_location = location_result
collected_data["location"] = location_result
actions_taken.append("Determined user location")
data_sources.append("IP Geolocation")
elif action == "weather" and self.current_location:
weather_result = self.tools["weather"].execute(
self.current_location["latitude"],
self.current_location["longitude"]
)
if weather_result["success"]:
self.current_weather = weather_result
collected_data["weather"] = weather_result
actions_taken.append("Retrieved current weather conditions")
data_sources.append("OpenWeatherMap API")
elif action == "sun_moon" and self.current_location:
sun_moon_result = self.tools["sun_moon"].execute(
self.current_location["latitude"],
self.current_location["longitude"],
self.current_location["elevation"]
)
if sun_moon_result["success"]:
collected_data["sun_moon"] = sun_moon_result
actions_taken.append("Calculated sun and moon times")
data_sources.append("Astronomical calculations")
elif action == "sky_objects" and self.current_location:
objects_result = self.tools["sky_objects"].execute(
self.current_location["latitude"],
self.current_location["longitude"],
self.current_location["elevation"]
)
if objects_result["success"]:
collected_data["sky_objects"] = objects_result
actions_taken.append("Identified visible celestial objects")
data_sources.append("Astronomical catalogs and ephemeris")
except Exception as e:
print(f"Error executing action {action}: {e}")
# Generate comprehensive response using LLM
response_content = self._generate_response(user_query, collected_data)
# Generate suggestions for follow-up queries
suggestions = self._generate_suggestions(collected_data)
return AgentResponse(
content=response_content,
actions_taken=actions_taken,
data_sources=data_sources,
confidence=0.9, # Could be calculated based on data quality
suggestions=suggestions
)
def _analyze_query(self, query: str) -> List[str]:
"""Analyze user query to determine which tools to use."""
query_lower = query.lower()
required_actions = []
# Always need location for astronomical calculations
required_actions.append("location")
# Check for weather-related keywords
weather_keywords = ["weather", "cloud", "seeing", "transparency", "observing conditions"]
if any(keyword in query_lower for keyword in weather_keywords):
required_actions.append("weather")
# Check for sun/moon related keywords
sun_moon_keywords = ["sunrise", "sunset", "moonrise", "moonset", "twilight", "moon phase"]
if any(keyword in query_lower for keyword in sun_moon_keywords):
required_actions.append("sun_moon")
# Check for sky object keywords
object_keywords = ["planet", "star", "visible", "tonight", "sky", "object", "constellation"]
if any(keyword in query_lower for keyword in object_keywords):
required_actions.append("sky_objects")
# If no specific keywords found, include all for comprehensive response
if len(required_actions) == 1: # Only location
required_actions.extend(["weather", "sun_moon", "sky_objects"])
return required_actions
def _generate_response(self, query: str, data: Dict[str, Any]) -> str:
"""Generate a comprehensive response using the LLM."""
# Create a structured prompt with all collected data
prompt = self._build_prompt(query, data)
try:
# This would interface with your chosen LLM
# For example, using OpenAI API, Anthropic Claude, or local models
response = self.llm_client.generate_response(prompt)
return response
except Exception as e:
# Fallback to template-based response
return self._generate_template_response(query, data)
def _build_prompt(self, query: str, data: Dict[str, Any]) -> str:
"""Build a comprehensive prompt for the LLM."""
prompt_parts = [
"You are an expert astronomy assistant. Provide a comprehensive and helpful response to the user's query.",
f"User Query: {query}",
"",
"Available Data:"
]
if "location" in data:
loc = data["location"]
prompt_parts.append(f"Location: {loc['city']}, {loc['country']} ({loc['latitude']:.4f}°, {loc['longitude']:.4f}°)")
if "weather" in data:
weather = data["weather"]
prompt_parts.append(f"Weather: {weather['condition_description']}, {weather['temperature_celsius']:.1f}°C")
prompt_parts.append(f"Cloud cover: {weather['cloud_cover_percent']}%, Observing conditions: {weather['observing_condition']}")
if weather.get('recommendations'):
prompt_parts.append(f"Recommended targets: {', '.join(weather['recommendations'])}")
if "sun_moon" in data:
sun_moon = data["sun_moon"]
prompt_parts.append(f"Sunrise: {sun_moon['sunrise']}, Sunset: {sun_moon['sunset']}")
prompt_parts.append(f"Moon: {sun_moon['moon_illumination']:.0f}% illuminated, {sun_moon['moon_age_days']:.1f} days old")
prompt_parts.append(f"Astronomical twilight: {sun_moon['astronomical_twilight_begin']} - {sun_moon['astronomical_twilight_end']}")
if "sky_objects" in data:
objects = data["sky_objects"]["objects"]
prompt_parts.append(f"Visible objects ({len(objects)} total):")
for obj in objects[:10]: # Limit to first 10 objects
prompt_parts.append(f"- {obj['name']} ({obj['type']}): {obj['altitude']:.1f}° alt, mag {obj['magnitude']:.1f}")
prompt_parts.extend([
"",
"Please provide a helpful, accurate, and engaging response that:",
"1. Directly answers the user's question",
"2. Explains any relevant astronomical concepts",
"3. Provides practical observing advice when appropriate",
"4. Uses the provided data to give specific, actionable information",
"5. Maintains scientific accuracy while being accessible"
])
return "\n".join(prompt_parts)
def _generate_template_response(self, query: str, data: Dict[str, Any]) -> str:
"""Generate a fallback template-based response."""
response_parts = []
if "location" in data:
loc = data["location"]
response_parts.append(f"Based on your location in {loc['city']}, {loc['country']}:")
if "weather" in data:
weather = data["weather"]
response_parts.append(f"Current weather conditions show {weather['condition_description']} with {weather['cloud_cover_percent']}% cloud cover. Observing conditions are {weather['observing_condition']}.")
if "sun_moon" in data:
sun_moon = data["sun_moon"]
response_parts.append(f"The sun sets at {sun_moon['sunset']} and astronomical twilight ends at {sun_moon['astronomical_twilight_end']}, providing the best time for deep sky observations.")
if "sky_objects" in data:
objects = data["sky_objects"]["objects"]
visible_planets = [obj for obj in objects if obj["type"] == "planet"]
if visible_planets:
planet_names = [obj["name"] for obj in visible_planets]
response_parts.append(f"Currently visible planets include: {', '.join(planet_names)}.")
return " ".join(response_parts)
def _generate_suggestions(self, data: Dict[str, Any]) -> List[str]:
"""Generate follow-up suggestions based on available data."""
suggestions = []
if "weather" in data and data["weather"]["observing_condition"] in ["good", "excellent"]:
suggestions.append("What are the best deep sky objects to observe tonight?")
suggestions.append("When is the optimal time for planetary observation?")
if "sky_objects" in data:
objects = data["sky_objects"]["objects"]
planets = [obj for obj in objects if obj["type"] == "planet"]
if planets:
suggestions.append(f"Tell me more about {planets[0]['name']}")
if "sun_moon" in data:
suggestions.append("What's the moon phase and how does it affect observations?")
suggestions.append("When does astronomical twilight begin and end?")
suggestions.append("What equipment would you recommend for tonight's conditions?")
suggestions.append("Show me a star chart for my location")
return suggestions[:3] # Limit to 3 suggestions
The LLM agent framework provides intelligent coordination of all system components, analyzing user queries to determine appropriate actions, executing tools in the correct sequence, and synthesizing comprehensive responses that combine multiple data sources into coherent and useful information.
Complete Running Example
Here is a comprehensive implementation that demonstrates the full astronomy agent system in action:
#!/usr/bin/env python3
"""
Complete Astronomy Agent Implementation
A comprehensive LLM-based astronomy assistant that provides location detection,
weather analysis, astronomical calculations, and sky object identification.
This is a fully functional implementation that can be run immediately.
import os
import sys
import json
import math
import requests
import ephem
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List, Tuple
from dataclasses import dataclass
from enum import Enum
from abc import ABC, abstractmethod
# Configuration - Replace with your actual API keys
OPENWEATHER_API_KEY = os.getenv('OPENWEATHER_API_KEY', 'your_openweather_api_key_here')
OPENCAGE_API_KEY = os.getenv('OPENCAGE_API_KEY', 'your_opencage_api_key_here')
class ObservingCondition(Enum):
"""Enumeration of observing condition quality levels."""
EXCELLENT = "excellent"
GOOD = "good"
FAIR = "fair"
POOR = "poor"
IMPOSSIBLE = "impossible"
class ObjectType(Enum):
"""Classification of astronomical objects."""
PLANET = "planet"
MOON = "moon"
STAR = "star"
DEEP_SKY = "deep_sky"
SATELLITE = "satellite"
COMET = "comet"
ASTEROID = "asteroid"
@dataclass
class LocationData:
"""Comprehensive location information container."""
latitude: float
longitude: float
elevation: float
city: str
country: str
timezone: str
accuracy_radius: float
@dataclass
class WeatherData:
"""Comprehensive weather information for astronomical observations."""
temperature_celsius: float
humidity_percent: float
pressure_hpa: float
wind_speed_ms: float
wind_direction_degrees: float
cloud_cover_percent: float
visibility_km: float
dew_point_celsius: float
uv_index: float
condition_description: str
timestamp: datetime
seeing_arcsec: Optional[float] = None
transparency_percent: Optional[float] = None
observing_condition: Optional[ObservingCondition] = None
@dataclass
class CelestialObject:
"""Comprehensive information about a celestial object."""
name: str
object_type: ObjectType
right_ascension: float
declination: float
altitude: float
azimuth: float
magnitude: float
distance_au: Optional[float] = None
constellation: Optional[str] = None
rise_time: Optional[datetime] = None
set_time: Optional[datetime] = None
transit_time: Optional[datetime] = None
is_visible: bool = False
visibility_quality: str = "unknown"
description: Optional[str] = None
@dataclass
class SunMoonData:
"""Solar and lunar information for a specific date and location."""
sunrise: datetime
sunset: datetime
solar_noon: datetime
moonrise: Optional[datetime]
moonset: Optional[datetime]
moon_phase: float
moon_illumination: float
moon_age_days: float
civil_twilight_begin: datetime
civil_twilight_end: datetime
nautical_twilight_begin: datetime
nautical_twilight_end: datetime
astronomical_twilight_begin: datetime
astronomical_twilight_end: datetime
class LocationProvider(ABC):
"""Abstract base class for location determination strategies."""
@abstractmethod
def get_location(self, query: str = None) -> Optional[LocationData]:
"""Retrieve location information based on the provided query or method."""
pass
class IPLocationProvider(LocationProvider):
"""Determines location based on IP address geolocation."""
def get_location(self, query: str = None) -> Optional[LocationData]:
"""Retrieve location based on current IP address."""
try:
response = requests.get("http://ip-api.com/json/?fields=status,message,country,city,lat,lon,timezone,query", timeout=10)
response.raise_for_status()
data = response.json()
if data.get('status') == 'success':
return LocationData(
latitude=data['lat'],
longitude=data['lon'],
elevation=0.0,
city=data['city'],
country=data['country'],
timezone=data['timezone'],
accuracy_radius=10000.0
)
except Exception as e:
print(f"IP location detection failed: {e}")
return None
class GeocodeLocationProvider(LocationProvider):
"""Determines location based on address or place name geocoding."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.opencagedata.com/geocode/v1/json"
def get_location(self, query: str) -> Optional[LocationData]:
"""Geocode an address or place name to coordinates."""
if not query or self.api_key == 'your_opencage_api_key_here':
return None
try:
params = {
'q': query,
'key': self.api_key,
'limit': 1,
'no_annotations': 0
}
response = requests.get(self.base_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data['results']:
result = data['results'][0]
geometry = result['geometry']
components = result['components']
annotations = result.get('annotations', {})
return LocationData(
latitude=geometry['lat'],
longitude=geometry['lng'],
elevation=annotations.get('elevation', {}).get('apparent', 0.0),
city=components.get('city', components.get('town', components.get('village', ''))),
country=components.get('country', ''),
timezone=annotations.get('timezone', {}).get('name', 'UTC'),
accuracy_radius=result.get('confidence', 1) * 1000
)
except Exception as e:
print(f"Geocoding failed: {e}")
return None
class CoordinateLocationProvider(LocationProvider):
"""Handles direct coordinate input with reverse geocoding for context."""
def __init__(self, api_key: str):
self.api_key = api_key
self.geocoder = GeocodeLocationProvider(api_key)
def get_location(self, query: str) -> Optional[LocationData]:
"""Parse coordinates and perform reverse geocoding for context."""
try:
coords = self._parse_coordinates(query)
if not coords:
return None
lat, lon = coords
if self.api_key != 'your_opencage_api_key_here':
reverse_query = f"{lat},{lon}"
location_data = self.geocoder.get_location(reverse_query)
if location_data:
location_data.latitude = lat
location_data.longitude = lon
location_data.accuracy_radius = 10.0
return location_data
return LocationData(
latitude=lat,
longitude=lon,
elevation=0.0,
city="Unknown",
country="Unknown",
timezone="UTC",
accuracy_radius=10.0
)
except Exception as e:
print(f"Coordinate parsing failed: {e}")
return None
def _parse_coordinates(self, coord_string: str) -> Optional[Tuple[float, float]]:
"""Parse various coordinate formats into decimal degrees."""
import re
coord_string = re.sub(r'\s+', ' ', coord_string.strip())
decimal_pattern = r'^(-?\d+\.?\d*),?\s*(-?\d+\.?\d*)$'
match = re.match(decimal_pattern, coord_string)
if match:
return float(match.group(1)), float(match.group(2))
dms_pattern = r'(\d+)[°\s]+(\d+)[\'′\s]*(\d*\.?\d*)[\"″\s]*([NSEW])\s*,?\s*(\d+)[°\s]+(\d+)[\'′\s]*(\d*\.?\d*)[\"″\s]*([NSEW])'
match = re.match(dms_pattern, coord_string.upper())
if match:
lat_deg, lat_min, lat_sec, lat_dir, lon_deg, lon_min, lon_sec, lon_dir = match.groups()
lat = float(lat_deg) + float(lat_min)/60 + float(lat_sec or 0)/3600
if lat_dir in ['S']:
lat = -lat
lon = float(lon_deg) + float(lon_min)/60 + float(lon_sec or 0)/3600
if lon_dir in ['W']:
lon = -lon
return lat, lon
return None
class LocationManager:
"""Manages multiple location providers and determines the best location data."""
def __init__(self, geocoding_api_key: str):
self.providers = {
'ip': IPLocationProvider(),
'geocode': GeocodeLocationProvider(geocoding_api_key),
'coordinates': CoordinateLocationProvider(geocoding_api_key)
}
self.cached_location = None
def get_location(self, user_input: str = None) -> Optional[LocationData]:
"""Determine location using the most appropriate method."""
if user_input:
if self._looks_like_coordinates(user_input):
location = self.providers['coordinates'].get_location(user_input)
if location:
self.cached_location = location
return location
location = self.providers['geocode'].get_location(user_input)
if location:
self.cached_location = location
return location
if not self.cached_location:
self.cached_location = self.providers['ip'].get_location()
return self.cached_location
def _looks_like_coordinates(self, text: str) -> bool:
"""Heuristic to determine if text contains coordinates."""
import re
patterns = [
r'-?\d+\.?\d*,\s*-?\d+\.?\d*',
r'\d+[°\s]+\d+[\'′\s]*\d*\.?\d*[\"″\s]*[NSEW]',
]
for pattern in patterns:
if re.search(pattern, text.upper()):
return True
return False
class WeatherProvider:
"""Integrates with OpenWeatherMap API for comprehensive weather data."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.openweathermap.org/data/2.5"
def get_current_weather(self, latitude: float, longitude: float) -> Optional[WeatherData]:
"""Retrieve current weather conditions for the specified location."""
if self.api_key == 'your_openweather_api_key_here':
return self._get_mock_weather()
try:
params = {
'lat': latitude,
'lon': longitude,
'appid': self.api_key,
'units': 'metric'
}
response = requests.get(f"{self.base_url}/weather", params=params, timeout=10)
response.raise_for_status()
data = response.json()
weather_data = WeatherData(
temperature_celsius=data['main']['temp'],
humidity_percent=data['main']['humidity'],
pressure_hpa=data['main']['pressure'],
wind_speed_ms=data.get('wind', {}).get('speed', 0),
wind_direction_degrees=data.get('wind', {}).get('deg', 0),
cloud_cover_percent=data.get('clouds', {}).get('all', 0),
visibility_km=data.get('visibility', 10000) / 1000,
dew_point_celsius=self._calculate_dew_point(
data['main']['temp'],
data['main']['humidity']
),
uv_index=0,
condition_description=data['weather'][0]['description'],
timestamp=datetime.fromtimestamp(data['dt'])
)
weather_data.seeing_arcsec = self._estimate_seeing(weather_data)
weather_data.transparency_percent = self._estimate_transparency(weather_data)
weather_data.observing_condition = self._assess_observing_conditions(weather_data)
return weather_data
except Exception as e:
print(f"Weather data retrieval failed: {e}")
return self._get_mock_weather()
def _get_mock_weather(self) -> WeatherData:
"""Provide mock weather data when API is not available."""
weather_data = WeatherData(
temperature_celsius=18.5,
humidity_percent=65,
pressure_hpa=1013.2,
wind_speed_ms=3.2,
wind_direction_degrees=225,
cloud_cover_percent=25,
visibility_km=15.0,
dew_point_celsius=12.1,
uv_index=0,
condition_description="partly cloudy",
timestamp=datetime.now()
)
weather_data.seeing_arcsec = self._estimate_seeing(weather_data)
weather_data.transparency_percent = self._estimate_transparency(weather_data)
weather_data.observing_condition = self._assess_observing_conditions(weather_data)
return weather_data
def _calculate_dew_point(self, temperature: float, humidity: float) -> float:
"""Calculate dew point using Magnus formula."""
a = 17.27
b = 237.7
alpha = ((a * temperature) / (b + temperature)) + math.log(humidity / 100.0)
return (b * alpha) / (a - alpha)
def _estimate_seeing(self, weather: WeatherData) -> float:
"""Estimate astronomical seeing based on weather conditions."""
base_seeing = 2.0
wind_factor = 1.0 + (weather.wind_speed_ms / 10.0) * 0.5
temp_factor = 1.0 + abs(weather.temperature_celsius - weather.dew_point_celsius) / 20.0
humidity_factor = 1.0 + (weather.humidity_percent - 50) / 100.0 * 0.3
estimated_seeing = base_seeing * wind_factor * temp_factor * humidity_factor
return min(max(estimated_seeing, 0.5), 10.0)
def _estimate_transparency(self, weather: WeatherData) -> float:
"""Estimate atmospheric transparency percentage."""
base_transparency = 85.0
cloud_impact = weather.cloud_cover_percent * 0.8
humidity_impact = max(0, weather.humidity_percent - 60) * 0.3
visibility_impact = max(0, (20 - weather.visibility_km) * 2)
transparency = base_transparency - cloud_impact - humidity_impact - visibility_impact
return max(min(transparency, 100.0), 0.0)
def _assess_observing_conditions(self, weather: WeatherData) -> ObservingCondition:
"""Assess overall observing conditions based on weather data."""
if weather.cloud_cover_percent > 80:
return ObservingCondition.IMPOSSIBLE
elif weather.cloud_cover_percent > 60:
return ObservingCondition.POOR
elif weather.cloud_cover_percent > 30:
return ObservingCondition.FAIR
elif weather.cloud_cover_percent > 10:
return ObservingCondition.GOOD
else:
return ObservingCondition.EXCELLENT
class AstronomicalCalculator:
"""Provides comprehensive astronomical calculations using PyEphem."""
def __init__(self):
self.observer = ephem.Observer()
self.planet_catalog = self._initialize_planet_catalog()
self.star_catalog = self._initialize_star_catalog()
def set_location(self, latitude: float, longitude: float, elevation: float = 0):
"""Configure observer location for calculations."""
self.observer.lat = str(latitude)
self.observer.lon = str(longitude)
self.observer.elevation = elevation
def set_time(self, observation_time: datetime = None):
"""Set observation time, defaults to current time."""
if observation_time is None:
observation_time = datetime.utcnow()
self.observer.date = observation_time
def get_sun_moon_data(self, date: datetime = None) -> SunMoonData:
"""Calculate comprehensive sun and moon data for the specified date."""
if date is None:
date = datetime.utcnow()
self.observer.date = date
sun = ephem.Sun()
moon = ephem.Moon()
try:
sunrise = self.observer.next_rising(sun).datetime()
sunset = self.observer.next_setting(sun).datetime()
solar_noon = self.observer.next_transit(sun).datetime()
except ephem.CircumpolarError:
sunrise = sunset = solar_noon = date
try:
moonrise = self.observer.next_rising(moon).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
moonrise = None
try:
moonset = self.observer.next_setting(moon).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
moonset = None
self.observer.horizon = '-6'
try:
civil_begin = self.observer.previous_rising(sun, use_center=True).datetime()
civil_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
civil_begin = civil_end = date
self.observer.horizon = '-12'
try:
nautical_begin = self.observer.previous_rising(sun, use_center=True).datetime()
nautical_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
nautical_begin = nautical_end = date
self.observer.horizon = '-18'
try:
astro_begin = self.observer.previous_rising(sun, use_center=True).datetime()
astro_end = self.observer.next_setting(sun, use_center=True).datetime()
except ephem.CircumpolarError:
astro_begin = astro_end = date
self.observer.horizon = '0'
moon.compute(self.observer)
moon_phase = moon.moon_phase
moon_illumination = moon.phase
previous_new_moon = ephem.previous_new_moon(date)
moon_age = (date - previous_new_moon.datetime()).days
return SunMoonData(
sunrise=sunrise,
sunset=sunset,
solar_noon=solar_noon,
moonrise=moonrise,
moonset=moonset,
moon_phase=moon_phase,
moon_illumination=moon_illumination,
moon_age_days=moon_age,
civil_twilight_begin=civil_begin,
civil_twilight_end=civil_end,
nautical_twilight_begin=nautical_begin,
nautical_twilight_end=nautical_end,
astronomical_twilight_begin=astro_begin,
astronomical_twilight_end=astro_end
)
def get_visible_planets(self) -> List[CelestialObject]:
"""Calculate positions and visibility of all planets."""
visible_planets = []
for planet_name, planet_obj in self.planet_catalog.items():
planet_obj.compute(self.observer)
ra_hours = float(planet_obj.ra) * 12 / math.pi
dec_degrees = float(planet_obj.dec) * 180 / math.pi
alt_degrees = float(planet_obj.alt) * 180 / math.pi
az_degrees = float(planet_obj.az) * 180 / math.pi
is_visible = alt_degrees > 0
visibility_quality = self._assess_visibility(alt_degrees, planet_obj.mag)
try:
rise_time = self.observer.next_rising(planet_obj).datetime()
set_time = self.observer.next_setting(planet_obj).datetime()
transit_time = self.observer.next_transit(planet_obj).datetime()
except (ephem.CircumpolarError, ephem.NeverUpError):
rise_time = set_time = transit_time = None
planet = CelestialObject(
name=planet_name,
object_type=ObjectType.PLANET,
right_ascension=ra_hours,
declination=dec_degrees,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=planet_obj.mag,
distance_au=planet_obj.earth_distance,
rise_time=rise_time,
set_time=set_time,
transit_time=transit_time,
is_visible=is_visible,
visibility_quality=visibility_quality,
description=self._get_planet_description(planet_name, planet_obj)
)
visible_planets.append(planet)
return visible_planets
def get_bright_stars(self, magnitude_limit: float = 3.0) -> List[CelestialObject]:
"""Get bright stars visible from current location."""
visible_stars = []
for star_name, star_data in self.star_catalog.items():
star = ephem.FixedBody()
star._ra = star_data['ra']
star._dec = star_data['dec']
star._epoch = '2000'
star.compute(self.observer)
if star.mag <= magnitude_limit:
alt_degrees = float(star.alt) * 180 / math.pi
az_degrees = float(star.az) * 180 / math.pi
is_visible = alt_degrees > 0
visibility_quality = self._assess_visibility(alt_degrees, star.mag)
star_obj = CelestialObject(
name=star_name,
object_type=ObjectType.STAR,
right_ascension=float(star.ra) * 12 / math.pi,
declination=float(star.dec) * 180 / math.pi,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=star.mag,
constellation=star_data.get('constellation'),
is_visible=is_visible,
visibility_quality=visibility_quality,
description=f"Bright star in constellation {star_data.get('constellation', 'Unknown')}"
)
visible_stars.append(star_obj)
return [star for star in visible_stars if star.is_visible]
def get_deep_sky_objects(self, magnitude_limit: float = 8.0) -> List[CelestialObject]:
"""Get visible deep sky objects from Messier catalog."""
messier_objects = {
'M31': {'ra': '00:42:44', 'dec': '+41:16:09', 'mag': 3.4, 'type': 'Galaxy', 'name': 'Andromeda Galaxy'},
'M42': {'ra': '05:35:17', 'dec': '-05:23:14', 'mag': 4.0, 'type': 'Nebula', 'name': 'Orion Nebula'},
'M45': {'ra': '03:47:29', 'dec': '+24:07:00', 'mag': 1.6, 'type': 'Star Cluster', 'name': 'Pleiades'},
'M13': {'ra': '16:41:41', 'dec': '+36:27:37', 'mag': 5.8, 'type': 'Globular Cluster', 'name': 'Hercules Cluster'},
'M57': {'ra': '18:53:35', 'dec': '+33:01:45', 'mag': 8.8, 'type': 'Planetary Nebula', 'name': 'Ring Nebula'},
'M1': {'ra': '05:34:32', 'dec': '+22:00:52', 'mag': 8.4, 'type': 'Supernova Remnant', 'name': 'Crab Nebula'},
'M51': {'ra': '13:29:53', 'dec': '+47:11:43', 'mag': 8.4, 'type': 'Galaxy', 'name': 'Whirlpool Galaxy'},
'M81': {'ra': '09:55:33', 'dec': '+69:03:55', 'mag': 6.9, 'type': 'Galaxy', 'name': 'Bode\'s Galaxy'},
'M104': {'ra': '12:39:59', 'dec': '-11:37:23', 'mag': 8.0, 'type': 'Galaxy', 'name': 'Sombrero Galaxy'},
'M27': {'ra': '19:59:36', 'dec': '+22:43:16', 'mag': 7.4, 'type': 'Planetary Nebula', 'name': 'Dumbbell Nebula'}
}
visible_objects = []
for obj_name, obj_data in messier_objects.items():
if obj_data['mag'] <= magnitude_limit:
obj = ephem.FixedBody()
obj._ra = obj_data['ra']
obj._dec = obj_data['dec']
obj._epoch = '2000'
obj.compute(self.observer)
alt_degrees = float(obj.alt) * 180 / math.pi
az_degrees = float(obj.az) * 180 / math.pi
is_visible = alt_degrees > 10
visibility_quality = self._assess_visibility(alt_degrees, obj_data['mag'])
if is_visible:
deep_sky_obj = CelestialObject(
name=obj_name,
object_type=ObjectType.DEEP_SKY,
right_ascension=float(obj.ra) * 12 / math.pi,
declination=float(obj.dec) * 180 / math.pi,
altitude=alt_degrees,
azimuth=az_degrees,
magnitude=obj_data['mag'],
is_visible=is_visible,
visibility_quality=visibility_quality,
description=f"{obj_data['name']} - {obj_data['type']}"
)
visible_objects.append(deep_sky_obj)
return visible_objects
def _initialize_planet_catalog(self) -> Dict[str, Any]:
"""Initialize catalog of planets for calculations."""
return {
'Mercury': ephem.Mercury(),
'Venus': ephem.Venus(),
'Mars': ephem.Mars(),
'Jupiter': ephem.Jupiter(),
'Saturn': ephem.Saturn(),
'Uranus': ephem.Uranus(),
'Neptune': ephem.Neptune()
}
def _initialize_star_catalog(self) -> Dict[str, Dict[str, Any]]:
"""Initialize catalog of bright stars."""
return {
'Sirius': {'ra': '06:45:08.9', 'dec': '-16:42:58', 'mag': -1.46, 'constellation': 'Canis Major'},
'Canopus': {'ra': '06:23:57.1', 'dec': '-52:41:44', 'mag': -0.74, 'constellation': 'Carina'},
'Arcturus': {'ra': '14:15:39.7', 'dec': '+19:10:56', 'mag': -0.05, 'constellation': 'Boötes'},
'Vega': {'ra': '18:36:56.3', 'dec': '+38:47:01', 'mag': 0.03, 'constellation': 'Lyra'},
'Capella': {'ra': '05:16:41.4', 'dec': '+45:59:53', 'mag': 0.08, 'constellation': 'Auriga'},
'Rigel': {'ra': '05:14:32.3', 'dec': '-08:12:06', 'mag': 0.13, 'constellation': 'Orion'},
'Procyon': {'ra': '07:39:18.1', 'dec': '+05:13:30', 'mag': 0.34, 'constellation': 'Canis Minor'},
'Betelgeuse': {'ra': '05:55:10.3', 'dec': '+07:24:25', 'mag': 0.50, 'constellation': 'Orion'},
'Aldebaran': {'ra': '04:35:55.2', 'dec': '+16:30:33', 'mag': 0.85, 'constellation': 'Taurus'},
'Antares': {'ra': '16:29:24.5', 'dec': '-26:25:55', 'mag': 1.09, 'constellation': 'Scorpius'},
'Spica': {'ra': '13:25:11.6', 'dec': '-11:09:41', 'mag': 1.04, 'constellation': 'Virgo'},
'Pollux': {'ra': '07:45:18.9', 'dec': '+28:01:34', 'mag': 1.14, 'constellation': 'Gemini'},
'Fomalhaut': {'ra': '22:57:39.0', 'dec': '-29:37:20', 'mag': 1.16, 'constellation': 'Piscis Austrinus'},
'Deneb': {'ra': '20:41:25.9', 'dec': '+45:16:49', 'mag': 1.25, 'constellation': 'Cygnus'},
'Regulus': {'ra': '10:08:22.3', 'dec': '+11:58:02', 'mag': 1.35, 'constellation': 'Leo'}
}
def _assess_visibility(self, altitude: float, magnitude: float) -> str:
"""Assess visibility quality based on altitude and magnitude."""
if altitude < 0:
return "not visible"
elif altitude < 10:
return "very low"
elif altitude < 30:
return "low"
elif altitude < 60:
return "good"
else:
return "excellent"
def _get_planet_description(self, planet_name: str, planet_obj: Any) -> str:
"""Generate descriptive information about a planet."""
descriptions = {
'Mercury': f"Closest planet to the Sun, currently {planet_obj.earth_distance:.2f} AU away",
'Venus': f"Brightest planet, currently {planet_obj.earth_distance:.2f} AU away",
'Mars': f"The Red Planet, currently {planet_obj.earth_distance:.2f} AU away",
'Jupiter': f"Largest planet, currently {planet_obj.earth_distance:.2f} AU away",
'Saturn': f"Ringed planet, currently {planet_obj.earth_distance:.2f} AU away",
'Uranus': f"Ice giant, currently {planet_obj.earth_distance:.2f} AU away",
'Neptune': f"Outermost planet, currently {planet_obj.earth_distance:.2f} AU away"
}
return descriptions.get(planet_name, f"Planet {planet_name}")
class AstronomicalWeatherAnalyzer:
"""Provides specialized analysis of weather data for astronomical purposes."""
@staticmethod
def calculate_limiting_magnitude(weather: WeatherData, light_pollution_class: str = "suburban") -> float:
"""Estimate limiting magnitude based on weather and light pollution."""
base_magnitudes = {
"rural": 6.5,
"suburban": 5.5,
"urban": 4.5,
"city": 3.5
}
base_mag = base_magnitudes.get(light_pollution_class, 5.5)
transparency_factor = weather.transparency_percent / 100.0
humidity_extinction = weather.humidity_percent / 100.0 * 0.5
estimated_magnitude = base_mag * transparency_factor - humidity_extinction
return max(min(estimated_magnitude, 7.0), 2.0)
@staticmethod
def recommend_observation_targets(weather: WeatherData) -> List[str]:
"""Recommend observation targets based on current conditions."""
recommendations = []
if weather.observing_condition == ObservingCondition.EXCELLENT:
recommendations.extend([
"Deep sky objects (galaxies, nebulae)",
"Double stars",
"Planetary details",
"Lunar surface features"
])
elif weather.observing_condition == ObservingCondition.GOOD:
recommendations.extend([
"Bright deep sky objects",
"Planets",
"Moon",
"Bright double stars"
])
elif weather.observing_condition == ObservingCondition.FAIR:
recommendations.extend([
"Planets",
"Moon",
"Bright stars"
])
elif weather.observing_condition == ObservingCondition.POOR:
recommendations.extend([
"Moon (if visible)",
"Bright planets"
])
return recommendations
class SimpleLLMClient:
"""Simple template-based response generator that mimics LLM behavior."""
def generate_response(self, prompt: str, context_data: Dict[str, Any]) -> str:
"""Generate a comprehensive response based on the prompt and context data."""
response_parts = []
# Extract location information
if 'location' in context_data:
loc = context_data['location']
response_parts.append(f"Based on your location in {loc['city']}, {loc['country']} "
f"(coordinates: {loc['latitude']:.4f}°, {loc['longitude']:.4f}°):")
# Weather analysis
if 'weather' in context_data:
weather = context_data['weather']
response_parts.append(f"\nCurrent weather conditions show {weather['condition_description']} "
f"with {weather['cloud_cover_percent']}% cloud cover and temperature of "
f"{weather['temperature_celsius']:.1f}°C. The observing conditions are "
f"{weather['observing_condition']} with estimated seeing of "
f"{weather['seeing_arcsec']:.1f} arcseconds and atmospheric transparency "
f"of {weather['transparency_percent']:.0f}%.")
if weather.get('recommendations'):
response_parts.append(f"\nRecommended observation targets for these conditions: "
f"{', '.join(weather['recommendations'])}.")
# Sun and moon information
if 'sun_moon' in context_data:
sun_moon = context_data['sun_moon']
response_parts.append(f"\nSun and Moon Information:")
response_parts.append(f"- Sunset: {sun_moon['sunset']}")
response_parts.append(f"- Astronomical twilight ends: {sun_moon['astronomical_twilight_end']}")
response_parts.append(f"- Moon: {sun_moon['moon_illumination']:.0f}% illuminated, "
f"{sun_moon['moon_age_days']:.1f} days old")
if sun_moon['moonrise'] != "No moonrise":
response_parts.append(f"- Moonrise: {sun_moon['moonrise']}")
if sun_moon['moonset'] != "No moonset":
response_parts.append(f"- Moonset: {sun_moon['moonset']}")
# Visible objects
if 'sky_objects' in context_data:
objects = context_data['sky_objects']['objects']
if objects:
response_parts.append(f"\nCurrently Visible Objects ({len(objects)} total):")
# Group objects by type
planets = [obj for obj in objects if obj['type'] == 'planet']
stars = [obj for obj in objects if obj['type'] == 'star']
deep_sky = [obj for obj in objects if obj['type'] == 'deep_sky']
if planets:
response_parts.append("\nPlanets:")
for planet in planets:
response_parts.append(f"- {planet['name']}: {planet['altitude']:.1f}° altitude, "
f"magnitude {planet['magnitude']:.1f}")
if planet['rise_time']:
response_parts.append(f" Rises: {planet['rise_time']}")
if planet['set_time']:
response_parts.append(f" Sets: {planet['set_time']}")
if planet['description']:
response_parts.append(f" {planet['description']}")
if stars:
response_parts.append(f"\nBright Stars ({len(stars)} visible):")
for star in stars[:5]: # Show top 5 stars
response_parts.append(f"- {star['name']} ({star['constellation']}): "
f"{star['altitude']:.1f}° altitude, magnitude {star['magnitude']:.1f}")
if deep_sky:
response_parts.append(f"\nDeep Sky Objects ({len(deep_sky)} visible):")
for obj in deep_sky:
response_parts.append(f"- {obj['name']}: {obj['altitude']:.1f}° altitude, "
f"magnitude {obj['magnitude']:.1f}")
if obj['description']:
response_parts.append(f" {obj['description']}")
# Add observing advice
if 'weather' in context_data and 'sun_moon' in context_data:
weather = context_data['weather']
sun_moon = context_data['sun_moon']
response_parts.append(f"\nObserving Advice:")
response_parts.append(f"The best time for deep sky observations begins after astronomical "
f"twilight ends at {sun_moon['astronomical_twilight_end']}.")
if weather['observing_condition'] in ['good', 'excellent']:
response_parts.append("Tonight's conditions are favorable for astronomical observations.")
elif weather['observing_condition'] == 'fair':
response_parts.append("Conditions are marginal - focus on brighter objects.")
else:
response_parts.append("Conditions are poor for observations tonight.")
moon_illumination = sun_moon['moon_illumination']
if moon_illumination > 75:
response_parts.append("The bright moon will wash out faint deep sky objects.")
elif moon_illumination < 25:
response_parts.append("The dark moon provides excellent conditions for faint objects.")
return " ".join(response_parts)
class AstronomyAgent:
"""Main astronomy agent that coordinates all components."""
def __init__(self):
self.location_manager = LocationManager(OPENCAGE_API_KEY)
self.weather_provider = WeatherProvider(OPENWEATHER_API_KEY)
self.calculator = AstronomicalCalculator()
self.llm_client = SimpleLLMClient()
self.current_location = None
self.current_weather = None
def process_query(self, user_query: str, user_location: str = None) -> str:
"""Process a user query and return a comprehensive response."""
print(f"Processing query: {user_query}")
# Step 1: Determine location
if user_location or not self.current_location:
print("Determining location...")
location = self.location_manager.get_location(user_location)
if location:
self.current_location = location
print(f"Location: {location.city}, {location.country} ({location.latitude:.4f}, {location.longitude:.4f})")
else:
return "I'm unable to determine your location. Please provide coordinates or a city name."
# Step 2: Get weather data
print("Retrieving weather data...")
weather = self.weather_provider.get_current_weather(
self.current_location.latitude,
self.current_location.longitude
)
if weather:
self.current_weather = weather
print(f"Weather: {weather.condition_description}, {weather.cloud_cover_percent}% clouds")
# Step 3: Configure astronomical calculator
self.calculator.set_location(
self.current_location.latitude,
self.current_location.longitude,
self.current_location.elevation
)
self.calculator.set_time()
# Step 4: Get sun/moon data
print("Calculating sun and moon data...")
sun_moon_data = self.calculator.get_sun_moon_data()
# Step 5: Get visible objects
print("Identifying visible celestial objects...")
planets = self.calculator.get_visible_planets()
stars = self.calculator.get_bright_stars(3.0)
deep_sky = self.calculator.get_deep_sky_objects(8.0)
# Combine all objects
all_objects = []
for planet in planets:
if planet.is_visible:
all_objects.append({
'name': planet.name,
'type': planet.object_type.value,
'altitude': planet.altitude,
'azimuth': planet.azimuth,
'magnitude': planet.magnitude,
'right_ascension': planet.right_ascension,
'declination': planet.declination,
'distance_au': planet.distance_au,
'rise_time': planet.rise_time.strftime("%H:%M") if planet.rise_time else None,
'set_time': planet.set_time.strftime("%H:%M") if planet.set_time else None,
'transit_time': planet.transit_time.strftime("%H:%M") if planet.transit_time else None,
'visibility_quality': planet.visibility_quality,
'description': planet.description
})
for star in stars:
all_objects.append({
'name': star.name,
'type': star.object_type.value,
'altitude': star.altitude,
'azimuth': star.azimuth,
'magnitude': star.magnitude,
'right_ascension': star.right_ascension,
'declination': star.declination,
'constellation': star.constellation,
'visibility_quality': star.visibility_quality,
'description': star.description
})
for obj in deep_sky:
all_objects.append({
'name': obj.name,
'type': obj.object_type.value,
'altitude': obj.altitude,
'azimuth': obj.azimuth,
'magnitude': obj.magnitude,
'right_ascension': obj.right_ascension,
'declination': obj.declination,
'visibility_quality': obj.visibility_quality,
'description': obj.description
})
# Step 6: Prepare context data
context_data = {
'location': {
'latitude': self.current_location.latitude,
'longitude': self.current_location.longitude,
'city': self.current_location.city,
'country': self.current_location.country
},
'weather': {
'temperature_celsius': weather.temperature_celsius,
'humidity_percent': weather.humidity_percent,
'cloud_cover_percent': weather.cloud_cover_percent,
'wind_speed_ms': weather.wind_speed_ms,
'visibility_km': weather.visibility_km,
'seeing_arcsec': weather.seeing_arcsec,
'transparency_percent': weather.transparency_percent,
'observing_condition': weather.observing_condition.value,
'condition_description': weather.condition_description,
'recommendations': AstronomicalWeatherAnalyzer.recommend_observation_targets(weather)
},
'sun_moon': {
'sunrise': sun_moon_data.sunrise.strftime("%H:%M"),
'sunset': sun_moon_data.sunset.strftime("%H:%M"),
'solar_noon': sun_moon_data.solar_noon.strftime("%H:%M"),
'moonrise': sun_moon_data.moonrise.strftime("%H:%M") if sun_moon_data.moonrise else "No moonrise",
'moonset': sun_moon_data.moonset.strftime("%H:%M") if sun_moon_data.moonset else "No moonset",
'moon_phase': sun_moon_data.moon_phase,
'moon_illumination': sun_moon_data.moon_illumination,
'moon_age_days': sun_moon_data.moon_age_days,
'astronomical_twilight_begin': sun_moon_data.astronomical_twilight_begin.strftime("%H:%M"),
'astronomical_twilight_end': sun_moon_data.astronomical_twilight_end.strftime("%H:%M")
},
'sky_objects': {
'objects': all_objects
}
}
# Step 7: Generate response
print("Generating response...")
response = self.llm_client.generate_response(user_query, context_data)
return response
def main():
"""Main function to demonstrate the astronomy agent."""
print("=== Astronomy Agent Demo ===")
print("This agent provides comprehensive astronomical information including:")
print("- Location detection")
print("- Weather conditions for observing")
print("- Sun and moon times")
print("- Visible celestial objects with coordinates")
print()
# Initialize the agent
agent = AstronomyAgent()
# Example queries
example_queries = [
"What can I observe tonight?",
"What are the current observing conditions?",
"When does astronomical twilight end tonight?",
"What planets are visible right now?",
"What deep sky objects can I see tonight?"
]
print("Example queries you can try:")
for i, query in enumerate(example_queries, 1):
print(f"{i}. {query}")
print()
# Interactive mode
while True:
try:
user_input = input("Enter your astronomy question (or 'quit' to exit): ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("Thank you for using the Astronomy Agent!")
break
if not user_input:
continue
# Check if user wants to specify location
location_input = None
if "location" in user_input.lower() or "coordinates" in user_input.lower():
location_input = input("Enter your location (city name or coordinates): ").strip()
print("\n" + "="*60)
response = agent.process_query(user_input, location_input)
print("\nResponse:")
print(response)
print("="*60 + "\n")
except KeyboardInterrupt:
print("\n\nThank you for using the Astronomy Agent!")
break
except Exception as e:
print(f"An error occurred: {e}")
print("Please try again with a different query.")
if __name__ == "__main__":
# Check for required dependencies
try:
import ephem
import requests
except ImportError as e:
print(f"Missing required dependency: {e}")
print("Please install required packages:")
print("pip install pyephem requests")
sys.exit(1)
# Run the main application
main()
This complete implementation provides a fully functional astronomy agent with the following capabilities:
Location Detection: The system can determine user location through IP geolocation, address geocoding, or direct coordinate input. It supports multiple coordinate formats and provides fallback mechanisms.
Weather Integration: Real-time weather data retrieval with specialized astronomical metrics including seeing estimates, transparency calculations, and observing condition assessments.
Astronomical Calculations: Comprehensive celestial mechanics calculations using PyEphem for planets, stars, and deep sky objects. Includes rise/set times, coordinates, and visibility predictions.
Sun and Moon Data: Precise calculations for sunrise, sunset, twilight times, moon phases, and lunar visibility information.
Intelligent Response Generation: Template-based response system that combines all data sources into coherent, informative responses tailored to astronomical observations.
Interactive Interface: Command-line interface that allows users to ask natural language questions about astronomical conditions and receive comprehensive answers.
The system handles error conditions gracefully, provides mock data when APIs are unavailable, and includes extensive documentation and comments throughout the codebase. Users can run this implementation immediately by installing the required dependencies (pyephem and requests) and optionally providing API keys for enhanced functionality.
To use the system with full functionality, users should obtain free API keys from OpenWeatherMap and OpenCage Data, then set them as environment variables or replace the placeholder values in the configuration section.
Advanced Features and Considerations
The astronomy agent system can be extended with numerous advanced features that enhance its utility and accuracy. These extensions demonstrate the flexibility of the modular architecture and the potential for sophisticated astronomical applications.
Machine Learning Integration: The system can incorporate machine learning models for improved weather prediction accuracy, seeing forecasting based on historical data, and personalized observation recommendations based on user preferences and equipment capabilities. Neural networks trained on meteorological data can provide more accurate predictions of observing conditions than simple heuristic methods.
Equipment Integration: Advanced implementations can interface with telescope control systems, camera equipment, and automated observatories. This allows the agent to not only recommend observations but also automatically configure equipment, capture images, and perform automated observation sequences based on current conditions.
Real-time Data Streams: Integration with space weather monitoring systems, satellite data feeds, and professional observatory networks can provide real-time updates on atmospheric conditions, solar activity, and other factors that affect astronomical observations.
Advanced Catalog Integration: The system can be extended to include comprehensive astronomical catalogs such as the complete NGC/IC catalogs, variable star databases, exoplanet catalogs, and real-time comet and asteroid orbital elements from the Minor Planet Center.
Multi-location Support: For users with multiple observing sites or those planning observing trips, the system can maintain profiles for different locations and provide comparative analysis of observing conditions across multiple sites.
Historical Analysis: By maintaining databases of past observations and conditions, the system can provide statistical analysis of observing quality, seasonal trends, and long-term climate patterns that affect astronomical observations.
Conclusion
The development of an LLM-based astronomy agentic AI represents a significant advancement in making astronomical knowledge and real-time observing information accessible to users of all experience levels. By combining the natural language understanding capabilities of modern LLMs with precise astronomical calculations and real-time environmental data, such systems create powerful tools that can significantly enhance the astronomical observing experience.
The modular architecture presented in this article demonstrates how complex astronomical applications can be built using clean, maintainable code that separates concerns and allows for easy extension and modification. The integration of multiple data sources including location services, weather APIs, and astronomical calculation libraries shows how modern software development practices can be applied to scientific applications.
The practical implementation provided offers a complete, functional system that can serve as a foundation for more sophisticated applications. The careful attention to error handling, fallback mechanisms, and user experience considerations ensures that the system remains useful even when external services are unavailable or when users provide incomplete information.
Future developments in this field will likely see increased integration with machine learning models for improved predictions, better integration with astronomical equipment and observatories, and more sophisticated natural language processing capabilities that can handle complex astronomical queries and provide detailed educational content.
The combination of artificial intelligence with astronomical science opens new possibilities for education, research, and amateur astronomy. As LLM capabilities continue to improve and astronomical databases become more comprehensive and accessible, systems like the one described in this article will become increasingly powerful tools for exploring and understanding the universe around us.
The success of such systems ultimately depends on their ability to make complex astronomical information accessible and actionable for users, whether they are beginning stargazers or experienced astronomers. By focusing on practical utility, scientific accuracy, and user experience, LLM-based astronomy agents can serve as valuable companions for anyone interested in exploring the night sky.
No comments:
Post a Comment