Coverage for src/km3dq_common/detector_status_classes.py: 0%

1#!/usr/bin/env python3 

2import datetime 

3import re 

4 

5from .lw_db_det_fact_library import read_det_fact_file_2025 

6 

7 

8RE_TIME = re.compile(r"(\d\d\d\d)-(\d\d)-(\d\d)\s(\d\d):(\d\d)") 

9 

10############################################################################### 

11def strtime_to_datetime(strtime): 

12 """ 

13 Convert a time string (like 2025-03-05 00:20) into a datetime 

14 """ 

15 

16 spl_time = RE_TIME.match(strtime) 

17 return datetime.datetime( 

18 int(spl_time.group(1)), 

19 int(spl_time.group(2)), 

20 int(spl_time.group(3)), 

21 int(spl_time.group(4)), 

22 int(spl_time.group(5)) 

23 ).timestamp() 

24 

25 

26############################################################################### 

27class DetectorStatus(dict): 

28 """ 

29 Detector fact classes ingeriting from a dict 

30 https://pynative.com/make-python-class-json-serializable/ 

31 https://realpython.com/inherit-python-dict/ 

32 """ 

33 

34 ####################################### 

35 def __init__(self, site, date_time=""): 

36 dict.__init__(self, 

37 site="", 

38 time="", 

39 datetimetime=0, 

40 hardware={} 

41 ) 

42 

43 self.site = site 

44 self.time = date_time # String "yyyy-mm-dd hh:mm" 

45 if self.time != "": # Past detector status (i.e not today) 

46 self.datetime = strtime_to_datetime(self.time) 

47 else: # Today 

48 self.datetime = 0 

49 self.hardware = {} 

50 

51 ################################# 

52 def set_installed_hardware(self, all_df): 

53 """ 

54 Loop on all detector facts of a site and define the installed hardware 

55 """ 

56 # all_df = read_det_fact_file_2025(self.site) 

57 # Loop on all detector facts 

58 for i_df in all_df: 

59 # Keep only the detector fact priors to the requested time 

60 # Id ds_time == 0 (i.e today), keep all detector facts 

61 i_df_time = strtime_to_datetime(i_df.time) 

62 if i_df_time > self.datetime and self.datetime != 0: 

63 # Detector fact after the detector status time 

64 continue 

65 

66 # The upi is now used as index 

67 # NB: the base autonomeous beacons have a _b suffix appended 

68 # to makek their upi different from the DU ones. 

69 i_hw_upi = i_df.upi 

70 

71 if i_hw_upi in self.hardware.keys(): 

72 # Position already known. Check if more recent 

73 if (i_df_time < self.hardware[i_hw_upi]["time"]): 

74 continue 

75 

76 if i_hw_upi in self.hardware.keys() and i_df.status == "removed": 

77 self.hardware.pop(i_hw_upi) 

78 else: 

79 self.hardware[i_hw_upi] = { 

80 "type": i_df.hardware, 

81 "position": i_df.position, 

82 "time": i_df_time, 

83 "run": i_df.following_run, 

84 "location": i_df.location, 

85 "coord_detx": i_df.coord_detx, 

86 "coord_utm": i_df.coord_utm, 

87 "waveform": i_df.waveform, 

88 "pattern": i_df.pattern, 

89 "emission_rate": i_df.emission_rate, 

90 "upi": re.sub("\s*$", "", i_df.upi), 

91 "status": i_df.status, 

92 "documentation": re.sub("\s*$", "", i_df.documentation) 

93 }