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

1#! /usr/bin/env python 

2############################################################################### 

3import re 

4import sys 

5import calendar 

6import time 

7import urllib.request 

8 

9 

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

11def read_basic_fact_file(det, fact_type): 

12 """ 

13 Read defect files stored on SFTP. Simply returns lines 

14 

15 === Arguments === 

16 - det : detector name - [string] - Ex: "D0ARCA021", "D0ORCA018"... 

17 - fact_type 

18 

19 === Output === 

20 - Array of lines 

21 

22 """ 

23 

24 if "ORCA" in det: 

25 site = "ORCA" 

26 else: 

27 site = "ARCA" 

28 

29 lines = [] 

30 fact_url = ( 

31 "https://sftp.km3net.de/data/km3dq_lw_db/" 

32 f"{site}/{det}/Facts/" 

33 f"{fact_type}.txt" 

34 ) 

35 try: 

36 with urllib.request.urlopen(fact_url) as def_file: 

37 tmp = (def_file.read()).split(b"\n") 

38 for i_line in tmp: 

39 if i_line != "": 

40 lines.append(i_line.decode("utf-8")) 

41 except urllib.error.URLError: 

42 # No fact file found 

43 lines = None 

44 

45 return lines 

46 

47 

48############################################################################### 

49def read_fact_file(det, fact_type): 

50 """ 

51 Read defect files stored on SFTP. 

52 

53 === Arguments === 

54 - det : detector name - [string] - Ex: "D0ARCA021", "D0ORCA018"... 

55 - fact_type 

56 

57 === Output === 

58 - Array of lines 

59 

60 """ 

61 

62 if "ORCA" in det: 

63 site = "ORCA" 

64 else: 

65 site = "ARCA" 

66 

67 re_line = re.compile( 

68 r"\s*(\S*)\s*\|" 

69 r"\s*(\S*)\s*\|" 

70 r"\s*(\S.*\d)\s\-\s(\S.*\d)\s*\|" 

71 r"(.*)\|" 

72 r"(.*)\|" 

73 r"(.*)" 

74 ) 

75 

76 re_line_no_end = re.compile( 

77 r"\s*(\S*)\s*\|" 

78 r".*\|" 

79 r"\s*(\S.*\d)\s\-\s*\|" 

80 r"(.*)\|" 

81 r"(.*)\|" 

82 r"(.*)" 

83 ) 

84 

85 fact_list = [] 

86 fact_url = ( 

87 "https://sftp.km3net.de/data/km3dq_lw_db/" 

88 f"{site}/{det}/Facts/" 

89 f"{fact_type}.txt" 

90 ) 

91 try: 

92 with urllib.request.urlopen(fact_url) as def_file: 

93 tmp = (def_file.read()).split(b"\n") 

94 for i_line in tmp: 

95 r_m = re_line.match(i_line.decode("utf-8")) 

96 if r_m: 

97 # Intermediate step to handle facts with and without second accuracy 

98 # Accuracy must be the same for both end/start by construction 

99 if r_m.group(3).count(":") == 1: 

100 t_s = int(calendar.timegm(time.strptime(r_m.group(3), "%a, %d %b %Y %H:%M"))) 

101 t_e = int(calendar.timegm(time.strptime(r_m.group(4), "%a, %d %b %Y %H:%M"))) 

102 else: 

103 t_s = int(calendar.timegm(time.strptime(r_m.group(3), "%a, %d %b %Y %H:%M:%S"))) 

104 t_e = int(calendar.timegm(time.strptime(r_m.group(4), "%a, %d %b %Y %H:%M:%S"))) 

105 

106 fact_list.append({ 

107 "run_start": r_m.group(1), 

108 "run_end": r_m.group(2), 

109 "time_start": t_s, 

110 "time_end": t_e, 

111 "comment": r_m.group(5), 

112 "documentation": r_m.group(6), 

113 "author": r_m.group(7), 

114 }) 

115 continue 

116 r_m = re_line_no_end.match(i_line.decode("utf-8")) 

117 if r_m: 

118 # Intermediate step to handle facts with and without second accuracy 

119 if r_m.group(2).count(":") == 1: 

120 t_s = int(calendar.timegm(time.strptime(r_m.group(2), "%a, %d %b %Y %H:%M"))) 

121 else: 

122 t_s = int(calendar.timegm(time.strptime(r_m.group(2), "%a, %d %b %Y %H:%M:%S"))) 

123 fact_list.append({ 

124 "run_start": r_m.group(1), 

125 "run_end": 1e6, 

126 "time_start": t_s, 

127 "time_end": 0, 

128 "comment": r_m.group(3), 

129 "documentation": r_m.group(4), 

130 "author": r_m.group(5), 

131 }) 

132 

133 

134 

135 except urllib.error.URLError: 

136 # No fact file found 

137 lines = None 

138 

139 

140 return fact_list