Coverage for lunar_toolkit/constants.py: 95%

80 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-19 02:58 +0000

1"""App name, data directories, PDS LROC archive URLs, datasets and shared defaults.""" 

2 

3import os 

4import re 

5from importlib.metadata import PackageNotFoundError, version 

6from pathlib import Path 

7 

8# Application name used for the user data / cache directories and the HTTP 

9# User-Agent. The only place the package name is spelled out at runtime. 

10APP_NAME = "lunar_toolkit" 

11try: 

12 APP_VERSION = version("lunar-toolkit") 

13except PackageNotFoundError: # running from a source tree that was never installed 

14 APP_VERSION = "0.0.0" 

15 

16# PDS archive roots 

17# The LROC SOC host (https://pds.lroc.im-ldi.com/data/<DATA_SET_ID>/...) now 

18# 302-redirects to the PDS Imaging Node S3 store below. Dataset directories 

19# there are lowercase and drop the "-V1.0" suffix, e.g. 

20# https://pds.lroc.im-ldi.com/data/LRO-L-LROC-2-EDR-V1.0/LROLRC_0066C/INDEX/ 

21# → https://pds.mcp.nasa.gov/data/store/img/lunar_reconnaissance_orbiter/ 

22# pds4/lroc/lro-l-lroc-2-edr/LROLRC_0066C/INDEX/ 

23 

24LEGACY_PDS_BASE_URL = "https://pds.lroc.im-ldi.com/data/" 

25 

26PDS_STORE_URL = "https://pds.mcp.nasa.gov/data/store/img/" 

27PDS_LROC_PREFIX = "lunar_reconnaissance_orbiter/pds4/lroc/" 

28PDS_BASE_URL = PDS_STORE_URL + PDS_LROC_PREFIX 

29 

30EDR_DATASET = "lro-l-lroc-2-edr/" 

31CDR_DATASET = "lro-l-lroc-3-cdr/" 

32RDR_DATASET = "lro-l-lroc-5-rdr/" 

33 

34EDR_BASE_URL = PDS_BASE_URL + EDR_DATASET 

35CDR_BASE_URL = PDS_BASE_URL + CDR_DATASET 

36RDR_BASE_URL = PDS_BASE_URL + RDR_DATASET 

37 

38# Fallback releases — the newest volume is auto-discovered from the S3 listing 

39# (see cumindex.discover_volumes); these are only used when that fails. 

40EDR_LATEST_RELEASE = "LROLRC_0068A/" 

41CDR_LATEST_RELEASE = "LROLRC_1068A/" 

42RDR_LATEST_RELEASE = "LROLRC_2001/" 

43 

44INDEX_DIR = "INDEX/" 

45 

46CUMINDEX_LBL_FILENAME = "CUMINDEX.LBL" 

47CUMINDEX_TAB_FILENAME = "CUMINDEX.TAB" 

48VOLINDEX_LBL_FILENAME = "INDEX.LBL" 

49VOLINDEX_TAB_FILENAME = "INDEX.TAB" 

50 

51# Dataset registry 

52# Each entry drives CumIndex URL construction and the name of the parquet file 

53# built by `lunar build-db <key>`. 

54 

55DATASETS: dict[str, dict] = { 

56 "edr": { 

57 "label": "EDR — Experiment Data Record (raw NAC/WAC images)", 

58 "data_set_id": "LRO-L-LROC-2-EDR-V1.0", 

59 "base_url": EDR_BASE_URL, 

60 "release": EDR_LATEST_RELEASE, 

61 "volume_re": r"LROLRC_0\d{3}[A-Z]?", 

62 "index_dir": INDEX_DIR, 

63 "lbl_filename": CUMINDEX_LBL_FILENAME, 

64 "tab_filename": CUMINDEX_TAB_FILENAME, 

65 "parquet": "cumindex_edr.parquet", 

66 }, 

67 "cdr": { 

68 "label": "CDR — Calibrated Data Record", 

69 "data_set_id": "LRO-L-LROC-3-CDR-V1.0", 

70 "base_url": CDR_BASE_URL, 

71 "release": CDR_LATEST_RELEASE, 

72 "volume_re": r"LROLRC_1\d{3}[A-Z]?", 

73 "index_dir": INDEX_DIR, 

74 "lbl_filename": CUMINDEX_LBL_FILENAME, 

75 "tab_filename": CUMINDEX_TAB_FILENAME, 

76 "parquet": "cumindex_cdr.parquet", 

77 }, 

78 "rdr": { 

79 "label": "RDR — Reduced Data Record (WAC mosaics, DEMs, etc.)", 

80 "data_set_id": "LRO-L-LROC-5-RDR-V1.0", 

81 "base_url": RDR_BASE_URL, 

82 "release": RDR_LATEST_RELEASE, 

83 "volume_re": r"LROLRC_2\d{3}[A-Z]?", 

84 "index_dir": INDEX_DIR, 

85 "lbl_filename": CUMINDEX_LBL_FILENAME, 

86 "tab_filename": CUMINDEX_TAB_FILENAME, 

87 "parquet": "cumindex_rdr.parquet", 

88 }, 

89} 

90 

91# PDS URL helpers 

92_LEGACY_DS_RE = re.compile(r"^(LRO-L-LROC-\d-[A-Z]+)-V\d+\.\d+$", re.IGNORECASE) 

93 

94 

95def pds_url_from_file_spec(file_spec: str) -> str: 

96 """Map a CUMINDEX ``FILE_SPECIFICATION_NAME`` to a download URL. 

97 

98 ``LRO-L-LROC-2-EDR-V1.0/LROLRC_0001/DATA/.../M101013931LE.IMG`` → 

99 ``<PDS_BASE_URL>lro-l-lroc-2-edr/LROLRC_0001/DATA/.../M101013931LE.IMG``. 

100 Paths without a dataset prefix are joined to ``PDS_BASE_URL`` unchanged. 

101 """ 

102 spec = str(file_spec).strip().lstrip("/") 

103 head, _, rest = spec.partition("/") 

104 m = _LEGACY_DS_RE.match(head) 

105 if m and rest: 105 ↛ 107line 105 didn't jump to line 107 because the condition on line 105 was always true

106 return PDS_BASE_URL + m.group(1).lower() + "/" + rest 

107 return PDS_BASE_URL + spec 

108 

109 

110def modernize_pds_url(url: str) -> str: 

111 """Rewrite a legacy ``pds.lroc.im-ldi.com/data/`` URL to the PDS store host. 

112 

113 Other URLs are returned unchanged. (The legacy host still redirects, but 

114 resolving up front saves a round-trip and keeps Range requests simple.) 

115 """ 

116 if url.startswith(LEGACY_PDS_BASE_URL): 

117 return pds_url_from_file_spec(url[len(LEGACY_PDS_BASE_URL) :]) 

118 return url 

119 

120 

121# Data directory resolution 

122# Environment variable overriding the data directory. 

123DATA_DIR_ENV_VAR = "LUNAR_DATA_DIR" 

124 

125 

126def _user_data_dir(app_name: str = APP_NAME) -> Path: 

127 try: 

128 from platformdirs import user_data_dir 

129 

130 return Path(user_data_dir(app_name, appauthor=False)) 

131 except ImportError: # pragma: no cover - platformdirs is a declared dependency 

132 base = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share") 

133 return Path(base) / app_name 

134 

135 

136def get_data_dir() -> Path: 

137 """Directory where Parquet databases are stored. 

138 

139 ``LUNAR_DATA_DIR`` if set, else the per-user data directory 

140 (``platformdirs.user_data_dir``), e.g. ``~/.local/share/lunar_toolkit`` 

141 on Linux (honours ``XDG_DATA_HOME``). 

142 """ 

143 env = os.environ.get(DATA_DIR_ENV_VAR) 

144 return Path(env) if env else _user_data_dir() 

145 

146 

147def find_parquet(name: str, data_dir: Path | None = None) -> Path: 

148 """Path of Parquet database *name* in *data_dir* (default: 

149 [`get_data_dir`][lunar_toolkit.constants.get_data_dir]).""" 

150 return Path(data_dir or get_data_dir()) / name 

151 

152 

153# User cache (raw TAB / LBL downloads) 

154DEFAULT_CACHE_DIR = Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / APP_NAME 

155 

156# Misc 

157DEFAULT_PROJECTION = "equirectangular" 

158 

159# gdal_translate flags for map-projected GeoTIFFs: native values, lossless and tiled 

160GEOTIFF_OPTIONS = ["-of", "GTiff", "-co", "COMPRESS=DEFLATE", "-co", "TILED=YES", "-co", "BIGTIFF=IF_SAFER"] 

161 

162# Mean lunar radius (km) used for polar stereographic footprints and searches 

163MOON_RADIUS_KM = 1737.4 

164 

165# |latitude| above which footprints also get polar stereographic coordinates 

166POLAR_LAT_THRESHOLD = 60.0 

167 

168# Index columns <corner>_LATITUDE / <corner>_LONGITUDE of a footprint, in ring order 

169FOOTPRINT_CORNERS = ("UPPER_LEFT", "UPPER_RIGHT", "LOWER_RIGHT", "LOWER_LEFT") 

170 

171# NAC product IDs: <observation><side L/R><type E = EDR, C = CDR>; the short 

172# base form (M1096757719L) omits the type 

173NAC_ID_RE = re.compile(r"^([A-Z]\d{9,12})([LR])([EC]?)$", re.IGNORECASE) 

174WAC_ID_RE = re.compile(r"^[A-Z]\d{9,12}[CMUV]E$", re.IGNORECASE) 

175NAC_PRODUCT_TYPES = {"E": "edr", "C": "cdr"} 

176NAC_SUFFIXES = ("LE", "RE", "LC", "RC") 

177 

178 

179def parse_nac_id(product_id: str) -> tuple[str, str, str] | None: 

180 """Split a NAC product ID into ``(observation, side, type)``. 

181 

182 ``"M1096757719LE"`` → ``("M1096757719", "L", "edr")``; a short base ID 

183 (``"M1096757719L"``) gets type ``"edr"``. None for anything else (WAC, 

184 ShadowCam, …). 

185 """ 

186 m = NAC_ID_RE.match(str(product_id).strip()) 

187 if m is None: 

188 return None 

189 obs, side, kind = m.groups() 

190 return obs.upper(), side.upper(), NAC_PRODUCT_TYPES[(kind or "E").upper()] 

191 

192 

193def nac_id(product_id: str, side: str | None = None, product_type: str | None = None) -> str | None: 

194 """Full NAC product ID of *product_id*, optionally for another side / type. 

195 

196 ``nac_id("M1096757719L")`` → ``"M1096757719LE"``; 

197 ``nac_id("M1096757719LE", side="R", product_type="cdr")`` → ``"M1096757719RC"``. 

198 None when *product_id* is not a NAC ID. 

199 """ 

200 parts = parse_nac_id(product_id) 

201 if parts is None: 

202 return None 

203 obs, own_side, own_type = parts 

204 letter = {v: k for k, v in NAC_PRODUCT_TYPES.items()}[(product_type or own_type).lower()] 

205 return f"{obs}{(side or own_side).upper()}{letter}" 

206 

207 

208def is_nac_id(product_id: str) -> bool: 

209 """True for a full NAC product ID (``LE`` / ``RE`` / ``LC`` / ``RC``).""" 

210 m = NAC_ID_RE.match(str(product_id).strip()) 

211 return bool(m and m.group(3)) 

212 

213 

214# SfS pipeline option values (lunar_toolkit.sfs.SfsConfig), here so the CLI stays light 

215REGISTRATION_MODES = ("stereo", "residuals", "none") 

216SELECTION_METHODS = ("coverage", "kmedoids", "image_subset", "all") 

217CAMERA_SOURCES = ("ba", "published") 

218SFS_CAMERA_MODELS = ("csm", "isis")