Coverage for lunar_toolkit/ode.py: 77%
198 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 02:58 +0000
1"""PDS ODE REST client (ShadowCam, LROC NAC, Kaguya TC) returning the observation schema."""
3from __future__ import annotations
5import json
6import logging
7import time
8import urllib.error
9import urllib.parse
10import urllib.request
11from collections.abc import Iterable
12from pathlib import Path
13from typing import TYPE_CHECKING, Any
15from .constants import modernize_pds_url
16from .downloader import _USER_AGENT, PdsDownloader
17from .geometry import aoi_mask, aoi_region, load_aoi, query_region, region_bbox
18from .observation import OBSERVATION_COLUMNS, normalize_observations
20if TYPE_CHECKING: # pragma: no cover
21 import pandas as pd
23log = logging.getLogger(__name__)
25ODE_REST_URL = "https://oderest.rsl.wustl.edu/live2/"
27# Instrument registry: CLI/API key → ODE instrument host id, instrument id and
28# friendly product-type aliases (raw ODE ``pt`` values are accepted too).
29INSTRUMENTS: dict[str, dict[str, Any]] = {
30 "shadowcam": {
31 "label": "ShadowCam",
32 "ihid": "KPLO",
33 "iid": "ShadowCam",
34 "product_types": {"raw": "RAW", "cal": "CAL", "cmos": "CMOS", "umos": "UMOS", "dtm": "DTM"},
35 "default": "cal",
36 },
37 "lroc": {
38 "label": "LROC NAC",
39 "ihid": "LRO",
40 "iid": "LROC",
41 "product_types": {"edr": "EDRNAC4", "cdr": "CDRNAC4"},
42 "default": "edr",
43 },
44 "kaguya_tc": {
45 "label": "Kaguya TC",
46 "ihid": "SLN",
47 "iid": "TC",
48 "product_types": {
49 "stereo": "TCWL2B", # Stereo Level 2B0
50 "mono": "TCSL2B", # Mono Level 2B0
51 "dtm": "TCDTMO", # DTM ortho
52 "ortho": "TCORT", # Ortho map
53 "morning": "TCMORM",
54 "evening": "TCEVEM",
55 },
56 "default": "stereo",
57 },
58 "kaguya_mi": {
59 "label": "Kaguya MI",
60 "ihid": "SLN",
61 "iid": "MI",
62 "product_types": {
63 "vis": "VIS2B", # VIS Level 2B2 (kaguyami2isis)
64 "nir": "NIR2B", # NIR Level 2B2 (kaguyami2isis)
65 "vis2c": "VIS2C", # VIS Level 2C2
66 "nir2c": "NIR2C", # NIR Level 2C2
67 "l3c5": "MI3C", # Level 3C5 reflectance, map-projected
68 "map": "MIMAP", # Map tiles
69 },
70 "default": "vis",
71 },
72}
74_PRODUCT_QUERY = {"query": "product", "output": "JSON", "target": "moon"}
76_ALIASES = {
77 "sc": "shadowcam",
78 "kplo": "shadowcam",
79 "nac": "lroc",
80 "lroc_nac": "lroc",
81 "kaguya": "kaguya_tc",
82 "tc": "kaguya_tc",
83 "selene": "kaguya_tc",
84 "mi": "kaguya_mi",
85}
88def resolve_instrument(instrument: str, product_type: str | None = None) -> tuple[str, dict, str]:
89 """Return ``(key, registry_entry, ode_pt)`` for user-facing names."""
90 key = instrument.strip().lower().replace("-", "_")
91 key = _ALIASES.get(key, key)
92 if key not in INSTRUMENTS:
93 raise ValueError(f"Unknown instrument {instrument!r}. Valid: {sorted(INSTRUMENTS)}")
94 entry = INSTRUMENTS[key]
95 pt_key = (product_type or entry["default"]).strip()
96 pts = entry["product_types"]
97 if pt_key.lower() in pts: 97 ↛ 99line 97 didn't jump to line 99 because the condition on line 97 was always true
98 return key, entry, pts[pt_key.lower()]
99 if pt_key.upper() in {v.upper() for v in pts.values()} or product_type:
100 return key, entry, pt_key.upper()
101 raise ValueError(f"Unknown product type {product_type!r} for {key}: {sorted(pts)}")
104class OdeError(RuntimeError):
105 """ODE returned an error status."""
108class OdeClient:
109 """Query ODE REST and return `OBSERVATION_COLUMNS` DataFrames.
111 Parameters
112 ----------
113 base_url : str
114 ODE REST endpoint (default: ``live2``).
115 timeout : float
116 Per-request timeout in seconds — ODE footprint queries can take
117 30 s or more.
118 page_size : int
119 Products per request (ODE ``limit``); results are paged with ``offset``.
120 """
122 def __init__(
123 self,
124 base_url: str = ODE_REST_URL,
125 timeout: float = 180.0,
126 page_size: int = 100,
127 retries: int = 2,
128 ) -> None:
129 self.base_url = base_url
130 self.timeout = timeout
131 self.page_size = page_size
132 self.retries = retries
134 # Low-level request
136 def build_params(
137 self,
138 instrument: str,
139 product_type: str | None = None,
140 lat: tuple[float, float] | None = None,
141 lon: tuple[float, float] | None = None,
142 point: tuple[float, float] | None = None,
143 radius_km: float = 0.0,
144 incidence: tuple[float, float] | None = None,
145 emission: tuple[float, float] | None = None,
146 phase: tuple[float, float] | None = None,
147 time_range: tuple[str, str] | None = None,
148 loc: str | None = None,
149 results: str = "fmp",
150 aoi=None,
151 ) -> dict[str, str]:
152 """Translate search arguments into ODE query parameters.
154 An ``aoi`` (see [`lunar_toolkit.geometry.load_aoi`][lunar_toolkit.geometry.load_aoi]) is sent as
155 its bounding box; [`search`][lunar_toolkit.ode.OdeClient.search] refines the results exactly.
156 """
157 _, entry, pt = resolve_instrument(instrument, product_type)
158 params: dict[str, str] = {
159 **_PRODUCT_QUERY,
160 "results": results,
161 "ihid": entry["ihid"],
162 "iid": entry["iid"],
163 "pt": pt,
164 }
165 if aoi is not None or point is not None or lat is not None or lon is not None: 165 ↛ 182line 165 didn't jump to line 182 because the condition on line 165 was always true
166 if aoi is not None:
167 if point is not None or lat is not None or lon is not None:
168 raise ValueError("Pass either aoi or lat/lon/point, not both")
169 region = aoi_region(aoi)
170 else:
171 region = query_region(lat=lat, lon=lon, point=point, radius_km=radius_km)
172 west, south, east, north = region_bbox(region)
173 params.update(
174 {
175 "minlat": _fmt(south),
176 "maxlat": _fmt(north),
177 # ODE: positive east 0–360, boxes may cross the meridian
178 "westernlon": _fmt(west),
179 "easternlon": _fmt(east),
180 }
181 )
182 for (lo, hi), rng in (
183 (("mininangle", "maxinangle"), incidence),
184 (("minemangle", "maxemangle"), emission),
185 (("minphangle", "maxphangle"), phase),
186 ):
187 if rng is not None:
188 params[lo], params[hi] = _fmt(rng[0]), _fmt(rng[1])
189 if time_range is not None: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 params["minobtime"], params["maxobtime"] = str(time_range[0]), str(time_range[1])
191 if loc: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 params["loc"] = loc
193 return params
195 def request(self, params: dict[str, str]) -> dict:
196 """GET the ODE endpoint and return the decoded ``ODEResults`` dict."""
197 url = self.base_url + "?" + urllib.parse.urlencode(params)
198 req = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT})
199 last: Exception | None = None
200 for attempt in range(self.retries + 1):
201 try:
202 with urllib.request.urlopen(req, timeout=self.timeout) as resp:
203 data = json.loads(resp.read().decode("utf-8", "replace"))
204 break
205 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
206 last = exc
207 time.sleep(2 * (attempt + 1))
208 else:
209 raise OdeError(f"ODE request failed: {last}") from last
210 res = data.get("ODEResults", data)
211 if str(res.get("Status", "")).lower() == "error":
212 raise OdeError(res.get("Error") or json.dumps(res)[:500])
213 return res
215 # Search
217 def search(
218 self,
219 instrument: str,
220 product_type: str | None = None,
221 lat: tuple[float, float] | None = None,
222 lon: tuple[float, float] | None = None,
223 point: tuple[float, float] | None = None,
224 radius_km: float = 0.0,
225 incidence: tuple[float, float] | None = None,
226 emission: tuple[float, float] | None = None,
227 phase: tuple[float, float] | None = None,
228 time_range: tuple[str, str] | None = None,
229 max_results: int | None = 1000,
230 loc: str | None = None,
231 aoi=None,
232 ) -> pd.DataFrame:
233 """Search ODE and return the common observation schema.
235 Region arguments follow [`CumIndex.search`][lunar_toolkit.lroc.cumindex.CumIndex.search]:
236 ``lat=(min, max)`` and ``lon=(west, east)`` (``west > east`` crosses 0/360), or
237 ``point=(lat, lon)`` with ``radius_km`` (converted to a bounding box;
238 circles reaching a pole become a full-longitude cap). ODE matches the
239 box against product footprints (``loc=f``, its default). With
240 ``aoi`` (GeoJSON / WKT path, WKT string or
241 [`Aoi`][lunar_toolkit.geometry.Aoi]) the AOI's
242 bounding box is queried and each ``footprint_wkt`` is then
243 intersected with the polygon; paging continues until
244 ``max_results`` products survive that refinement.
246 Extra columns: ``files`` (list of ODE product-file dicts) and
247 ``ode_url`` (ODE product page).
248 """
249 key, entry, _ = resolve_instrument(instrument, product_type)
250 if aoi is not None:
251 aoi = load_aoi(aoi)
252 params = self.build_params(
253 instrument,
254 product_type,
255 lat=lat,
256 lon=lon,
257 point=point,
258 radius_km=radius_km,
259 incidence=incidence,
260 emission=emission,
261 phase=phase,
262 time_range=time_range,
263 loc=loc,
264 aoi=aoi,
265 )
266 records: list[dict] = []
267 offset = 0
268 while True:
269 limit = self.page_size
270 if max_results is not None: 270 ↛ 276line 270 didn't jump to line 276 because the condition on line 270 was always true
271 remaining = max_results - len(records)
272 if remaining <= 0: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 break
274 if aoi is None:
275 limit = min(limit, remaining)
276 res = self.request({**params, "limit": str(limit), "offset": str(offset)})
277 products = extract_products(res)
278 page = [parse_product(p, instrument_label=entry["label"]) for p in products]
279 if aoi is not None and page:
280 page = [r for r, keep in zip(page, aoi_mask(to_dataframe(page), aoi)) if keep]
281 records += page
282 if len(products) < limit:
283 break
284 offset += len(products)
285 if max_results is not None: 285 ↛ 287line 285 didn't jump to line 287 because the condition on line 285 was always true
286 records = records[:max_results]
287 return to_dataframe(records)
289 def lookup(self, product_ids: Iterable[str], instrument_label: str | None = None) -> pd.DataFrame:
290 """Observations for exact product IDs, one ODE ``productid`` request each.
292 Unknown IDs are skipped with a warning.
293 """
294 records: list[dict] = []
295 for pid in product_ids:
296 pid = pid.strip().upper()
297 products = extract_products(self.request({**_PRODUCT_QUERY, "results": "fmp", "productid": pid}))
298 if not products:
299 log.warning("ODE has no product %s", pid)
300 records += [parse_product(p, instrument_label=instrument_label) for p in products]
301 return to_dataframe(records)
303 def count(self, instrument: str, product_type: str | None = None, **kwargs) -> int:
304 """Number of matching products (``results=c``)."""
305 params = self.build_params(instrument, product_type, results="c", **kwargs)
306 res = self.request(params)
307 return int(res.get("Count", 0) or 0)
309 # Download
311 def download(
312 self,
313 df: pd.DataFrame,
314 output_dir: Path,
315 kind: str = "product",
316 match: str | None = None,
317 workers: int = 4,
318 force: bool = False,
319 ) -> dict[str, Path | None]:
320 """Download files listed in the ``files`` column of a search result.
322 Parameters
323 ----------
324 kind : str
325 ODE file ``Type`` to select: ``"product"`` (default; data files
326 only, labels skipped), ``"label"``, ``"browse"``, ``"referenced"``,
327 ``"derived"`` or ``"all"``.
328 match : str, optional
329 Case-insensitive substring the file name must contain
330 (e.g. ``"_cog.tif"``).
331 force : bool
332 Download again even when the file already exists in *output_dir*.
333 """
334 urls = select_file_urls(df, kind=kind, match=match)
335 if not urls: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 log.warning("No files selected for download.")
337 return {}
338 log.info("Downloading %d file(s) to %s", len(urls), output_dir)
339 return PdsDownloader(max_workers=workers).download_urls(urls, Path(output_dir), force=force)
342# Response parsing (pure functions — unit-tested against saved JSON)
345def extract_products(res: dict) -> list[dict]:
346 """Return the product list from an ``ODEResults`` dict (handles 0/1/N)."""
347 prods = res.get("Products")
348 if not isinstance(prods, dict): # "No Products Found" string or missing
349 return []
350 p = prods.get("Product")
351 if p is None: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 return []
353 return p if isinstance(p, list) else [p]
356def _files(prod: dict) -> list[dict]:
357 pf = (prod.get("Product_files") or {}).get("Product_file") or []
358 return pf if isinstance(pf, list) else [pf]
361def _num(v) -> float | None:
362 try:
363 return float(v)
364 except (TypeError, ValueError):
365 return None
368def parse_product(prod: dict, instrument_label: str | None = None) -> dict:
369 """Map one ODE product dict to the observation schema (+ extras)."""
370 pdsid = str(prod.get("pdsid", "")).strip()
371 product_id = pdsid.rsplit(".", 1)[-1].upper() if pdsid else None
372 files = _files(prod)
373 for f in files:
374 if f.get("URL"): 374 ↛ 373line 374 didn't jump to line 373 because the condition on line 374 was always true
375 f["URL"] = modernize_pds_url(f["URL"])
377 data_files = [f for f in files if str(f.get("Type", "")).lower() == "product"]
378 url = next(
379 (f["URL"] for f in data_files if not str(f.get("FileName", "")).upper().endswith((".XML", ".LBL"))),
380 None,
381 )
382 label_url = prod.get("LabelURL") or next(
383 (f["URL"] for f in data_files if str(f.get("FileName", "")).upper().endswith((".XML", ".LBL"))),
384 None,
385 )
386 if label_url: 386 ↛ 389line 386 didn't jump to line 389 because the condition on line 386 was always true
387 label_url = modernize_pds_url(label_url)
389 sub_solar_az = None
390 for k in ("Sub_solar_azimuth", "Subsolar_azimuth", "Solar_azimuth"):
391 if k in prod: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 sub_solar_az = _num(prod[k])
393 break
395 # ODE echoes the LROC CUMINDEX row in ODE_notes ("NOTE: Index Record:"
396 # followed by the CSV record). Its 83 columns follow CUMINDEX.LBL, so
397 # 0-based index = COLUMN_NUMBER - 1; 999.99-style N/A constants → None.
398 geom: dict[str, float | None] = {}
399 notes = (prod.get("ODE_notes") or {}).get("ODE_note") or []
400 notes = notes if isinstance(notes, list) else [notes]
401 for i, note in enumerate(notes[:-1]):
402 if str(note).startswith("NOTE: Index Record"):
403 import csv
404 import io
406 row = next(csv.reader(io.StringIO(str(notes[i + 1]))), [])
407 if len(row) == 83: 407 ↛ 418line 407 didn't jump to line 418 because the condition on line 407 was always true
408 for key, col in (
409 ("slew_angle", 9),
410 ("north_azimuth", 61),
411 ("sub_solar_azimuth", 62),
412 ("sub_spacecraft_lat", 65),
413 ("sub_spacecraft_lon", 66),
414 ("spacecraft_altitude", 79),
415 ):
416 v = _num(row[col].strip())
417 geom[key] = None if v is None or abs(v) >= 999.0 else v
418 break
419 if sub_solar_az is None: 419 ↛ 422line 419 didn't jump to line 422 because the condition on line 419 was always true
420 sub_solar_az = geom.get("sub_solar_azimuth")
422 return {
423 "product_id": product_id,
424 "instrument": instrument_label or f"{prod.get('ihid', '')} {prod.get('iid', '')}".strip(),
425 "product_type": prod.get("pt"),
426 "start_time": prod.get("UTC_start_time") or prod.get("Observation_time"),
427 "center_lat": _num(prod.get("Center_latitude")),
428 "center_lon": _num(prod.get("Center_longitude")),
429 "min_lat": _num(prod.get("Minimum_latitude")),
430 "max_lat": _num(prod.get("Maximum_latitude")),
431 "lon_west": _num(prod.get("Westernmost_longitude")),
432 "lon_east": _num(prod.get("Easternmost_longitude")),
433 "footprint_wkt": prod.get("Footprint_geometry") or prod.get("Footprint_C0_geometry"),
434 "incidence": _num(prod.get("Incidence_angle")),
435 "emission": _num(prod.get("Emission_angle")),
436 "phase": _num(prod.get("Phase_angle")),
437 "slew_angle": geom.get("slew_angle"),
438 "sub_solar_azimuth": sub_solar_az,
439 "north_azimuth": geom.get("north_azimuth"),
440 "sub_spacecraft_lat": geom.get("sub_spacecraft_lat"),
441 "sub_spacecraft_lon": geom.get("sub_spacecraft_lon"),
442 "spacecraft_altitude": geom.get("spacecraft_altitude"),
443 "pixel_scale": _num(prod.get("Map_resolution") or prod.get("Map_scale")),
444 "url": url,
445 "label_url": label_url,
446 "source": "ode",
447 "files": files,
448 "ode_url": prod.get("ProductURL"),
449 }
452def to_dataframe(records: list[dict]) -> pd.DataFrame:
453 """Observation DataFrame with the ``files`` / ``ode_url`` extras appended."""
454 df = normalize_observations(records)
455 df["files"] = [r.get("files", []) for r in records] if records else []
456 df["ode_url"] = [r.get("ode_url") for r in records] if records else []
457 return df[OBSERVATION_COLUMNS + ["files", "ode_url"]]
460def select_file_urls(df: pd.DataFrame, kind: str = "product", match: str | None = None) -> list[str]:
461 """Pick file URLs from the ``files`` column (see
462 [`OdeClient.download`][lunar_toolkit.ode.OdeClient.download])."""
463 kind = kind.lower()
464 urls: list[str] = []
465 for files in df.get("files", []):
466 for f in files or []:
467 name = str(f.get("FileName", ""))
468 ftype = str(f.get("Type", "")).lower()
469 is_label = name.upper().endswith((".XML", ".LBL"))
470 if kind == "label":
471 ok = ftype == "product" and is_label
472 elif kind == "product":
473 ok = ftype == "product" and not is_label
474 elif kind == "all": 474 ↛ 475line 474 didn't jump to line 475 because the condition on line 474 was never true
475 ok = True
476 else:
477 ok = ftype == kind
478 if ok and match and match.lower() not in name.lower():
479 ok = False
480 if ok and f.get("URL") and f["URL"] not in urls:
481 urls.append(f["URL"])
482 return urls
485def _fmt(x: float) -> str:
486 return f"{float(x):.6g}"