Merge pull request #7 from dgarciabriseno/test-normalizer

Patch normalizer to use earth as center of SphericalScreen

Author: dgarciabriseno

app/normalizer.py

@@ -29,11 +29,10 @@ def normalize_hpc(x: float, y: float, coord_time: Time, target: Time) -> SkyCoor
         Desired time for the new coordinates
     """
     with transform_with_sun_center():
-        real_coord = SkyCoord(
-            x * u.arcsecond, y * u.arcsecond, frame=get_earth_frame(coord_time)
-        )
+        earth_frame = get_earth_frame(coord_time)
         hv_frame = get_helioviewer_frame(target)
-        with SphericalScreen(hv_frame.observer, only_off_disk=True):
+        with SphericalScreen(earth_frame.observer, only_off_disk=True):
+            real_coord = SkyCoord(x * u.arcsecond, y * u.arcsecond, frame=earth_frame)
             return solar_rotate_coordinate(real_coord, hv_frame.observer)
 
 
@@ -62,13 +61,15 @@ def normalize_hpc_batch(coordinates: List[Dict], target: Time) -> List[Dict]:
         xs = [c["x"] for c in coordinates]
         ys = [c["y"] for c in coordinates]
         coord_times = [c["coord_time"] for c in coordinates]
-        real_coord = SkyCoord(
-            xs,
-            ys,
-            unit="arcsec,arcsec",
-            frame=get_earth_frame(coord_times),
-        )
-        with SphericalScreen(hv_frame.observer, only_off_disk=True):
+
+        earth_frame = get_earth_frame(coord_times)
+        with SphericalScreen(earth_frame.observer, only_off_disk=True):
+            real_coord = SkyCoord(
+                xs,
+                ys,
+                unit="arcsec,arcsec",
+                frame=earth_frame,
+            )
             result = solar_rotate_coordinate(real_coord, hv_frame.observer)
     return [{"x": c.Tx.value.item(), "y": c.Ty.value.item()} for c in result]
 

app/test/test_api.py

@@ -1,4 +1,6 @@
 import pytest
+import random
+import math
 from astropy import units as u
 from fastapi.testclient import TestClient
 from sunpy.coordinates import get_earth
@@ -147,6 +149,75 @@ def test_hpc_post(client: TestClient):
     assert pytest.approx(batch_coord["y"], abs=1e-10) == individual_coord["y"]
 
 
+def test_hpc_single_vs_batched(client: TestClient):
+    """
+    Test that processing 100 random coordinates one-by-one gives the same
+    results as processing them all at once with the batch endpoint.
+    """
+    # Generate 100 random coordinates
+    num_coords = 100
+    random.seed(42)  # For reproducibility
+
+    # Set a common target time
+    target = "2020-01-01T00:00:00"
+
+    coordinates = []
+    for _ in range(num_coords):
+        x = random.uniform(-1000, 1000)
+        y = random.uniform(-1000, 1000)
+        # Use coord_time within 24 hours before target time
+        hours_before = random.uniform(0, 24)
+        from datetime import datetime, timedelta
+
+        target_dt = datetime.fromisoformat(target)
+        coord_dt = target_dt - timedelta(hours=hours_before)
+        coord_time = coord_dt.strftime("%Y-%m-%dT%H:%M:%S")
+
+        coordinates.append({"x": x, "y": y, "coord_time": coord_time})
+
+    # Process each coordinate individually using GET
+    individual_results = []
+    for i, coord in enumerate(coordinates):
+        response = client.get(
+            f"/hpc?x={coord['x']}&y={coord['y']}&coord_time={coord['coord_time']}&target={target}"
+        )
+        assert response.status_code == 200
+        result = response.json()
+        # Check for NaN values
+        assert not math.isnan(
+            result["x"]
+        ), f"Individual request {i} returned NaN for x (input: {coord})"
+        assert not math.isnan(
+            result["y"]
+        ), f"Individual request {i} returned NaN for y (input: {coord})"
+        individual_results.append(result)
+
+    # Process all coordinates at once using POST
+    batch_data = {"coordinates": coordinates, "target": target}
+    batch_response = client.post("/hpc", json=batch_data)
+    assert batch_response.status_code == 200
+    batch_results = batch_response.json()["coordinates"]
+
+    # Compare results
+    assert len(individual_results) == len(batch_results)
+    for i, (individual, batched) in enumerate(zip(individual_results, batch_results)):
+        # Check batch results for NaN
+        assert not math.isnan(
+            batched["x"]
+        ), f"Batch result {i} returned NaN for x (input: {coordinates[i]})"
+        assert not math.isnan(
+            batched["y"]
+        ), f"Batch result {i} returned NaN for y (input: {coordinates[i]})"
+
+        # Compare individual vs batch results
+        assert (
+            pytest.approx(individual["x"], abs=1e-10) == batched["x"]
+        ), f"Mismatch at index {i}: x values differ"
+        assert (
+            pytest.approx(individual["y"], abs=1e-10) == batched["y"]
+        ), f"Mismatch at index {i}: y values differ"
+
+
 def test_hgs2hpc(client: TestClient):
     # Missing lat
     response = client.get("/hgs2hpc?lon=0&coord_time=2012-01-01")