Coverage for lunar_toolkit/cli/search.py: 77%

1"""search - footprint search of the local PDS index (poles and 0/360 safe).""" 

2 

3import argparse 

4import sys 

5from pathlib import Path 

6 

7COMMAND = "search" 

8 

9 

10def add_parser(sub: argparse._SubParsersAction) -> None: 

11 p = sub.add_parser( 

12 COMMAND, 

13 help="Find images whose footprint intersects a box, point+radius or AOI polygon (local index).", 

14 formatter_class=argparse.RawDescriptionHelpFormatter, 

15 epilog=""" 

16Examples 

17-------- 

18 lunar search --bbox 0 -90 360 -85 --nac-only --inc 80 95 

19 lunar search --bbox 355 -5 5 5 # crosses the 0/360 seam 

20 lunar search --point -89.67 129.8 --radius-km 15 --csv shackleton.csv 

21 lunar search --aoi aoi.geojson --nac-only # QuickMap-exported polygon 

22 lunar search --aoi "POLYGON((350 -5, 10 -5, 10 5, 350 5, 350 -5))" 

23 

24--bbox is WEST SOUTH EAST NORTH (degrees, positive east; WEST > EAST crosses 0/360). 

25--aoi takes a GeoJSON / WKT file or a WKT string (lon lat, degrees, positive 

26east); bbox matches are refined by exact polygon intersection. 

27""", 

28 ) 

29 p.add_argument("--dataset", choices=["edr", "cdr"], default="edr", help="Index to search (default: edr).") 

30 region = p.add_mutually_exclusive_group(required=True) 

31 region.add_argument("--bbox", nargs=4, type=float, metavar=("WEST", "SOUTH", "EAST", "NORTH")) 

32 region.add_argument("--point", nargs=2, type=float, metavar=("LAT", "LON")) 

33 region.add_argument( 

34 "--aoi", metavar="FILE_OR_WKT", help="Area of interest: GeoJSON / WKT file or WKT polygon string." 

35 ) 

36 p.add_argument("--radius-km", type=float, default=1.0, metavar="KM") 

37 p.add_argument("--inc", nargs=2, type=float, metavar=("MIN", "MAX"), help="Incidence angle range.") 

38 p.add_argument("--emi", nargs=2, type=float, metavar=("MIN", "MAX"), help="Emission angle range.") 

39 p.add_argument("--phase", nargs=2, type=float, metavar=("MIN", "MAX"), help="Phase angle range.") 

40 p.add_argument("--nac-only", action="store_true", help="Only NAC products.") 

41 p.add_argument("--limit", type=int, default=None, metavar="N") 

42 p.add_argument("--csv", metavar="FILE", help="Write matching rows to CSV.") 

43 p.add_argument("--wkt", action="store_true", help="Include an FP_WKT footprint column.") 

44 p.add_argument( 

45 "--data-dir", type=Path, default=None, metavar="DIR", help="Override Parquet data directory." 

46 ) 

47 

48 

49def run(args: argparse.Namespace) -> int: 

50 from ..lroc.cumindex import CumIndex 

51 from ._utils import region_args 

52 

53 ci = CumIndex(dataset=args.dataset, data_dir=args.data_dir) 

54 try: 

55 df = ci.search( 

56 **region_args(args), 

57 incidence_angle=tuple(args.inc) if args.inc else None, 

58 emission_angle=tuple(args.emi) if args.emi else None, 

59 phase_angle=tuple(args.phase) if args.phase else None, 

60 nac_only=args.nac_only, 

61 wkt=args.wkt, 

62 limit=args.limit, 

63 ) 

64 except FileNotFoundError as exc: 

65 print(exc, file=sys.stderr) 

66 return 1 

67 except ValueError as exc: 

68 print(f"Error: {exc}", file=sys.stderr) 

69 return 1 

70 

71 print(f"{len(df)} image(s)") 

72 if not df.empty: 72 ↛ 87line 72 didn't jump to line 87 because the condition on line 72 was always true

73 cols = [ 

74 c 

75 for c in ( 

76 "PRODUCT_ID", 

77 "START_TIME", 

78 "CENTER_LATITUDE", 

79 "CENTER_LONGITUDE", 

80 "INCIDENCE_ANGLE", 

81 "EMISSION_ANGLE", 

82 "RESOLUTION", 

83 ) 

84 if c in df.columns 

85 ] 

86 print(df[cols].to_string(index=False, max_rows=40)) 

87 if args.csv: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 df.to_csv(args.csv, index=False) 

89 print(f"Wrote {args.csv}") 

90 return 0