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

1#! /usr/bin/env python 

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

3import re 

4 

5import urllib.request 

6 

7from km3dq_common.config_library import configure_defect 

8 

9 

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

11def read_defect_file(det, def_tag="def-HEAD", type_filter=None): 

12 """ 

13 Read defect files stored on SFTP 

14 

15 === Arguments === 

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

17 - def_tag : defect tag - [string] 

18 - type_filter : if not empty, filter on the defect type - If empty, read 

19 all defect files except the sign_off (see function below) 

20 [array of strings] - Ex: ["daq", "operation"] 

21 

22 === Output === 

23 - Dictionnary with def_[defect] as keys and a value for each affected run. 

24 This value will be decoded by the decode_defect_diag_byte function. 

25 Ex: {'def_daq': {}, 'def_operation': {16501: 1, 16503: 2}...} 

26 """ 

27 

28 defects0 = configure_defect() 

29 

30 res = {} 

31 

32 if "ORCA" in det: 

33 site = "ORCA" 

34 else: 

35 site = "ARCA" 

36 

37 re_line = re.compile(r"\s*([0-9]+)\s*\|.*") 

38 for i_type in defects0['type']: 

39 if type_filter is not None and i_type not in type_filter: 

40 continue 

41 

42 res[f"def_{i_type}"] = {} 

43 for i_diag in defects0['bit'][i_type].keys(): 

44 lines = [] 

45 defect_url = ("https://sftp.km3net.de/data/km3dq_lw_db/" 

46 f"{site}/{det}/Defects/{def_tag}/" 

47 f"{i_type}_{i_diag}.txt") 

48 try: 

49 with urllib.request.urlopen(defect_url) as def_file: 

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

51 for i_line in tmp: 

52 if i_line != "": 

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

54 except urllib.error.URLError: 

55 # No defect file found 

56 continue 

57 

58 for i_line in lines: 

59 r_m = re_line.match(i_line) 

60 if r_m is not None: 

61 run_nb = int(r_m.group(1)) 

62 if run_nb in res[f"def_{i_type}"].keys(): 

63 res[f"def_{i_type}"][run_nb] |= 1 << \ 

64 defects0['bit'][i_type][i_diag] 

65 else: 

66 res[f"def_{i_type}"][run_nb] = 1 << \ 

67 defects0['bit'][i_type][i_diag] 

68 

69 return res 

70 

71 

72############################################################################### 

73def read_basic_defect_file(det, def_type, def_diag, def_tag="def-HEAD"): 

74 """ 

75 Read defect files stored on SFTP. Simply returns lines 

76 

77 === Arguments === 

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

79 - defect_type 

80 - defect_diag 

81 - def_tag : defect tag - [string] 

82 

83 === Output === 

84 - Array of lines 

85 

86 """ 

87 

88 if "ORCA" in det: 

89 site = "ORCA" 

90 else: 

91 site = "ARCA" 

92 

93 lines = [] 

94 defect_url = ("https://sftp.km3net.de/data/km3dq_lw_db/" 

95 f"{site}/{det}/Defects/{def_tag}/" 

96 f"{def_type}_{def_diag}.txt") 

97 try: 

98 with urllib.request.urlopen(defect_url) as def_file: 

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

100 for i_line in tmp: 

101 if i_line != "": 

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

103 except urllib.error.URLError: 

104 # No defect file found 

105 lines = None 

106 

107 return lines