get_constellation

Instructions

Implementation Reference

• src/functions/celestial/impl.py:123-145 (handler)

  The primary handler for the 'get_constellation' MCP tool. Decorated with @mcp.tool() for registration. Processes input parameters, calls helper function get_constellation_center in a thread, and returns formatted response. Function parameters and docstring define the input schema.

  @mcp.tool()
  async def get_constellation(
      constellation_name: str,
      lon: float,
      lat: float,
      time: str,
      time_zone: str
  ) -> Dict[str, Any]:
      """Get the position (altitude/azimuth) of the center of a constellation.
      
      Args:
          constellation_name: Name of constellation (e.g. "Orion", "Ursa Major")
          lon: Observer longitude in degrees
          lat: Observer latitude in degrees
          time: Observation time string "YYYY-MM-DD HH:MM:SS"
          time_zone: IANA timezone string
          
      Returns:
          Dict with keys "data", "_meta". "data" contains name, altitude, azimuth.
      """
      location, time_info = process_location_and_time(lon, lat, time, time_zone)
      result = await asyncio.to_thread(get_constellation_center, constellation_name, location, time_info)
      return format_response(result)

• src/celestial.py:211-258 (helper)

  Core helper function that calculates the altitude and azimuth of a constellation's center. Uses pre-loaded JSON data for centers, with fallbacks to specific stars or SIMBAD queries.

  def get_constellation_center(
      constellation_name: str,
      observer_location: EarthLocation,
      time: Union[Time, datetime]
  ) -> Dict[str, Any]:
      """
      Return the apparent Alt/Az of a constellation's representative center using local data.
      """
      # Convert local time to UTC if input is datetime
      if isinstance(time, datetime):
          if time.tzinfo is None:
              raise ValueError("Input datetime must be timezone-aware for local time.")
          time = Time(time.astimezone(pytz.UTC))
  
      centers = _load_constellation_centers()
      centers_map = {item["name"].lower(): item for item in centers}
      key = constellation_name.lower()
      if key in centers_map:
          ra = float(centers_map[key]["ra"])
          dec = float(centers_map[key]["dec"])
          center_coord = SkyCoord(ra=ra*u.deg, dec=dec*u.deg, frame='icrs')
      else:
          fallback = {
              "ursa major": "Alioth",
              "ursa minor": "Polaris",
              "cassiopeia": "Schedar",
              "southern cross": "Acrux",
              "crux": "Acrux",
              "orion": "Betelgeuse",
              "scorpius": "Antares",
              "leo": "Regulus",
              "gemini": "Pollux",
              "taurus": "Aldebaran",
              "canis major": "Sirius"
          }
          if key in fallback:
              center_coord = _resolve_simbad_object(fallback[key])
          else:
              center_coord = _resolve_simbad_object(constellation_name)
  
      altaz_frame = AltAz(obstime=time, location=observer_location)
      altaz = center_coord.transform_to(altaz_frame)
      
      return {
          "name": constellation_name,
          "altitude": float(altaz.alt.deg),
          "azimuth": float(altaz.az.deg)
      }

• src/celestial.py:281-291 (helper)

  Utility function to load precomputed constellation center coordinates from JSON file, with caching.

  def _load_constellation_centers():
      global CONSTELLATIONS_CACHE
      if CONSTELLATIONS_CACHE is not None:
          return CONSTELLATIONS_CACHE
      data_path = os.path.join(os.path.dirname(__file__), 'data/constellation_centers.json')
      try:
          with open(data_path, 'r') as f:
              CONSTELLATIONS_CACHE = json.load(f)
      except FileNotFoundError:
          CONSTELLATIONS_CACHE = []
      return CONSTELLATIONS_CACHE

• src/functions/celestial/impl.py:5-123 (registration)

  The @mcp.tool() decorator registers the get_constellation function as an MCP tool.

  from src.server_instance import mcp
  from src.celestial import celestial_pos, celestial_rise_set, calculate_moon_info, get_visible_planets, get_constellation_center, calculate_nightly_forecast
  from src.utils import process_location_and_time
  
  from src.response import format_response
  
  @mcp.tool()
  async def get_celestial_pos(
      celestial_object: str,
      lon: float,
      lat: float,
      time: str,
      time_zone: str
  ) -> Dict[str, Any]:
      """Calculate the altitude and azimuth angles of a celestial object.
  
      Args:
          celestial_object: Name of object (e.g. "sun", "moon", "andromeda")
          lon: Observer longitude in degrees
          lat: Observer latitude in degrees
          time: Observation time string "YYYY-MM-DD HH:MM:SS"
          time_zone: IANA timezone string
  
      Returns:
          Dict with keys "data", "_meta". "data" contains "altitude" and "azimuth" (degrees).
      """
      location, time_info = process_location_and_time(lon, lat, time, time_zone)
      # Run synchronous celestial calculations in a separate thread to avoid blocking the event loop
      alt, az = await asyncio.to_thread(celestial_pos, celestial_object, location, time_info)
      return format_response({
          "altitude": alt,
          "azimuth": az
      })
  
  @mcp.tool()
  async def get_celestial_rise_set(
      celestial_object: str,
      lon: float,
      lat: float,
      time: str,
      time_zone: str
  ) -> Dict[str, Any]:
      """Calculate the rise and set times of a celestial object.
  
      Args:
          celestial_object: Name of object (e.g. "sun", "moon", "andromeda")
          lon: Observer longitude in degrees
          lat: Observer latitude in degrees
          time: Date string "YYYY-MM-DD HH:MM:SS"
          time_zone: IANA timezone string
  
      Returns:
          Dict with keys "data", "_meta". "data" contains "rise_time" and "set_time".
      """
      location, time_info = process_location_and_time(lon, lat, time, time_zone)
      # Run synchronous celestial calculations in a separate thread
      rise_time, set_time = await asyncio.to_thread(celestial_rise_set, celestial_object, location, time_info)
      return format_response({
          "rise_time": rise_time.isoformat() if rise_time else None,
          "set_time": set_time.isoformat() if set_time else None
      })
  
  @mcp.tool()
  async def get_moon_info(
      time: str,
      time_zone: str
  ) -> Dict[str, Any]:
      """Get detailed information about the Moon's phase and position.
      
      Args:
          time: Date string "YYYY-MM-DD HH:MM:SS"
          time_zone: IANA timezone string
          
      Returns:
          Dict with keys "data", "_meta". "data" contains illumination, phase_name, age_days, etc.
      """
      try:
          # Try standard format first
          dt = datetime.strptime(time, "%Y-%m-%d %H:%M:%S")
      except ValueError:
          try:
              # Try ISO format
              dt = datetime.fromisoformat(time)
          except ValueError:
               raise ValueError(f"Time string '{time}' matches neither '%Y-%m-%d %H:%M:%S' nor ISO format.")
  
      if dt.tzinfo is None:
          tz = pytz.timezone(time_zone)
          dt = tz.localize(dt)
          
      result = await asyncio.to_thread(calculate_moon_info, dt)
      return format_response(result)
  
  @mcp.tool()
  async def get_visible_planets(
      lon: float,
      lat: float,
      time: str,
      time_zone: str
  ) -> Dict[str, Any]:
      """Get a list of solar system planets currently visible (above horizon).
      
      Args:
          lon: Observer longitude in degrees
          lat: Observer latitude in degrees
          time: Observation time string "YYYY-MM-DD HH:MM:SS"
          time_zone: IANA timezone string
          
      Returns:
          Dict with keys "data", "_meta". "data" is a list of planet dicts (name, altitude, azimuth).
      """
      location, time_info = process_location_and_time(lon, lat, time, time_zone)
      # Note: Function name collision with imported function 'get_visible_planets'
      # Use the imported function from src.celestial
      from src.celestial import get_visible_planets as calc_visible_planets
      planets = await asyncio.to_thread(calc_visible_planets, location, time_info)
      return format_response(planets)
  
  @mcp.tool()