Coverage for rta_to_lstchain_formator/online_alt_az_dl1_maker.py: 88%

1#!/usr/bin/env python 

2# S. Caroff 

3 

4""" 

5ONLINE analysis: 

6Produce online alt az from RA DEC of the pointing position 

7 

8Usage: 

9$> online_alt_az_dl1_maker 

10 --input-file "./dl1_5202_0_10000evt.h5" 

11 --RA 83.254 

12 --DEC 102.35 

13""" 

14 

15import argparse 

16import time 

17from pathlib import Path 

18 

19import astropy.units as u 

20import numpy as np 

21import tables 

22from astropy.coordinates import AltAz, EarthLocation, SkyCoord 

23from astropy.time import Time 

24from astropy.utils import iers 

25 

26from rta_to_lstchain_formator.utils.rta_argparse import EnumAction 

27from rta_to_lstchain_formator.utils.rta_logging import LOGGING_LEVELS, init_logging 

28 

29iers.conf.auto_download = False 

30 

31 

32def online_alt_az_dl1_maker_arg_parser() -> argparse.ArgumentParser: 

33 parser = argparse.ArgumentParser( 

34 description="Add alt-az values in-place in DL1 params in a file, from RA-DEC values already in the file", 

35 formatter_class=argparse.ArgumentDefaultsHelpFormatter, 

36 ) 

37 parser.add_argument( 

38 "--DEC", type=float, dest="DEC", help="Declination of the telescope", default=None, required=True 

39 ) 

40 parser.add_argument( 

41 "--input-file", 

42 "-f", 

43 type=str, 

44 dest="input_file", 

45 help='DL1 file, example : "dl1_5202_0_10000evt.h5"', 

46 default=None, 

47 required=True, 

48 ) 

49 parser.add_argument( 

50 "--log_file", 

51 "-l", 

52 type=Path, 

53 dest="log_file_path", 

54 help="Path to a file where this process will log its execution", 

55 ) 

56 parser.add_argument( 

57 "--log_level", 

58 type=LOGGING_LEVELS, 

59 action=EnumAction, 

60 dest="logging_level", 

61 help="Logging level: all messages with a level severity above or equal the specified level will be logged.", 

62 default=LOGGING_LEVELS.WARNING, 

63 ) 

64 parser.add_argument( 

65 "--nbins", 

66 type=int, 

67 dest="nbins", 

68 help="number of bins for the interpolation of alt az", 

69 default=10, 

70 required=False, 

71 ) 

72 parser.add_argument( 

73 "--nb_tries", 

74 type=int, 

75 dest="nb_tries", 

76 help="Number of times the software will attempt to open the input file", 

77 default=20, 

78 required=False, 

79 ) 

80 parser.add_argument( 

81 "--RA", type=float, dest="RA", help="Right Ascension of the telescope", default=None, required=True 

82 ) 

83 parser.add_argument( 

84 "--timeout", 

85 type=float, 

86 dest="open_timeout", 

87 help="Time to wait between input file opening attempts", 

88 default=0.5, 

89 required=False, 

90 ) 

91 

92 return parser 

93 

94 

95def main(): 

96 parser = online_alt_az_dl1_maker_arg_parser() 

97 args = parser.parse_args() 

98 

99 init_logging( 

100 log_header_str="RTA_online_altaz_maker", log_filename=args.log_file_path, log_level=args.logging_level 

101 ) 

102 

103 input_file = args.input_file # "dl3_LST-1.Run04190.*.fits" 

104 obs_location = EarthLocation.from_geodetic(-17.891485 * u.deg, 28.761518 * u.deg, 2187 * u.m) 

105 

106 # Failing to open a file on fefs doesn't necessarily mean we won't succeed next time ! 

107 for _ in range(args.nb_tries - 1): 107 ↛ 116line 107 didn't jump to line 116 because the loop on line 107 didn't complete

108 try: 

109 hfile = tables.open_file(input_file, mode="r+") 

110 break 

111 except Exception: 

112 print("Failed to open input file {}, re-trying in {}".format(args.input_file, args.open_timeout)) 

113 time.sleep(args.open_timeout) 

114 else: 

115 # try 1 last time, let tables error raise if failing again 

116 hfile = tables.open_file(input_file, mode="r+") 

117 

118 Trigger_time = hfile.root.dl1.event.telescope.parameters.tel_001.col("trigger_time") 

119 is_good_event_table = hfile.root.dl1.event.telescope.parameters.tel_001.col("is_good_event") 

120 selection_mask = np.logical_and(Trigger_time < 4000000000.0, is_good_event_table > 0.5) 

121 is_good_event_table = np.multiply(selection_mask, 1) 

122 

123 time_table = np.linspace(Trigger_time[selection_mask][0], Trigger_time[selection_mask][-1], args.nbins) 

124 

125 pointing_coord = SkyCoord(ra=args.RA, dec=args.DEC, unit=u.deg) 

126 event_altaz = pointing_coord.transform_to( 

127 AltAz( 

128 obstime=Time(time_table, format="unix"), 

129 location=obs_location, 

130 obswl=0.35 * u.micron, 

131 relative_humidity=0.5, 

132 temperature=10 * u.deg_C, 

133 pressure=790 * u.hPa, 

134 ) 

135 ) 

136 

137 alt_table = np.interp(Trigger_time, time_table, event_altaz.alt.rad) 

138 az_table = np.interp(Trigger_time, time_table, event_altaz.az.rad) 

139 

140 table = hfile.root.dl1.event.telescope.parameters.tel_001 

141 table.modify_column(start=0, stop=len(alt_table), step=1, column=alt_table, colname="alt_tel") 

142 table.modify_column(start=0, stop=len(az_table), step=1, column=az_table, colname="az_tel") 

143 table.modify_column( 

144 start=0, stop=len(is_good_event_table), step=1, column=is_good_event_table, colname="is_good_event" 

145 ) 

146 

147 table.flush() 

148 

149 hfile.close() 

150 

151 

152if __name__ == "__main__": 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 main()