Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from pydantic import Field

from hgs2hpc import hgs2hpc, hgs2hpc_batch
from normalizer import normalize_hpc, gse_frame, jsonify_skycoord
from normalizer import normalize_hpc, normalize_hpc_batch, gse_frame, jsonify_skycoord
from ephemeris import get_position
from validation import AstropyTime, HvBaseModel

Expand Down Expand Up @@ -94,6 +94,29 @@ def _normalize_hpc(params: Annotated[NormalizeHpcQueryParameters, Query()]):
return {"x": coord.Tx.value, "y": coord.Ty.value}


class HpcCoordInput(HvBaseModel):
x: float
y: float
coord_time: AstropyTime


class HpcBatchInput(HvBaseModel):
coordinates: List[HpcCoordInput]
target: AstropyTime


@app.post("/hpc", summary="Batch normalize HPC coordinates for Helioviewer POV")
def _normalize_hpc_post(params: HpcBatchInput):
"Normalize multiple HPC coordinates to Helioviewer's POV at the given target time"
coords_input = [
{"x": c.x, "y": c.y, "coord_time": c.coord_time} for c in params.coordinates
]

results = normalize_hpc_batch(coords_input, params.target)

return {"coordinates": results}


class GSECoordInput(HvBaseModel):
x: float
y: float
Expand Down
37 changes: 37 additions & 0 deletions app/normalizer.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from typing import List, Dict
from astropy.coordinates import SkyCoord
from astropy.time import Time
import astropy.units as u
Expand Down Expand Up @@ -36,6 +37,42 @@ def normalize_hpc(x: float, y: float, coord_time: Time, target: Time) -> SkyCoor
return solar_rotate_coordinate(real_coord, hv_frame.observer)


def normalize_hpc_batch(coordinates: List[Dict], target: Time) -> List[Dict]:
"""
Batch process multiple HPC coordinate normalizations

Parameters
----------
coordinates : List[Dict]
List of coordinate dictionaries with keys: x, y, coord_time
target : Time
Target observation time (same for all coordinates)

Returns
-------
List[Dict]
List of results with keys: x, y
"""
if not coordinates:
return []

hv_frame = get_helioviewer_frame(target)

with transform_with_sun_center():
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):
result = solar_rotate_coordinate(real_coord, hv_frame.observer)
return [{"x": c.Tx.value.item(), "y": c.Ty.value.item()} for c in result]


def skycoord_to_3dframe(coord: SkyCoord) -> SkyCoord:
"""
Accepts a SkyCoord and transforms it to
Expand Down
100 changes: 100 additions & 0 deletions app/test/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,106 @@ def test_hpc(client: TestClient):
assert 9 < coord["x"] and coord["x"] < 10


def test_hpc_post(client: TestClient):
"""
Test POST /hpc endpoint with batch coordinate normalization
"""
# Test valid batch request using RHESSI Flare 12070596
batch_data = {
"coordinates": [
{"x": 515, "y": -342, "coord_time": "2012-07-05T13:01:46"},
{"x": 100, "y": 200, "coord_time": "2012-07-05T13:02:00"},
],
"target": "2012-07-05T13:01:46",
}
response = client.post("/hpc", json=batch_data)
assert response.status_code == 200
data = response.json()
assert "coordinates" in data
assert len(data["coordinates"]) == 2
# First coordinate should match the known RHESSI flare result
assert pytest.approx(data["coordinates"][0]["x"], abs=0.1) == 523.6178
assert pytest.approx(data["coordinates"][0]["y"], abs=0.1) == -347.7228

# Test with differential rotation (target time different from coord_time)
batch_data = {
"coordinates": [
{"x": 0, "y": 0, "coord_time": "2012-01-01T00:00:00"},
],
"target": "2012-01-01T01:00:00",
}
response = client.post("/hpc", json=batch_data)
assert response.status_code == 200
data = response.json()
assert len(data["coordinates"]) == 1
# The x coordinate should move approximately 9 arcseconds
assert 9 < data["coordinates"][0]["x"] and data["coordinates"][0]["x"] < 10

# Test missing coordinates field
response = client.post("/hpc", json={"target": "2012-01-01T00:00:00Z"})
assert response.status_code == 422

# Test missing target field
response = client.post(
"/hpc",
json={"coordinates": [{"x": 0, "y": 0, "coord_time": "2012-01-01T00:00:00Z"}]},
)
assert response.status_code == 422

# Test invalid time format
batch_data = {
"coordinates": [{"x": 0, "y": 0, "coord_time": "NotAValidTime"}],
"target": "2012-01-01T00:00:00Z",
}
response = client.post("/hpc", json=batch_data)
assert response.status_code == 422

# Test empty coordinates array
batch_data = {"coordinates": [], "target": "2012-01-01T00:00:00Z"}
response = client.post("/hpc", json=batch_data)
assert response.status_code == 200
data = response.json()
assert len(data["coordinates"]) == 0

# Test multiple coordinates with different coord_times
batch_data = {
"coordinates": [
{"x": 100, "y": 200, "coord_time": "2024-01-01T00:00:00"},
{"x": -300, "y": 450, "coord_time": "2025-01-01T00:00:00"},
{"x": 500, "y": -100, "coord_time": "2022-01-01T00:00:00"},
],
"target": "2024-01-02T00:00:00",
}
response = client.post("/hpc", json=batch_data)
assert response.status_code == 200
data = response.json()
assert len(data["coordinates"]) == 3
# Verify all coordinates have x and y values
for coord in data["coordinates"]:
assert "x" in coord
assert "y" in coord
assert isinstance(coord["x"], (int, float))
assert isinstance(coord["y"], (int, float))

# Test batch results match individual GET results
batch_data = {
"coordinates": [
{"x": 515, "y": -342, "coord_time": "2012-07-05T13:01:46"},
],
"target": "2012-07-05T13:01:46",
}
batch_response = client.post("/hpc", json=batch_data)
individual_response = client.get(
"/hpc?x=515&y=-342&coord_time=2012-07-05T13:01:46&target=2012-07-05T13:01:46"
)
assert batch_response.status_code == 200
assert individual_response.status_code == 200
batch_coord = batch_response.json()["coordinates"][0]
individual_coord = individual_response.json()
assert pytest.approx(batch_coord["x"], abs=1e-10) == individual_coord["x"]
assert pytest.approx(batch_coord["y"], abs=1e-10) == individual_coord["y"]


def test_hgs2hpc(client: TestClient):
# Missing lat
response = client.get("/hgs2hpc?lon=0&coord_time=2012-01-01")
Expand Down