Coverage for lunar_toolkit/observation.py: 79%

1"""Common observation schema shared by all search backends. 

2 

3Longitudes are 0–360 east (``lon_west > lon_east`` crosses the seam); angles in degrees, 

4``pixel_scale`` in m/px, ``spacecraft_altitude`` in km; missing values are None.""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING 

9 

10from .constants import is_nac_id 

11 

12if TYPE_CHECKING: # pragma: no cover 

13 import pandas as pd 

14 

15OBSERVATION_COLUMNS: list[str] = [ 

16 "product_id", 

17 "instrument", # e.g. "LROC NAC", "ShadowCam", "Kaguya TC" 

18 "product_type", # e.g. "EDR", "CDR", "CAL", "RAW", "DTM" 

19 "start_time", # pandas Timestamp (UTC, tz-naive) 

20 "center_lat", 

21 "center_lon", 

22 "min_lat", 

23 "max_lat", 

24 "lon_west", 

25 "lon_east", 

26 "footprint_wkt", # lat/lon WKT if known (ODE: 0–360 cylindrical) 

27 "incidence", 

28 "emission", 

29 "phase", 

30 "slew_angle", 

31 "sub_solar_azimuth", # None when the source does not provide it 

32 "north_azimuth", # PDS NORTH_AZIMUTH (image-frame), None if unknown 

33 "sub_spacecraft_lat", # sub-spacecraft point at image centre time 

34 "sub_spacecraft_lon", 

35 "spacecraft_altitude", # km above the surface 

36 "pixel_scale", 

37 "url", # primary product file 

38 "label_url", # PDS label (.xml / .lbl) 

39 "source", # "cumindex" or "ode" 

40] 

41 

42 

43def empty_observations() -> pd.DataFrame: 

44 import pandas as pd 

45 

46 return pd.DataFrame({c: pd.Series(dtype="object") for c in OBSERVATION_COLUMNS}) 

47 

48 

49def normalize_observations(records: list[dict]) -> pd.DataFrame: 

50 """Build an observation DataFrame from dicts (missing keys → None).""" 

51 import pandas as pd 

52 

53 if not records: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 return empty_observations() 

55 df = pd.DataFrame.from_records( 

56 [{c: r.get(c) for c in OBSERVATION_COLUMNS} for r in records], 

57 columns=OBSERVATION_COLUMNS, 

58 ) 

59 for c in ( 

60 "center_lat", 

61 "center_lon", 

62 "min_lat", 

63 "max_lat", 

64 "lon_west", 

65 "lon_east", 

66 "incidence", 

67 "emission", 

68 "phase", 

69 "slew_angle", 

70 "sub_solar_azimuth", 

71 "pixel_scale", 

72 "north_azimuth", 

73 "sub_spacecraft_lat", 

74 "sub_spacecraft_lon", 

75 "spacecraft_altitude", 

76 ): 

77 df[c] = pd.to_numeric(df[c], errors="coerce") 

78 df["start_time"] = pd.to_datetime(df["start_time"], errors="coerce", utc=True).dt.tz_localize(None) 

79 return df 

80 

81 

82def observations_from_cumindex(df: pd.DataFrame, dataset: str | None = "edr") -> pd.DataFrame: 

83 """Map enriched ``CumIndex`` query results to `OBSERVATION_COLUMNS`.""" 

84 import pandas as pd 

85 

86 if df is None or df.empty: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 return empty_observations() 

88 

89 def g(row, col): 

90 v = row.get(col) 

91 try: 

92 return None if pd.isna(v) else v 

93 except (TypeError, ValueError): # array-like value 

94 return v 

95 

96 recs = [] 

97 for _, row in df.iterrows(): 

98 pid = str(row.get("PRODUCT_ID", "")) 

99 recs.append( 

100 { 

101 "product_id": pid, 

102 "instrument": "LROC NAC" if is_nac_id(pid) else "LROC WAC", 

103 "product_type": (dataset or "").upper() or None, 

104 "start_time": g(row, "START_TIME"), 

105 "center_lat": g(row, "CENTER_LATITUDE"), 

106 "center_lon": g(row, "CENTER_LONGITUDE"), 

107 "min_lat": g(row, "FP_MIN_LAT"), 

108 "max_lat": g(row, "FP_MAX_LAT"), 

109 "lon_west": g(row, "FP_LON_WEST"), 

110 "lon_east": g(row, "FP_LON_EAST"), 

111 "footprint_wkt": g(row, "FP_WKT"), 

112 "incidence": g(row, "INCIDENCE_ANGLE"), 

113 "emission": g(row, "EMISSION_ANGLE"), 

114 "phase": g(row, "PHASE_ANGLE"), 

115 "slew_angle": g(row, "SLEW_ANGLE"), 

116 "sub_solar_azimuth": g(row, "SUB_SOLAR_AZIMUTH"), 

117 "north_azimuth": g(row, "NORTH_AZIMUTH"), 

118 "sub_spacecraft_lat": g(row, "SUB_SPACECRAFT_LATITUDE"), 

119 "sub_spacecraft_lon": g(row, "SUB_SPACECRAFT_LONGITUDE"), 

120 "spacecraft_altitude": g(row, "SPACECRAFT_ALTITUDE"), 

121 "pixel_scale": g(row, "RESOLUTION") if "RESOLUTION" in row else g(row, "PIXEL_SCALE"), 

122 "url": g(row, "PDS_URL"), 

123 "label_url": g(row, "LABEL_URL"), 

124 "source": "cumindex", 

125 } 

126 ) 

127 return normalize_observations(recs)