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

1#!/usr/bin/env python3 

2import datetime 

3import re 

4 

5from km3dq_common.lw_db_fact_library import read_det_fact_file 

6 

7 

8############################################################################### 

9class DetectorFact(dict): 

10 """ 

11 Detector fact classes ingeriting from a dict 

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

13 """ 

14 def __init__(self): 

15 dict.__init__(self, 

16 hardware="", 

17 location="", 

18 position=0, 

19 status="", 

20 upi=0, 

21 site="", 

22 det="", 

23 time="", 

24 following_run=0, 

25 comment="", 

26 waveform="", 

27 documentation="", 

28 author="") 

29 

30 

31############################################################################### 

32class DetectorStatus(dict): 

33 """ 

34 Detector fact classes ingeriting from a dict 

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

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

37 """ 

38 

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

40 dict.__init__(self, 

41 site="", 

42 time="", 

43 hardware={} 

44 ) 

45 

46 self.site = site 

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

48 self.hardware = {} 

49 

50 def set_installed_hardware(self): 

51 re_time = re.compile(r"(\d\d\d\d)-(\d\d)-(\d\d)\s(\d\d):(\d\d)") 

52 

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

54 spl_time = re_time.match(self.time) 

55 ds_time = datetime.datetime(int(spl_time.group(1)), 

56 int(spl_time.group(2)), 

57 int(spl_time.group(3)), 

58 int(spl_time.group(4)), 

59 int(spl_time.group(5))).timestamp() 

60 else: # Today 

61 ds_time = 0 

62 

63 (all_df, _) = read_det_fact_file(self.site) 

64 # Loop on all detector facts 

65 for i_df in all_df: 

66 if i_df.hardware == "det": 

67 continue 

68 

69 # Extract the time of the detector fact 

70 spl_time = re_time.match(i_df.time) 

71 if spl_time: 

72 i_time = datetime.datetime(int(spl_time.group(1)), 

73 int(spl_time.group(2)), 

74 int(spl_time.group(3)), 

75 int(spl_time.group(4)), 

76 int(spl_time.group(5))).timestamp() 

77 else: 

78 print("Wrong time format - Skip") 

79 continue 

80 

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

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

83 if i_time > ds_time and ds_time != 0: 

84 # Detector fact after the detector status time 

85 continue 

86 

87 # The upi is now used as index 

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

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

90 i_hw_upi = i_df.upi 

91 

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

93 # Position already known. Check if more recent 

94 if (i_time < self.hardware[i_hw_upi]["time"]): 

95 continue 

96 

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

98 self.hardware.pop(i_hw_upi) 

99 else: 

100 self.hardware[i_hw_upi] = {"type": i_df.hardware, 

101 "position": i_df.position, 

102 "time": i_time, 

103 "run": i_df.following_run, 

104 "location": i_df.location, 

105 "waveform": i_df.waveform, 

106 "upi": i_df.upi, 

107 "status": i_df.status}