|
1 | | -from fastapi import Request, Body |
| 1 | +from fastapi import Request, Body, Query |
2 | 2 | from typing import Any, Iterable, Union, List |
3 | 3 | import httpx |
4 | 4 | import dotenv |
|
21 | 21 | from api.models.field_site import FieldSite, Location, Photo, PlanarOrientation, Observation, BeddingFacing |
22 | 22 | from typing import Optional |
23 | 23 |
|
24 | | -interchange_router = APIRouter( |
25 | | - prefix="/interchange-data", |
26 | | - tags=["interchange-data"], |
| 24 | +convert_router = APIRouter( |
| 25 | + prefix="/convert", |
| 26 | + tags=["convert"], |
27 | 27 | responses={404: {"description": "Not found"}}, |
28 | 28 | ) |
29 | 29 |
|
@@ -154,11 +154,10 @@ def spot_to_fieldsite(feat) -> FieldSite: |
154 | 154 | observations=observations, |
155 | 155 | ) |
156 | 156 |
|
157 | | -@interchange_router.post("/spot-to-fieldsite") |
158 | 157 | def multiple_spot_to_fieldsite(feat: Union[dict, List[dict]] = Body(...)) -> List[FieldSite]: |
159 | 158 | """ |
160 | 159 | Accept a single FeatureCollection or a list of FeatureCollections and |
161 | | - return a FieldSite for each qualifying Point feature (non-image-basemap). |
| 160 | + return a FieldSite for each qualifying Point feature. |
162 | 161 | """ |
163 | 162 | out: list[FieldSite] = [] |
164 | 163 |
|
@@ -203,25 +202,47 @@ def fieldsite_to_checkin(fs: FieldSite) -> dict: |
203 | 202 | return d |
204 | 203 |
|
205 | 204 |
|
206 | | -@interchange_router.post("/fieldsite-to-checkin") |
207 | 205 | def multiple_fieldsite_to_checkin(fieldsites: list[FieldSite] = Body(...)) -> list[dict]: |
208 | 206 | out: list[dict] = [] |
209 | 207 | for fs in fieldsites: |
210 | 208 | try: |
| 209 | + if not isinstance(fs, FieldSite): |
| 210 | + fs = FieldSite(**fs) |
211 | 211 | out.append(fieldsite_to_checkin(fs)) |
212 | 212 | except Exception: |
213 | 213 | continue |
214 | 214 | return out |
215 | 215 |
|
216 | | -@interchange_router.post("/spot-to-checkin") |
217 | | -async def spot_to_checkin(request: Request, spot: Union[dict, List[dict]] = Body(...)) -> list[dict]: |
| 216 | +def spot_to_checkin(spot: Union[dict, List[dict]] = Body(...)) -> list[dict]: |
218 | 217 | """Pipeline: Spot JSON (FeatureCollection[s]) or FieldSite list -> Checkin list.""" |
219 | 218 | # If it's already a list of FieldSite-like dicts (has 'location'), skip the first hop |
220 | 219 | if isinstance(spot, list) and spot and isinstance(spot[0], dict) and "location" in spot[0]: |
221 | 220 | fieldsites: List[FieldSite] = spot # already FieldSite-shaped |
222 | 221 | else: |
223 | 222 | # Convert FeatureCollection (or list of them) -> FieldSite list |
224 | 223 | fieldsites = multiple_spot_to_fieldsite(spot) |
225 | | - |
226 | 224 | # Convert FieldSite list -> Checkin list |
227 | 225 | return multiple_fieldsite_to_checkin(fieldsites) |
| 226 | + |
| 227 | + |
| 228 | + |
| 229 | +@convert_router.post("/field-site") |
| 230 | +async def convert_field_site( |
| 231 | + payload: Union[dict, List[dict]] = Body(...), |
| 232 | + in_: str = Query(..., alias="in"), |
| 233 | + out: str = Query(..., alias="out"), |
| 234 | +) -> Any: |
| 235 | + """ |
| 236 | + Unified converter: |
| 237 | + - ?in=spot&out=fieldsite -> spot FeatureCollection(s) -> FieldSite |
| 238 | + - ?in=fieldsite&out=checkin -> FieldSite -> Checkin |
| 239 | + - ?in=spot&out=checkin -> spot FeatureCollection(s) -> FieldSite -> Checkin |
| 240 | + """ |
| 241 | + key = (in_.lower(), out.lower()) |
| 242 | + if key == ("spot", "fieldsite"): |
| 243 | + return multiple_spot_to_fieldsite(payload) |
| 244 | + if key == ("fieldsite", "checkin"): |
| 245 | + return multiple_fieldsite_to_checkin(payload) |
| 246 | + if key == ("spot", "checkin"): |
| 247 | + return spot_to_checkin(payload) |
| 248 | + raise HTTPException(status_code=400, detail="Unsupported conversion. Use in=[spot|fieldsite], out=[fieldsite|checkin].") |
0 commit comments