Coverage for lunar_toolkit/shadowcam/sc.py: 81%
197 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 02:58 +0000
1"""KPLO ShadowCam: ODE file selection, download, SPICE and CSM cameras for ASP."""
3from __future__ import annotations
5import logging
6import re
7from collections.abc import Iterable, Sequence
8from concurrent.futures import ThreadPoolExecutor, as_completed
9from dataclasses import dataclass
10from pathlib import Path
11from typing import TYPE_CHECKING, Any
13from .. import asp
14from ..constants import DEFAULT_PROJECTION
15from ..downloader import download_file, url_filename
16from ..isis import isd_generate, map_project, spiceinit
17from ..tools import is_dry_run, reuse_existing, run
19if TYPE_CHECKING: # pragma: no cover
20 import pandas as pd
22 from ..ode import OdeClient
24log = logging.getLogger(__name__)
26# Observation M013178466S ships M013178466SE (raw, 8-bit, 3144 samples incl.
27# bias columns; ODE "RAW") and M013178466SC (calibrated, float, 3072 samples;
28# ODE "CAL"). Both are ISIS cubes with the Instrument group attached.
29SC_ID_RE = re.compile(r"^M\d{9}S([CE]?)$", re.IGNORECASE)
31PRODUCT_SUFFIX = {"raw": "E", "cal": "C"}
33# ASP 3.7 example (docs/examples/shadowcam.rst): plain ``spiceinit from=<cube>``.
34SPICEINIT_PARAMS: dict[str, Any] = {"shape": "system"}
36# ``cam_test`` ISIS-vs-CSM comparison from the ASP example.
37CAM_TEST_SAMPLE_RATE = 50
39_POINTING_RE = re.compile(rb"^\s*InstrumentPointing\s*=", re.MULTILINE)
40_HISTORY_RE = re.compile(
41 rb"^Object = History\s+Name\s*=\s*IsisCube\s+StartByte\s*=\s*(\d+)\s+Bytes\s*=\s*(\d+)", re.MULTILINE
42)
45@dataclass(frozen=True)
46class ScProduct:
47 """Cube and PDS4 label URLs of one ShadowCam product."""
49 product_id: str
50 cube_url: str
51 label_url: str | None = None
52 cube_kbytes: float | None = None
54 @property
55 def observation_id(self) -> str:
56 """``M013178466S`` for ``M013178466SE`` / ``M013178466SC``."""
57 return self.product_id[:-1]
59 @property
60 def is_calibrated(self) -> bool:
61 return self.product_id.endswith("C")
63 @property
64 def cube_name(self) -> str:
65 return f"{self.product_id}.cub"
68@dataclass
69class ScOutputs:
70 """Files written for one product (None for skipped steps)."""
72 product_id: str
73 cube: Path
74 camera: Path | None = None
75 cam_test: Path | None = None
76 geotiff: Path | None = None
79def is_sc_product_id(product_id: str) -> bool:
80 """True for a ShadowCam observation (``M…S``) or product (``M…SE`` / ``M…SC``) ID."""
81 return bool(SC_ID_RE.match(product_id.strip()))
84def product_id_for(source: str, product: str = "raw") -> str:
85 """Product ID for an observation or product ID.
87 Parameters
88 ----------
89 source : str
90 ``M013178466S`` (observation), ``M013178466SE`` or ``M013178466SC``.
91 product : str
92 ``"raw"`` (``…SE``, the cube in ASP's example) or ``"cal"`` (``…SC``);
93 only used for observation IDs.
94 """
95 m = SC_ID_RE.match(source.strip())
96 if not m:
97 raise ValueError(f"Not a ShadowCam ID: {source!r}")
98 pid = source.strip().upper()
99 if m.group(1):
100 return pid
101 try:
102 return pid + PRODUCT_SUFFIX[product.lower()]
103 except KeyError:
104 raise ValueError(f"Unknown ShadowCam product {product!r}: {sorted(PRODUCT_SUFFIX)}") from None
107def select_sc_files(files: Iterable[dict], product_id: str | None = None) -> ScProduct | None:
108 """Pick the cube and label of a ShadowCam product from its ODE ``files`` list.
110 Parameters
111 ----------
112 files : iterable of dict
113 ODE product-file dicts (``FileName``, ``Type``, ``URL``, ``KBytes``).
114 product_id : str, optional
115 Product ID; derived from the cube file name when omitted.
117 Returns
118 -------
119 ScProduct or None
120 None when the list has no ``Product`` ``.cub`` (mosaics, DTMs). The
121 map-projected COGs and smear maps (``Referenced``) are never picked.
122 """
123 cube: dict | None = None
124 label: dict | None = None
125 for f in files or []:
126 if str(f.get("Type", "")).lower() != "product" or not f.get("URL"):
127 continue
128 name = str(f.get("FileName") or url_filename(f["URL"])).upper()
129 if name.endswith(".CUB") and cube is None:
130 cube = f
131 elif name.endswith(".XML") and label is None: 131 ↛ 125line 131 didn't jump to line 125 because the condition on line 131 was always true
132 label = f
133 if cube is None:
134 return None
135 pid = (product_id or url_filename(cube["URL"]).split(".")[0]).upper()
136 try:
137 kbytes = float(cube["KBytes"]) if cube.get("KBytes") is not None else None
138 except (TypeError, ValueError):
139 kbytes = None
140 return ScProduct(pid, cube["URL"], label["URL"] if label else None, kbytes)
143def sc_products(df: pd.DataFrame) -> list[ScProduct]:
144 """ShadowCam products with a downloadable cube from an ODE search.
146 Rows that are not ``…SE`` / ``…SC`` products (mosaics, DTMs) or lack a
147 cube are skipped with a warning.
148 """
149 out: list[ScProduct] = []
150 pids = df["product_id"] if "product_id" in df else [None] * len(df)
151 for pid, files in zip(pids, df.get("files", [])):
152 if pid and not is_sc_product_id(pid):
153 log.warning("Skipping %s: not a ShadowCam RAW / CAL product", pid)
154 continue
155 product = select_sc_files(files, pid)
156 if product is None: 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true
157 log.warning("Skipping %s: no .cub product file", pid)
158 continue
159 out.append(product)
160 return out
163def lookup_products(
164 product_ids: Iterable[str],
165 client: OdeClient | None = None,
166 product: str = "raw",
167) -> pd.DataFrame:
168 """Query ODE by observation / product ID (see
169 [`product_id_for`][lunar_toolkit.shadowcam.sc.product_id_for],
170 [`OdeClient.lookup`][lunar_toolkit.ode.OdeClient.lookup])."""
171 from ..ode import OdeClient
173 ids = [product_id_for(p, product) for p in product_ids]
174 return (client or OdeClient()).lookup(ids, instrument_label="ShadowCam")
177def download_product(
178 product: ScProduct,
179 output_dir: Path,
180 force: bool = False,
181 show_progress: bool = True,
182) -> Path:
183 """Download one product's cube (and PDS4 label) into *output_dir*.
185 Existing files are reused unless *force*; interrupted downloads resume.
186 Returns the cube path.
187 """
188 output_dir = Path(output_dir)
189 output_dir.mkdir(parents=True, exist_ok=True)
190 if product.label_url: 190 ↛ 194line 190 didn't jump to line 194 because the condition on line 190 was always true
191 label = output_dir / url_filename(product.label_url)
192 if not reuse_existing(label, force):
193 download_file(product.label_url, label, show_progress=show_progress)
194 cube = output_dir / product.cube_name
195 if not reuse_existing(cube, force):
196 download_file(product.cube_url, cube, show_progress=show_progress)
197 return cube
200def has_spice(cube: Path) -> bool:
201 """True when ``spiceinit`` completed on *cube*.
203 A failed ``spiceinit`` can leave SPICE tables in the label, so the cube's
204 History must also record a ``spiceinit`` run.
205 """
206 try:
207 with open(cube, "rb") as f:
208 head = f.read(1 << 20)
209 end = head.find(b"\nEnd\n")
210 label = head if end < 0 else head[:end]
211 m = _HISTORY_RE.search(label)
212 if not (m and _POINTING_RE.search(label)):
213 return False
214 f.seek(int(m.group(1)) - 1)
215 return b"Object = spiceinit" in f.read(int(m.group(2)))
216 except OSError:
217 return False
220class ScProcessor:
221 """Download ShadowCam cubes and prepare them for ASP (SPICE + CSM camera).
223 Follows the ASP 3.7 ShadowCam example: ``spiceinit`` on the delivered cube
224 (in place), ``isd_generate -k <cube> <cube>`` for the CSM camera and,
225 optionally, ``cam_test`` and ``mapproject``. Needs ASP 3.7's custom ISIS /
226 ALE build and the KPLO kernels under ``$ISISDATA/kplo/kernels`` (rsync
227 from ``rsync.im-ldi.com::kplo``) plus ``$ISISDATA/base/kernels``.
229 Per product *output_dir* gets ``<id>.cub``, ``<id>.json`` (when *csm*),
230 ``<id>.cam_test.txt`` (when *validate*) and ``<id>.tif`` (when
231 *mapproject*). Steps whose output exists (for ``spiceinit``: a cube
232 whose History records it, see [`has_spice`][lunar_toolkit.shadowcam.sc.has_spice]) are skipped unless
233 *force*.
235 Parameters
236 ----------
237 output_dir : Path
238 Where cubes, labels and cameras are written.
239 product : str
240 ``"raw"`` (``…SE``, the cube in ASP's example) or ``"cal"`` (``…SC``,
241 radiometrically calibrated, bias columns removed) for observation IDs.
242 web : bool
243 Use the USGS SPICE web service (it serves the KPLO kernels) in
244 ``spiceinit``. ALE's ShadowCam driver only reads NAIF kernel files,
245 so ``isd_generate`` needs the local kernels either way.
246 csm : bool
247 Write a CSM camera with ``isd_generate``.
248 validate : bool
249 Run ``cam_test`` (ISIS vs CSM camera) and save its report.
250 mapproject : bool
251 Write a GeoTIFF: ASP ``mapproject`` onto *dem* when given, else
252 ISIS ``cam2map`` onto the ``spiceinit`` shape.
253 dem : Path, optional
254 DEM for ASP ``mapproject``.
255 resolution : float, optional
256 Map resolution in m/pixel (ASP ``--tr`` / ``cam2map`` resolution).
257 t_srs : str, optional
258 PROJ string for ASP ``mapproject`` (default: the DEM's projection).
259 projection : str
260 ISIS projection for ``mosrange`` without *dem*.
261 keep : bool
262 Keep the ISIS map-projected cube.
263 force : bool
264 Redo every processing step (downloads are always reused).
265 spiceinit_params : dict, optional
266 Overrides merged into `SPICEINIT_PARAMS`.
267 """
269 def __init__(
270 self,
271 output_dir: Path = Path("shadowcam"),
272 *,
273 product: str = "raw",
274 web: bool = False,
275 csm: bool = True,
276 validate: bool = False,
277 mapproject: bool = False,
278 dem: Path | None = None,
279 resolution: float | None = None,
280 t_srs: str | None = None,
281 projection: str = DEFAULT_PROJECTION,
282 keep: bool = False,
283 force: bool = False,
284 spiceinit_params: dict[str, Any] | None = None,
285 ) -> None:
286 if product.lower() not in PRODUCT_SUFFIX: 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 raise ValueError(f"Unknown ShadowCam product {product!r}: {sorted(PRODUCT_SUFFIX)}")
288 self.output_dir = Path(output_dir)
289 self.product = product.lower()
290 self.web = web
291 self.csm = csm
292 self.validate = validate
293 self.mapproject = mapproject
294 self.dem = Path(dem) if dem else None
295 self.resolution = resolution
296 self.t_srs = t_srs
297 self.projection = projection
298 self.keep = keep
299 self.force = force
300 self.spiceinit_params = {**SPICEINIT_PARAMS, **(spiceinit_params or {})}
302 def process(self, source: ScProduct | str | Path) -> ScOutputs:
303 """Download (if needed) and process one product.
305 *source* is a [`ScProduct`][lunar_toolkit.shadowcam.sc.ScProduct], a local ``.cub`` path or an
306 observation / product ID (looked up on ODE).
307 """
308 if isinstance(source, ScProduct):
309 cube = download_product(source, self.output_dir)
310 elif Path(source).suffix.lower() == ".cub":
311 cube = Path(source)
312 else:
313 products = sc_products(lookup_products([str(source)], product=self.product))
314 if not products:
315 raise ValueError(f"No ShadowCam product {source!r} on ODE")
316 cube = download_product(products[0], self.output_dir)
317 return self.process_cube(cube)
319 def process_batch(
320 self,
321 sources: Sequence[ScProduct | str | Path],
322 max_workers: int = 2,
323 ) -> list[ScOutputs | None]:
324 """Process several products in parallel (None for failures)."""
325 results: list[ScOutputs | None] = [None] * len(sources)
326 with ThreadPoolExecutor(max_workers=max_workers) as pool:
327 futures = {pool.submit(self.process, s): i for i, s in enumerate(sources)}
328 for future in as_completed(futures):
329 i = futures[future]
330 try:
331 results[i] = future.result()
332 except Exception as exc:
333 log.error("Failed %s: %s", _source_name(sources[i]), exc)
334 return results
336 def cube_path(self, product: ScProduct) -> Path:
337 """Where [`process`][lunar_toolkit.shadowcam.sc.ScProcessor.process] downloads *product*'s cube."""
338 return self.output_dir / product.cube_name
340 def process_cube(self, cube: Path) -> ScOutputs:
341 """Run the SPICE / CSM chain on a downloaded ShadowCam cube."""
342 cube = Path(cube)
343 pid = cube.name.split(".")[0].upper()
344 if not is_dry_run(): 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 self.output_dir.mkdir(parents=True, exist_ok=True)
346 out = ScOutputs(pid, cube)
348 if self.force or not has_spice(cube):
349 spiceinit(cube, web=self.web, **self.spiceinit_params)
350 else:
351 log.info("Reusing SPICE in %s", cube)
353 if self.csm: 353 ↛ 358line 353 didn't jump to line 358 because the condition on line 353 was always true
354 out.camera = self.output_dir / f"{pid}.json"
355 if not reuse_existing(out.camera, self.force):
356 isd_generate(cube, out.camera)
358 if self.validate and out.camera:
359 out.cam_test = self.output_dir / f"{pid}.cam_test.txt"
360 if not reuse_existing(out.cam_test, self.force): 360 ↛ 363line 360 didn't jump to line 363 because the condition on line 360 was always true
361 self._cam_test(cube, out.camera, out.cam_test)
363 if self.mapproject:
364 out.geotiff = self.output_dir / f"{pid}.tif"
365 if not reuse_existing(out.geotiff, self.force): 365 ↛ 367line 365 didn't jump to line 367 because the condition on line 365 was always true
366 self._mapproject(cube, out.camera, out.geotiff)
367 return out
369 def _cam_test(self, cube: Path, camera: Path, report: Path) -> None:
370 args = ["--image", cube, "--cam1", cube, "--cam2", camera]
371 args += ["--session1", "isis", "--session2", "csm", "--sample-rate", CAM_TEST_SAMPLE_RATE]
372 res = run("cam_test", args, capture=True)
373 if not is_dry_run(): 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true
374 report.write_text(res.stdout or "")
376 def _mapproject(self, cube: Path, camera: Path | None, tif: Path) -> None:
377 if self.dem:
378 asp.mapproject(self.dem, cube, tif, camera, tr=self.resolution, t_srs=self.t_srs)
379 return
380 map_cube = tif.with_suffix(".map.cub")
381 map_project(cube, map_cube, self.projection, self.resolution, keep=self.keep)
382 run("gdal_translate", ["-of", "GTiff", "-co", "COMPRESS=DEFLATE", "-co", "TILED=YES", map_cube, tif])
383 if not self.keep and not is_dry_run(): 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true
384 map_cube.unlink(missing_ok=True)
387def import_sc(cube: Path, output_dir: Path | None = None, **kwargs: Any) -> ScOutputs:
388 """Process one downloaded ShadowCam cube; see [`ScProcessor`][lunar_toolkit.shadowcam.sc.ScProcessor] for
389 *kwargs*."""
390 cube = Path(cube)
391 return ScProcessor(output_dir or cube.parent, **kwargs).process_cube(cube)
394def _source_name(source: Any) -> str:
395 return source.product_id if isinstance(source, ScProduct) else str(source)