Coverage for core / src / sensorkit / astro / observer.py: 95%

1# SPDX-License-Identifier: Apache-2.0 

2import asyncio 

3import functools 

4from datetime import datetime 

5from typing import ClassVar, cast 

6 

7from astropy.coordinates import EarthLocation 

8from skyfield import almanac as almanac 

9from skyfield import api as skyfield 

10from skyfield.jpllib import SpiceKernel 

11 

12from sensorkit.astro.coords import astropy_unit 

13 

14 

15# TODO: Present incarnation is largely driven by Agent configuration usage. This may change 

16# substantially or be entirely replaced. 

17class EarthObserver: 

18 """High-level topocentric observer backed by skyfield.""" 

19 

20 timescale: ClassVar[skyfield.Timescale | None] = None 

21 ephem: ClassVar[SpiceKernel | None] = None 

22 _bootstrap_lock: ClassVar[asyncio.Lock] = asyncio.Lock() 

23 

24 @staticmethod 

25 async def bootstrap(): 

26 """Load the skyfield timescale and ephemeris data if not already loaded.""" 

27 async with EarthObserver._bootstrap_lock: 

28 if EarthObserver.timescale is None: 

29 # Load both before publishing either. `timescale` is the gate variable 

30 # checked by get(); assign it LAST so "timescale set" always implies 

31 # "ephem loaded". Otherwise a concurrent get() can observe timescale set 

32 # while the slow de421.bsp load is still in flight and construct with 

33 # ephem=None. The two assignments have no await between them, so no other 

34 # task can interleave (this also makes bootstrap cancellation-atomic). 

35 timescale = await asyncio.to_thread(skyfield.load.timescale) 

36 ephem = await asyncio.to_thread(skyfield.load, "de421.bsp") 

37 EarthObserver.ephem = ephem 

38 EarthObserver.timescale = timescale 

39 

40 @classmethod 

41 async def get(cls, *args, **kwargs): 

42 """Bootstrap skyfield and return a new EarthObserver instance.""" 

43 if cls.timescale is None: 

44 await cls.bootstrap() 

45 return cls(*args, **kwargs) 

46 

47 def __init__(self, lat_deg: float, lon_deg: float, elev_m: float = 0.0): 

48 self.topos = skyfield.Topos( 

49 latitude_degrees=lat_deg, 

50 longitude_degrees=lon_deg, 

51 elevation_m=elev_m, 

52 ) 

53 self.observer = self.ephem["earth"] + self.topos 

54 

55 @functools.cache 

56 def to_astropy(self): 

57 """Return an astropy EarthLocation for this observer (cached).""" 

58 deg = astropy_unit("deg") 

59 return EarthLocation( 

60 lon=self.topos.longitude.degrees * deg, 

61 lat=self.topos.latitude.degrees * deg, 

62 height=self.topos.elevation.m * astropy_unit("m"), 

63 ) 

64 

65 def get_sunrise_times(self, from_time: datetime, to_time: datetime): 

66 """Return all sunrise times between *from_time* and *to_time*.""" 

67 return tuple( 

68 cast(datetime, time.utc_datetime()) 

69 for time in almanac.find_risings( 

70 self.observer, 

71 self.ephem["sun"], 

72 self.timescale.from_datetime(from_time), 

73 self.timescale.from_datetime(to_time), 

74 )[0] 

75 ) 

76 

77 def get_sunrise_time(self, from_time: datetime, to_time: datetime, latest: bool = False): 

78 """Return the first (or last if *latest*) sunrise time in the window, or None.""" 

79 times = self.get_sunrise_times(from_time, to_time) 

80 return times[-1 if latest else 0] if times else None 

81 

82 def get_sunset_times(self, from_time: datetime, to_time: datetime): 

83 """Return all sunset times between *from_time* and *to_time*.""" 

84 return tuple( 

85 cast(datetime, time.utc_datetime()) 

86 for time in almanac.find_settings( 

87 self.observer, 

88 self.ephem["sun"], 

89 self.timescale.from_datetime(from_time), 

90 self.timescale.from_datetime(to_time), 

91 )[0] 

92 ) 

93 

94 def get_sunset_time(self, from_time: datetime, to_time: datetime, latest: bool = False): 

95 """Return the first (or last if *latest*) sunset time in the window, or None.""" 

96 times = self.get_sunset_times(from_time, to_time) 

97 return times[-1 if latest else 0] if times else None