Add ephemeris API via Sunpy/Horizons

server/api/ephemeris/__init__.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+from typing import Iterable
+
+from astropy.coordinates import SkyCoord
+
+from .horizons import Horizons
+from helios_exceptions import HeliosException
+
+# Use lower case for key names
+_providers = {
+    'horizons': Horizons
+}
+
+def Get(provider: str, body: str, dates: Iterable[datetime]) -> SkyCoord:
+    try:
+        # Use lower case so any variation of PrOvIdEr will execute the
+        # selected provider.
+        return _providers[provider.lower()].Get(body, dates)
+    except KeyError:
+        raise HeliosException(f"Unknown provider: {provider}")

server/api/ephemeris/horizons.py

@@ -0,0 +1,22 @@
+from datetime import datetime
+from typing import Iterable
+
+from astropy.coordinates import SkyCoord
+from sunpy.coordinates import get_horizons_coord
+
+from api.ephemeris.provider import CoordinateProvider
+
+class Horizons(CoordinateProvider):
+    @classmethod
+    def Query(cls, body: str, dates: Iterable[datetime]) -> list[SkyCoord]:
+        """
+        Queries JPL Horizons for coordinates and returns them
+
+        Parameters
+        ----------
+        body: `str`
+            Body to request coordinates for
+        dates: `list[datetime]`
+            List of dates. A coordinate will be returned for each date in the list
+        """
+        return get_horizons_coord(body, dates)

server/api/ephemeris/provider.py

@@ -0,0 +1,47 @@
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import Iterable
+
+from astropy.coordinates import SkyCoord
+
+from get_heeq import convert_skycoords_to_heeq
+
+class CoordinateProvider(ABC):
+    @classmethod
+    def Get(cls, body: str, dates: Iterable[datetime]) -> list[dict]:
+        """
+        Queries the underlying source for coordinates and transforms them to
+        Helios's coordinate system.
+
+        Parameters
+        ----------
+        body: `str`
+            The body to query coordinates for. Could be an observatory, planet, etc.
+        dates: `list[datetime]`
+            A list of datetime objects to query coordinates for.
+            One coordinate will be returned for each date.
+        """
+        coordinates = cls.Query(body, dates)
+        return CoordinateProvider.ConvertSkyCoords(coordinates)
+
+    @classmethod
+    def ConvertSkyCoords(cls, coords: Iterable[SkyCoord]) -> list[dict]:
+        """
+        Converts SkyCoords to Helios coordinates.
+        """
+        return list(map(lambda x: convert_skycoords_to_heeq(x), coords))
+
+    @classmethod
+    @abstractmethod
+    def Query(cls, body: str, dates: Iterable[datetime]) -> list[SkyCoord]:
+        """
+        Queries the underlying source for coordinates and returns them
+
+        Parameters
+        ----------
+        body: `str`
+            Body to request coordinates for
+        dates: `list[datetime]`
+            List of dates. A coordinate will be returned for each date in the list
+        """
+        raise NotImplementedError()

server/api/ephemeris/test/test_horizons.py

@@ -0,0 +1,21 @@
+from datetime import datetime
+
+import sunpy
+import pytest
+
+from api.ephemeris.horizons import Horizons
+from get_heeq import convert_skycoords_to_heeq
+
+def test_horizons_get():
+    test_dates = [datetime(2023, 1, 1), datetime(2022, 1, 1)]
+    # 399 is the unique ID for Earth's geocenter. "Earth" is ambiguous because
+    # it can be Earth's geocenter or the Earth-Moon barycenter and Horizons
+    # requests that you use the ID to select one.
+    coordinates = Horizons.Get("399", test_dates)
+    for idx, test_date in enumerate(test_dates):
+        sunpy_earth = convert_skycoords_to_heeq(sunpy.coordinates.get_earth(test_date))
+        # Assert that horizons and sunpy's coordinates for earth are approximately the same.
+        # This confirms the query is working as expected.
+        assert coordinates[idx]['x'] == pytest.approx(sunpy_earth['x'])
+        assert coordinates[idx]['y'] == pytest.approx(sunpy_earth['y'])
+        assert coordinates[idx]['z'] == pytest.approx(sunpy_earth['z'])

server/main.py

@@ -9,6 +9,7 @@
 from helios_exceptions import HeliosException
 from database import rest as database_endpoints
 from get_heeq import convert_skycoords_to_heeq
+import api.ephemeris as ephemeris
 
 logging.basicConfig(filename="helios_server.log", level=logging.DEBUG)
 
@@ -127,8 +128,13 @@ def event_position():
 def get_earth(date):
     return convert_skycoords_to_heeq(sunpy.coordinates.get_earth(date))
 
-
-
+@app.route("/ephemeris/<provider>/<body>")
+def get_positions(provider: str, body: str):
+    def fn():
+        date_inputs = request.args.getlist('date')
+        dates = map(lambda date: _parse_date(date), date_inputs)
+        return ephemeris.Get(provider, body, dates)
+    return _exec(lambda: fn())
 
 
 

server/test_routes.py

@@ -0,0 +1,28 @@
+import pytest
+
+from main import app
+
+from api.ephemeris.horizons import Horizons
+
+@pytest.fixture
+def client():
+    return app.test_client()
+
+def test_ephemeris_no_provider(client):
+    response = client.get("/ephemeris/shouldbreak/SDO")
+    assert response.status_code == 400
+    assert "Unknown provider" in response.json['error']
+
+def test_ephemeris_horizons(client, monkeypatch):
+    # Don't actually query horizons for this test.
+    def fake_horizons(*args, **kwargs) -> list[dict]:
+        return [{'x': 1, 'y': 2, 'z': 3}]
+    monkeypatch.setattr(Horizons, "Get", fake_horizons)
+
+    horizons_providers = ['Horizons', 'horizons']
+    for horizons in horizons_providers:
+        response = client.get(f"/ephemeris/{horizons}/SDO")
+        assert response.status_code == 200
+        assert response.json[0]['x'] == 1
+        assert response.json[0]['y'] == 2
+        assert response.json[0]['z'] == 3