Coverage for lunar_toolkit/cli/_utils.py: 57%

86 statements  

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

1"""Shared helpers for CLI commands.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import sys 

7from collections.abc import Callable 

8from pathlib import Path 

9from typing import Any 

10from urllib.parse import urlparse 

11 

12 

13def load_sources(args) -> list[str]: 

14 """Collect source strings from positional args and/or --from-file.""" 

15 sources: list[str] = list(getattr(args, "sources", []) or []) 

16 from_file = getattr(args, "from_file", None) 

17 if from_file: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 p = Path(from_file) 

19 if not p.exists(): 

20 print(f"Error: file not found: {from_file}", file=sys.stderr) 

21 sys.exit(1) 

22 for line in p.read_text().splitlines(): 

23 line = line.strip() 

24 if line and not line.startswith("#"): 

25 sources.append(line) 

26 return sources 

27 

28 

29def extract_product_id(source: str) -> str: 

30 """Derive a product ID from a product ID string, local path, or URL.""" 

31 s = source.strip() 

32 if s.startswith(("http://", "https://")): 

33 return Path(urlparse(s).path).stem.upper() 

34 if not _looks_like_id(s): 

35 return Path(s).stem.upper() 

36 return s.upper() 

37 

38 

39def _looks_like_id(s: str) -> bool: 

40 from ..constants import NAC_ID_RE, WAC_ID_RE 

41 

42 return bool(NAC_ID_RE.match(s) or WAC_ID_RE.match(s)) 

43 

44 

45def resolve_images(sources: list[str], raw_dir: Path, data_dir=None, download: bool = True) -> list[Path]: 

46 """Local .IMG / .cub paths as-is; NAC product IDs downloaded (EDR) to *raw_dir*. 

47 

48 With ``download=False`` IDs map to ``<raw_dir>/<ID>.IMG`` without fetching. 

49 """ 

50 from ..lroc.nac import NacImageCollection 

51 

52 paths = [Path(s) for s in sources if Path(s).exists()] 

53 ids = [s for s in sources if not Path(s).exists()] 

54 if ids and not download: 

55 return paths + [Path(raw_dir) / f"{i.upper()}.IMG" for i in ids] 

56 if ids: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 found = NacImageCollection.from_sources(ids, data_dir=data_dir) 

58 missing = {i.upper() for i in ids} - {img.product_id for img in found} 

59 if missing: 

60 raise SystemExit( 

61 f"Not in the local index: {', '.join(sorted(missing))} " 

62 "(run `lunar build-db edr` or pass .IMG paths)" 

63 ) 

64 downloaded = found.download_edrs(raw_dir) 

65 failed = [pid for pid, p in downloaded.items() if p is None] 

66 if failed: 

67 raise SystemExit(f"Download failed: {', '.join(failed)}") 

68 paths += [p for p in (downloaded[img.product_id] for img in found) if p is not None] 

69 return paths 

70 

71 

72def region_args(args: argparse.Namespace) -> dict[str, Any]: 

73 """Search-region keywords from ``--bbox`` / ``--point`` + ``--radius-km`` / ``--aoi`` (empty if none).""" 

74 if args.bbox: 

75 west, south, east, north = args.bbox 

76 return {"lat": (south, north), "lon": (west, east)} 

77 if args.point: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 return {"point": tuple(args.point), "radius_km": args.radius_km} 

79 if getattr(args, "aoi", None): 

80 return {"aoi": args.aoi} 

81 return {} 

82 

83 

84def ode_filters(args: argparse.Namespace) -> dict[str, Any]: 

85 """[`OdeClient.search`][lunar_toolkit.ode.OdeClient.search] filters from ``--inc`` / ``--emi`` / 

86 ``--phase`` / ``--time``.""" 

87 return { 

88 "incidence": tuple(args.inc) if args.inc else None, 

89 "emission": tuple(args.emi) if args.emi else None, 

90 "phase": tuple(args.phase) if args.phase else None, 

91 "time_range": tuple(args.time) if args.time else None, 

92 } 

93 

94 

95def print_dry_run(func: Callable[[], object]) -> None: 

96 """Call *func* with tool commands recorded instead of run, then print them.""" 

97 from ..tools import dry_run, format_command 

98 

99 with dry_run() as cmds: 

100 func() 

101 for c in cmds: 

102 print(format_command(c)) 

103 

104 

105def parse_options(items: list[str] | None) -> dict: 

106 """``["max_iterations=5", "float_albedo=true", "x=none"]`` -> typed dict.""" 

107 out: dict = {} 

108 for item in items or []: 

109 key, sep, value = item.partition("=") 

110 if not sep: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 raise SystemExit(f"Expected KEY=VALUE, got {item!r}") 

112 out[key.strip().replace("-", "_")] = _typed(value.strip()) 

113 return out 

114 

115 

116def _typed(value: str): 

117 low = value.lower() 

118 if low in ("true", "yes", "on", ""): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 return True 

120 if low in ("false", "no", "off"): 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 return False 

122 if low == "none": 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 return None 

124 for cast in (int, float): 124 ↛ 129line 124 didn't jump to line 129 because the loop on line 124 didn't complete

125 try: 

126 return cast(value) 

127 except ValueError: 

128 pass 

129 return value