Implement POST /observability

This endpoint exposes a bit more complexity. It shows how the "work" of
the application resides in the domain layer, and how the service and
HTTP layers can cleanly use that functionality.

src/fastapibootcamp/domain/models.py

@@ -2,9 +2,13 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 
 from astroplan import Observer as AstroplanObserver
+from astropy.coordinates import AltAz, Angle, SkyCoord
+from astropy.time import Time
 
 # We're re-exporting Observer from the app's domain models to simulate the idea
 # that the Astroplan Observer is the app's domain model (i.e., this is a
@@ -29,3 +33,57 @@ def __init__(
         self.observer_id = observer_id
         self.aliases = list(aliases) if aliases else []
         self.local_timezone = local_timezone
+
+
+@dataclass(kw_only=True)
+class TargetObservability:
+    """The observability of a target for an observer."""
+
+    observer: Observer
+    target: SkyCoord
+    time: datetime
+    airmass: float | None
+    altaz: AltAz
+    is_above_horizon: bool
+    is_night: bool
+    moon_up: bool
+    moon_separation: Angle
+    moon_altaz: AltAz
+    moon_illumination: float
+
+    @classmethod
+    def compute(
+        cls,
+        observer: Observer,
+        target: SkyCoord,
+        time: datetime,
+    ) -> TargetObservability:
+        """Compute the observability of a target for an observer."""
+        astropy_time = Time(time)
+        is_up = observer.target_is_up(astropy_time, target)
+
+        altaz = target.transform_to(
+            AltAz(obstime=astropy_time, location=observer.location)
+        )
+        airmass = altaz.secz if is_up else None
+
+        is_night = observer.is_night(astropy_time)
+
+        moon_altaz = observer.moon_altaz(astropy_time)
+        moon_up = moon_altaz.alt > 0
+        moon_separation = altaz.separation(moon_altaz)
+        moon_illumination = observer.moon_illumination(astropy_time)
+
+        return cls(
+            observer=observer,
+            target=target,
+            time=time,
+            is_above_horizon=is_up,
+            airmass=airmass,
+            altaz=altaz,
+            is_night=is_night,
+            moon_up=moon_up,
+            moon_separation=moon_separation,
+            moon_altaz=moon_altaz,
+            moon_illumination=moon_illumination,
+        )

src/fastapibootcamp/handlers/astroplan/endpoints.py

@@ -11,7 +11,11 @@
 )
 from fastapibootcamp.exceptions import ObserverNotFoundError
 
-from .models import ObserverModel
+from .models import (
+    ObservabilityResponseModel,
+    ObservationRequestModel,
+    ObserverModel,
+)
 
 astroplan_router = APIRouter()
 
@@ -97,3 +101,36 @@ async def get_observers(
         ObserverModel.from_domain(observer=observer, request=context.request)
         for observer in observers
     ]
+
+
+# This is a POST endpoint. A POST request lets the client send a JSON payload.
+# Often this is used to create a new resource, but in this case we're using it
+# to trigger a calculation that's too complex to be done in a query string.
+#
+# e.g. POST /astroplan/observers/rubin/observable
+
+
+@astroplan_router.post(
+    "/observers/{observer_id}/observable",
+    summary=(
+        "Check if a coordinate is observable for an observer at a given time."
+    ),
+    response_model=ObservabilityResponseModel,
+)
+async def post_observable(
+    observer_id: str,
+    request_data: ObservationRequestModel,
+    context: Annotated[RequestContext, Depends(context_dependency)],
+) -> ObservabilityResponseModel:
+    factory = context.factory
+    observer_service = factory.create_observer_service()
+
+    observability = await observer_service.get_target_observability(
+        observer_id=observer_id,
+        sky_coord=request_data.get_target(),
+        time=request_data.time,
+    )
+
+    return ObservabilityResponseModel.from_domain(
+        observability=observability, request=context.request
+    )

src/fastapibootcamp/handlers/astroplan/models.py

@@ -11,12 +11,22 @@
 
 from __future__ import annotations
 
+from datetime import datetime
+from typing import Self
+
+from astropy.coordinates import SkyCoord
 from fastapi import Request
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator, model_validator
+from safir.datetime import current_datetime
+from safir.pydantic import normalize_isodatetime
 
-from fastapibootcamp.domain.models import Observer
+from fastapibootcamp.domain.models import Observer, TargetObservability
 
-__all__ = ["ObserverModel"]
+__all__ = [
+    "ObserverModel",
+    "ObservationRequestModel",
+    "ObservabilityResponseModel",
+]
 
 
 class ObserverModel(BaseModel):
@@ -74,3 +84,122 @@ def from_domain(
                 )
             ),
         )
+
+
+class ObservationRequestModel(BaseModel):
+    """Model for an observation request."""
+
+    ra: str = Field(
+        ...,
+        description="Target right ascension (HHhMMmSSs).",
+        examples=["5h23m34.6s"],
+    )
+
+    dec: str = Field(
+        ...,
+        description="Target declination (DDdMMmSSm).",
+        examples=["-69d45m22s"],
+    )
+
+    time: datetime = Field(
+        description="Time of the observation. Defaults to now if unset.",
+        default_factory=current_datetime,
+    )
+
+    # This ensures that the time is always provided in UTC.
+
+    _normalize_time = field_validator("time", mode="before")(
+        normalize_isodatetime
+    )
+
+    @model_validator(mode="after")
+    def validate_coordinate(self) -> Self:
+        # Validate ra and dec by try to parse them into an astropy SkyCoord
+        # object. If they're invalid, this will raise a ValueError.
+        try:
+            self.get_target()
+        except Exception as e:
+            raise ValueError(
+                f"Invalid coordinates: ra={self.ra}, dec={self.dec}"
+            ) from e
+        return self
+
+    def get_target(self) -> SkyCoord:
+        """Get the target RA and Dec as an astropy SkyCoord."""
+        return SkyCoord(
+            self.ra,
+            self.dec,
+            frame="icrs",
+        )
+
+
+class ObservabilityResponseModel(BaseModel):
+    """Model for an observability response."""
+
+    ra: str = Field(
+        ...,
+        description="Target right ascension (HHhMMmSSs).",
+        examples=["5h23m34.6s"],
+    )
+
+    dec: str = Field(
+        ...,
+        description="Target declination (DDdMMmSSm).",
+        examples=["-69d45m22s"],
+    )
+
+    time: datetime = Field(
+        ...,
+        description="Time of the observation (UTC).",
+    )
+
+    is_night: bool = Field(
+        ...,
+        description="Whether it is night time at the time of the observation.",
+    )
+
+    above_horizon: bool = Field(
+        ..., description="Whether the target is above the horizon."
+    )
+
+    airmass: float | None = Field(
+        None, description="Airmass of the target (null if below horizon)."
+    )
+
+    alt: float = Field(..., description="Altitude of the target (degrees).")
+
+    az: float = Field(..., description="Azimuth of the target (degrees).")
+
+    moon_separation: float = Field(
+        ..., description="Separation from the moon (degrees)."
+    )
+
+    moon_illumination: float = Field(
+        ..., description="Illumination of the moon (fraction)."
+    )
+
+    observer_url: str = Field(..., description="URL to the observer resource.")
+
+    @classmethod
+    def from_domain(
+        cls, *, observability: TargetObservability, request: Request
+    ) -> ObservabilityResponseModel:
+        """Create a model from a TargetObservability domain object."""
+        return cls(
+            ra=observability.target.ra.to_string(unit="hourangle"),
+            dec=observability.target.dec.to_string(unit="degree"),
+            time=observability.time,
+            above_horizon=observability.is_above_horizon,
+            is_night=observability.is_night,
+            airmass=observability.airmass,
+            alt=observability.altaz.alt.deg,
+            az=observability.altaz.az.deg,
+            moon_separation=observability.moon_separation.deg,
+            moon_illumination=observability.moon_illumination,
+            observer_url=str(
+                request.url_for(
+                    "get_observer",
+                    observer_id=observability.observer.observer_id,
+                )
+            ),
+        )

src/fastapibootcamp/services/observerservice.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from datetime import datetime
+
+from astropy.coordinates import SkyCoord
 from structlog.stdlib import BoundLogger
 
-from ..domain.models import Observer
+from ..domain.models import Observer, TargetObservability
 from ..exceptions import ObserverNotFoundError
 from ..storage.observerstore import ObserverStore
 
@@ -54,3 +57,27 @@ async def get_observers(
         return await self._observer_store.get_observers(
             name_pattern=name_pattern
         )
+
+    async def get_target_observability(
+        self, observer_id: str, sky_coord: SkyCoord, time: datetime
+    ) -> TargetObservability:
+        """Get the observability of a target for an observer.
+
+        Parameters
+        ----------
+        observer_id
+            The ID of the observer.
+        sky_coord
+            The target's coordinates.
+        time
+            The time of observation.
+
+        Returns
+        -------
+        TargetObservability
+            The observability domain model based on observer, target, and time.
+        """
+        observer = await self.get_observer_by_id(observer_id)
+        return TargetObservability.compute(
+            observer=observer, target=sky_coord, time=time
+        )

tests/handlers/astroplan_test.py

@@ -41,11 +41,11 @@ async def test_get_observers_rubin(client: AsyncClient) -> None:
     data = response.json()
     assert len(data) == 2
     assert data[0]["name"] == "Rubin Observatory"
-    assert data[1]["name"] == "Rubin Observatory"
+    assert data[1]["name"] == "Rubin AuxTel"
 
 
 @pytest.mark.asyncio
-async def test_get_observers_with_alises(client: AsyncClient) -> None:
+async def test_get_observers_with_aliases(client: AsyncClient) -> None:
     """Test finding observing sites with LSST in their aliases."""
     response = await client.get(
         f"{config.path_prefix}/astroplan/observers?name=lsst"
@@ -54,4 +54,23 @@ async def test_get_observers_with_alises(client: AsyncClient) -> None:
     data = response.json()
     assert len(data) == 2
     assert data[0]["name"] == "Rubin Observatory"
-    assert data[1]["name"] == "Rubin Observatory"
+    assert data[1]["name"] == "Rubin AuxTel"
+
+
+@pytest.mark.asyncio
+async def test_rubin_lmc_observability(client: AsyncClient) -> None:
+    """Test ``POST /astroplan/observers/rubin/observable``."""
+    path = f"{config.path_prefix}/astroplan/observers/rubin/observable"
+    response = await client.post(
+        path,
+        json={
+            "ra": "05h23m34.6s",
+            "dec": "-69d45m22s",
+            "time": "2024-04-24T00:00:00Z",
+        },
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["is_night"] is True
+    assert data["above_horizon"] is True
+    assert data["observer_url"].endswith("/astroplan/observers/rubin")