Coverage for lunar_toolkit/kaguya/mi.py: 71%
126 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 02:58 +0000
1"""Kaguya Multiband Imager Level 2B2: ODE file selection, download and ISIS import."""
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 ..constants import DEFAULT_PROJECTION
14from ..isis import isd_generate, isis, map_project, spiceinit
15from ..tools import is_dry_run, reuse_existing, run
16from .tc import TcProduct, download_product, image_pointer, select_tc_files
18if TYPE_CHECKING: # pragma: no cover
19 import pandas as pd
21 from ..ode import OdeClient
23log = logging.getLogger(__name__)
25# Level 2B2 MI product IDs, e.g. MVA_2B2_01_01935N340E0939 (V = VIS, N = NIR).
26MI_ID_RE = re.compile(r"^M[VN][A-Z]_2B2_\d{2}_\d{5}[NS]\d{3}E\d{4}$", re.IGNORECASE)
28# kaguyami2isis: keep every DN, as for TC.
29KAGUYAMI2ISIS_PARAMS: dict[str, Any] = {
30 "setnullrange": False,
31 "sethrsrange": False,
32 "sethisrange": False,
33 "setlrsrange": False,
34 "setlisrange": False,
35}
37# spiceinit: reconstructed CK / SPK, LOLA shape.
38SPICEINIT_PARAMS: dict[str, Any] = {
39 "ckrecon": True,
40 "ckpredicted": False,
41 "spkrecon": True,
42 "spkpredicted": False,
43 "shape": "system",
44}
47@dataclass(frozen=True)
48class MiProduct:
49 """Data and label URLs of one Level 2B2 MI product."""
51 product_id: str
52 image_url: str
53 label_url: str
54 image_kbytes: float | None = None
56 @property
57 def sensor(self) -> str:
58 """``"VIS"`` (5 bands) or ``"NIR"`` (4 bands)."""
59 return "VIS" if self.product_id[1] == "V" else "NIR"
62@dataclass
63class MiOutputs:
64 """Files written for one product (None for skipped steps)."""
66 product_id: str
67 cube: Path
68 camera: Path | None = None
69 map_cube: Path | None = None
70 geotiff: Path | None = None
73def is_mi_product_id(product_id: str) -> bool:
74 """True for a Level 2B2 MI product ID that ``kaguyami2isis`` can import."""
75 return bool(MI_ID_RE.match(product_id.strip()))
78def select_mi_files(files: Iterable[dict], product_id: str | None = None) -> MiProduct | None:
79 """Pick the data and label files of an MI product from its ODE ``files`` list.
81 MI products have the TC layout (detached ``.lbl`` + ``.img.gz``), so
82 this follows [`select_tc_files`][lunar_toolkit.kaguya.tc.select_tc_files].
84 Returns
85 -------
86 MiProduct or None
87 None when the list lacks a ``.img`` / ``.img.gz`` data file or a
88 ``.lbl`` label.
89 """
90 p = select_tc_files(files, product_id)
91 return None if p is None else MiProduct(p.product_id, p.image_url, p.label_url, p.image_kbytes)
94def mi_products(df: pd.DataFrame) -> list[MiProduct]:
95 """MI Level 2B2 products with downloadable data + label files from an ODE search.
97 Rows that are not Level 2B2 products (2C2, 3C5, maps) or lack either
98 file are skipped with a warning.
99 """
100 out: list[MiProduct] = []
101 pids = df["product_id"] if "product_id" in df else [None] * len(df)
102 for pid, files in zip(pids, df.get("files", [])):
103 if pid and not is_mi_product_id(pid):
104 log.warning("Skipping %s: not a Level 2B2 MI product", pid)
105 continue
106 product = select_mi_files(files, pid)
107 if product is None: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 log.warning("Skipping %s: no .img/.lbl product files", pid)
109 continue
110 out.append(product)
111 return out
114def lookup_mi_products(product_ids: Iterable[str], client: OdeClient | None = None) -> pd.DataFrame:
115 """Query ODE by product ID and return the observation DataFrame (see
116 [`OdeClient.lookup`][lunar_toolkit.ode.OdeClient.lookup])."""
117 from ..ode import OdeClient
119 return (client or OdeClient()).lookup(product_ids, instrument_label="Kaguya MI")
122def download_mi_product(
123 product: MiProduct,
124 output_dir: Path,
125 force: bool = False,
126 keep_gz: bool = False,
127 show_progress: bool = True,
128) -> Path:
129 """Download (and gunzip) one product's label and image; returns the label path.
131 See [`download_product`][lunar_toolkit.kaguya.tc.download_product].
132 """
133 tc = TcProduct(product.product_id, product.image_url, product.label_url, product.image_kbytes)
134 return download_product(tc, output_dir, force=force, keep_gz=keep_gz, show_progress=show_progress)
137class MiProcessor:
138 """Download and import Kaguya MI Level 2B2 products into ISIS.
140 Each product ends up in *output_dir* as ``<id>.cub`` (multi-band,
141 spiceinit'ed), ``<id>.json`` (CSM camera, when *csm*) and, when
142 *mapproject*, ``<id>.tif``. Steps whose output exists are skipped
143 unless *force*.
145 Parameters
146 ----------
147 output_dir : Path
148 Where labels, images, cubes and cameras are written.
149 web : bool
150 Use the USGS SPICE web service in ``spiceinit`` (no local Kaguya
151 kernels needed). Set False when ``$ISISDATA/kaguya`` is installed.
152 csm : bool, optional
153 Write a CSM camera with ``isd_generate``. ALE only has NAIF-SPICE
154 MI drivers, which need local kernels, so the default is ``not web``.
155 mapproject : bool
156 Also ``cam2map`` the cube and write a GeoTIFF.
157 projection : str
158 ISIS projection for ``mosrange`` (default: equirectangular).
159 resolution : float, optional
160 Map resolution in m/pixel (default: ``mosrange``'s estimate).
161 keep : bool
162 Keep the PDS ``.img`` and the map-projected cube after use.
163 force : bool
164 Redo every step even when its output exists.
165 kaguyami2isis_params, spiceinit_params : dict, optional
166 Overrides merged into `KAGUYAMI2ISIS_PARAMS` /
167 `SPICEINIT_PARAMS`.
168 """
170 def __init__(
171 self,
172 output_dir: Path = Path("kaguya"),
173 *,
174 web: bool = True,
175 csm: bool | None = None,
176 mapproject: bool = False,
177 projection: str = DEFAULT_PROJECTION,
178 resolution: float | None = None,
179 keep: bool = False,
180 force: bool = False,
181 kaguyami2isis_params: dict[str, Any] | None = None,
182 spiceinit_params: dict[str, Any] | None = None,
183 ) -> None:
184 self.output_dir = Path(output_dir)
185 self.web = web
186 self.csm = (not web) if csm is None else csm
187 self.mapproject = mapproject
188 self.projection = projection
189 self.resolution = resolution
190 self.keep = keep
191 self.force = force
192 self.kaguyami2isis_params = {**KAGUYAMI2ISIS_PARAMS, **(kaguyami2isis_params or {})}
193 self.spiceinit_params = {**SPICEINIT_PARAMS, **(spiceinit_params or {})}
195 def process(self, source: MiProduct | str | Path) -> MiOutputs:
196 """Download (if needed) and import one product.
198 *source* is a [`MiProduct`][lunar_toolkit.kaguya.mi.MiProduct], a local ``.lbl`` path or a product
199 ID (looked up on ODE).
200 """
201 if isinstance(source, MiProduct):
202 label = download_mi_product(source, self.output_dir, force=self.force)
203 elif Path(source).suffix.lower() == ".lbl":
204 label = Path(source)
205 else:
206 products = mi_products(lookup_mi_products([str(source)]))
207 if not products:
208 raise ValueError(f"No Kaguya MI Level 2B2 product {source!r} on ODE")
209 label = download_mi_product(products[0], self.output_dir, force=self.force)
210 return self.import_label(label)
212 def process_batch(
213 self,
214 sources: Sequence[MiProduct | str | Path],
215 max_workers: int = 2,
216 ) -> list[MiOutputs | None]:
217 """Process several products in parallel (None for failures)."""
218 results: list[MiOutputs | None] = [None] * len(sources)
219 with ThreadPoolExecutor(max_workers=max_workers) as pool:
220 futures = {pool.submit(self.process, s): i for i, s in enumerate(sources)}
221 for future in as_completed(futures):
222 i = futures[future]
223 try:
224 results[i] = future.result()
225 except Exception as exc:
226 src = sources[i]
227 log.error("Failed %s: %s", src.product_id if isinstance(src, MiProduct) else src, exc)
228 return results
230 def import_label(self, label: Path) -> MiOutputs:
231 """Run the ISIS chain on a downloaded PDS3 label (image beside it)."""
232 label = Path(label)
233 pid = label.name.split(".")[0].upper()
234 if not is_dry_run():
235 self.output_dir.mkdir(parents=True, exist_ok=True)
236 cube = self.output_dir / f"{pid}.cub"
237 out = MiOutputs(pid, cube)
239 if not reuse_existing(cube, self.force):
240 try:
241 isis("kaguyami2isis", from_=label, to=cube, **self.kaguyami2isis_params)
242 spiceinit(cube, web=self.web, **self.spiceinit_params)
243 except Exception:
244 cube.unlink(missing_ok=True) # never reuse a cube without SPICE
245 raise
246 if not self.keep and not is_dry_run(): 246 ↛ 247line 246 didn't jump to line 247 because the condition on line 246 was never true
247 name = image_pointer(label) if label.exists() else None
248 if name:
249 (label.parent / name).unlink(missing_ok=True)
251 if self.csm:
252 out.camera = self.output_dir / f"{pid}.json"
253 if not reuse_existing(out.camera, self.force):
254 isd_generate(cube, out.camera, only_naif_spice=True)
256 if self.mapproject:
257 out.geotiff = self.output_dir / f"{pid}.tif"
258 map_cube = self.output_dir / f"{pid}.map.cub"
259 if not reuse_existing(out.geotiff, self.force): 259 ↛ 267line 259 didn't jump to line 267 because the condition on line 259 was always true
260 map_project(cube, map_cube, self.projection, self.resolution, keep=self.keep)
261 run(
262 "gdal_translate",
263 ["-of", "GTiff", "-co", "COMPRESS=DEFLATE", "-co", "TILED=YES", map_cube, out.geotiff],
264 )
265 if not self.keep and not is_dry_run(): 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 map_cube.unlink(missing_ok=True)
267 if self.keep: 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true
268 out.map_cube = map_cube
269 return out
272def import_mi(label: Path, output_dir: Path | None = None, **kwargs: Any) -> MiOutputs:
273 """Import one downloaded MI product; see [`MiProcessor`][lunar_toolkit.kaguya.mi.MiProcessor] for
274 *kwargs*."""
275 label = Path(label)
276 return MiProcessor(output_dir or label.parent, **kwargs).import_label(label)