Coverage for hiperta_stream/config/stream_configuration.py: 0%

1#!/usr/bin/env python 

2 

3from pathlib import Path 

4from pkg_resources import resource_filename 

5import logging 

6 

7from ..utils.subprocess_utils import subprocess_run_and_raise_exception_on_error 

8 

9 

10def create_drs4(drs4_run, output_directory): 

11 """Run the `lstchain_data_create_drs4_pedestal_file` entry point 

12 

13 If a file already exists in the output_directory, it will be replaced. 

14 

15 Parameters 

16 ---------- 

17 drs4_run : pathlib.Path 

18 Path to drs4 run .fits.fz file 

19 output_directory: pathlib.Path 

20 Path to output directory 

21 

22 Returns 

23 ------- 

24 pathlib.Path 

25 The path to the successfully created drs4 file. 

26 """ 

27 

28 drs4_file = Path(output_directory, drs4_run.name.replace(".fits.fz", ".fits").replace("LST-", "drs4_LST-")) 

29 logging.info(f"Starting creation of drs4 file {drs4_file}.") 

30 if drs4_file.exists(): 

31 logging.warning(f"DRS4 file {drs4_file} already exists ! Erasing it and creating it again.") 

32 drs4_file.unlink() 

33 

34 cmd_drs4 = [ 

35 "lstchain_data_create_drs4_pedestal_file", 

36 "--input-file", 

37 f"{drs4_run.absolute().as_posix()}", 

38 "--output-file", 

39 f"{drs4_file.absolute().as_posix()}", 

40 ] 

41 

42 subprocess_run_and_raise_exception_on_error( 

43 cmd_drs4, 

44 "DRS4 file successfully created: {}".format(drs4_file.name), 

45 "DRS4 file not correctly created!", 

46 ) 

47 

48 

49def create_calibration(drs4_file, calib_run, lstchain_config, output_directory): 

50 """Run the `lstchain_create_calibration_file` entry point 

51 

52 If a file already exists in the output directory, it will be replaced. 

53 

54 Parameters 

55 ---------- 

56 drs4_file: pathlib.Path 

57 Path to drs4 .fits file 

58 calib_run: pathlib.Path 

59 Path to calibration run .fits.fz file 

60 lstchain_config: pathlib.Path 

61 Path to lstchain .json configuration file 

62 output_directory: pathlib.Path 

63 Path to output directory 

64 

65 Returns 

66 ------- 

67 pathlib.Path 

68 Path to the successfully created calibration file. 

69 """ 

70 

71 find_run = calib_run.name.find("Run") 

72 calibration_file = Path(output_directory, f"calibration_{calib_run.name[find_run:find_run+8]}.h5") 

73 

74 logging.info("Starting creation of calibration file.") 

75 if calibration_file.exists(): 

76 logging.warning("Ped-cal file already exists ! Erasing it and creating it again.") 

77 calibration_file.unlink() 

78 

79 cmd_calib_file = [ 

80 "lstchain_create_calibration_file", 

81 f"--input_file={calib_run.absolute().as_posix()}", 

82 f"--pedestal_file={drs4_file.absolute().as_posix()}", 

83 f"--config={lstchain_config.absolute().as_posix()}", 

84 f"--output_file={calibration_file.absolute().as_posix()}", 

85 ] 

86 

87 subprocess_run_and_raise_exception_on_error( 

88 cmd_calib_file, 

89 "Calibration file successfully created: {}".format(calibration_file.name), 

90 "Calibration file not correctly created!", 

91 ) 

92 

93 

94def find_camera_pix_order(drs4_run, output_directory): 

95 """Run the `rta_get_zfits_pix_order` entry point 

96 

97 Parameters 

98 ---------- 

99 

100 drs4_run : pathlib.Path 

101 Path to drs4 run .fits.fz file 

102 output_directory : pathlib.Path 

103 Path to output directory 

104 

105 Returns 

106 ------- 

107 pathlib.Path 

108 Path to the successfully created pixel order file. 

109 """ 

110 

111 pix_order_file = Path(output_directory, drs4_run.name.replace(".fits.fz", "_pix_order_bin.npy")) 

112 logging.info(f"Starting creation of binary pix order file {pix_order_file}") 

113 

114 cmd_pix_order = [ 

115 f"rta_get_zfits_pix_order", 

116 "--input", 

117 f"{drs4_run.absolute().as_posix()}", 

118 "--output", 

119 f"{pix_order_file.absolute().as_posix()}", 

120 ] 

121 

122 subprocess_run_and_raise_exception_on_error( 

123 cmd_pix_order, 

124 "Binary file with pix order successfully created: {}".format(pix_order_file.name), 

125 "Binary file containing pixel order of the camera not correctly created!", 

126 ) 

127 

128 

129def create_stream_base_structure(pix_order, calibration_file, output_directory): 

130 """Run the `rta_create_base_config` entry point 

131 

132 Parameters 

133 ---------- 

134 pix_order: pathlib.Path 

135 Path to pix_order.npy binary file 

136 calibration_file: pathlib.Path 

137 Path to calibration.h5 file create in a previous stage 

138 output_directory: pathlib.Path 

139 Path to output directory 

140 

141 Returns 

142 ------- 

143 pathlib.Path 

144 Path to the successfully created base_config file. 

145 """ 

146 

147 base_config_file = Path( 

148 output_directory, f'base_structure_hdf5_{calibration_file.name.split("Run")[-1].lstrip("0")}' 

149 ) 

150 logging.info(f"Starting creation of base hdf5 configuration file {base_config_file}") 

151 

152 # TODO subprocess does not like env vars. This is not the best way to solve it, we know it ;-) 

153 # Patch-1 - substituted `getenv` for `pkg_resources.resource_filename` 

154 # If it breaks again, we will put everything in a single str. 

155 cmd_create_base_structure = [ 

156 "rta_create_base_config", 

157 "--config_file", 

158 f'{resource_filename("hiperta_stream", "dataset/default_configuration.h5")}', 

159 "--pixel_order", 

160 f"{pix_order.absolute().as_posix()}", 

161 "--gain_pedestal", 

162 f"{calibration_file.absolute().as_posix()}", 

163 "--output_file", 

164 f"{base_config_file.absolute().as_posix()}", 

165 ] 

166 

167 subprocess_run_and_raise_exception_on_error( 

168 cmd_create_base_structure, 

169 "Base hdf5 configuration file successfully created: {}".format(base_config_file.name), 

170 "Base hdf5 configuration file not correctly created!", 

171 )