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

1#!/usr/bin/env python3 

2import datetime 

3import re 

4 

5from .lw_db_fact_library import read_det_fact_file 

6 

7 

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

9class DetectorStatus(dict): 

10 """ 

11 Detector fact classes ingeriting from a dict 

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

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

14 """ 

15 

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

17 dict.__init__(self, 

18 site="", 

19 time="", 

20 hardware={} 

21 ) 

22 

23 self.site = site 

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

25 self.hardware = {} 

26 

27 def set_installed_hardware(self): 

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

29 

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

31 spl_time = re_time.match(self.time) 

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

33 int(spl_time.group(2)), 

34 int(spl_time.group(3)), 

35 int(spl_time.group(4)), 

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

37 else: # Today 

38 ds_time = 0 

39 

40 all_df = read_det_fact_file(self.site) 

41 # Loop on all detector facts 

42 for i_df in all_df: 

43 # Extract the time of the detector fact 

44 spl_time = re_time.match(i_df.time) 

45 if spl_time: 

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

47 int(spl_time.group(2)), 

48 int(spl_time.group(3)), 

49 int(spl_time.group(4)), 

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

51 else: 

52 print("Wrong time format - Skip") 

53 continue 

54 

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

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

57 if i_time > ds_time and ds_time != 0: 

58 # Detector fact after the detector status time 

59 continue 

60 

61 # The upi is now used as index 

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

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

64 i_hw_upi = i_df.upi 

65 

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

67 # Position already known. Check if more recent 

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

69 continue 

70 

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

72 self.hardware.pop(i_hw_upi) 

73 else: 

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

75 "position": i_df.position, 

76 "time": i_time, 

77 "run": i_df.following_run, 

78 "location": i_df.location, 

79 "coord_detx": i_df.coord_detx, 

80 "coord_utm": i_df.coord_utm, 

81 "waveform": i_df.waveform, 

82 "pattern": i_df.pattern, 

83 "emission_rate": i_df.emission_rate, 

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

85 "status": i_df.status, 

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