Coverage for src / lstautorta / shared_observation.py: 30%

1import json 

2import logging 

3import os 

4import tempfile 

5import time 

6from typing import NamedTuple 

7 

8DEFAULT_SHARED_OBS_PATH = "/fefs/onsite/pipeline/rta/data/current_obs.json" 

9 

10 

11class SharedObsInfo(NamedTuple): 

12 obs_id: int | None 

13 time_start_camera: float | None 

14 RA: float | None 

15 DEC: float | None 

16 source_RA: float | None 

17 source_DEC: float | None 

18 updated_ts: float | None 

19 

20 

21def write_shared_obs(path: str, obs_info) -> None: 

22 payload = { 

23 "obs_id": obs_info.obs_id, 

24 "time_start_camera": obs_info.time_start_camera, 

25 "ra": obs_info.RA, 

26 "dec": obs_info.DEC, 

27 "source_ra": obs_info.source_RA, 

28 "source_dec": obs_info.source_DEC, 

29 "updated_ts": time.time(), 

30 } 

31 parent_dir = os.path.dirname(path) 

32 os.makedirs(parent_dir, exist_ok=True) 

33 fd, tmp_path = tempfile.mkstemp(dir=parent_dir, prefix=".current_obs_", suffix=".json") 

34 try: 

35 with os.fdopen(fd, "w") as f: 

36 json.dump(payload, f) 

37 f.flush() 

38 os.fsync(f.fileno()) 

39 os.replace(tmp_path, path) 

40 finally: 

41 if os.path.exists(tmp_path): 

42 os.unlink(tmp_path) 

43 

44 

45def _parse_shared_obs(data: dict) -> SharedObsInfo | None: 

46 if not isinstance(data, dict): 

47 return None 

48 obs_id = data.get("obs_id") 

49 return SharedObsInfo( 

50 obs_id=obs_id, 

51 time_start_camera=data.get("time_start_camera"), 

52 RA=data.get("ra"), 

53 DEC=data.get("dec"), 

54 source_RA=data.get("source_ra"), 

55 source_DEC=data.get("source_dec"), 

56 updated_ts=data.get("updated_ts"), 

57 ) 

58 

59 

60def read_shared_obs(path: str) -> SharedObsInfo | None: 

61 if not os.path.exists(path): 

62 return None 

63 try: 

64 with open(path) as f: 

65 data = json.load(f) 

66 except json.JSONDecodeError: 

67 return None 

68 except OSError as e: 

69 logging.exception("Failed to read shared observation file %s: %s", path, e) 

70 return None 

71 return _parse_shared_obs(data) 

72 

73 

74def wait_for_shared_obs( 

75 path: str, 

76 poll_s: int = 5, 

77 timeout_s: int = 12 * 3600, 

78) -> SharedObsInfo: 

79 start_time = time.time() 

80 

81 while True: 

82 obs_info = read_shared_obs(path) 

83 if obs_info is not None and obs_info.obs_id is not None: 

84 return obs_info 

85 

86 if time.time() - start_time > timeout_s: 

87 raise TimeoutError(f"No shared observation available in {path} after {timeout_s} seconds") 

88 

89 time.sleep(poll_s)