Coverage for src/susi/reduc/validation/batch_check.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3""" 

4module provides BatchCheck 

5 

6@author: hoelken 

7""" 

8from ...utils import Collections 

9from ...base import Logging 

10 

11logger = Logging.get_logger() 

12 

13 

14class BatchCheck: 

15 """ 

16 ## BatchCheck 

17 Performs (predefined) validation checks on FITS headers on batches. 

18 Takes a loaded(!) fits batch on init 

19 """ 

20 

21 def __init__(self, batch, config): 

22 #: The (loaded) batch of FITSes to process 

23 self.fb = batch 

24 #: The config data item which provides the thresholds for the checks 

25 self.config = config 

26 # Warnings found during header checks 

27 self.warnings = {} 

28 

29 def validate(self): 

30 """ 

31 Performs all available checks. 

32 If warnings are generated they can be accessed from the `self.warnings` dict. 

33 

34 :return: `True` iff all checks pass. 

35 """ 

36 valid = True 

37 valid = self.check_temp() and valid 

38 valid = self.check_timestamp() and valid 

39 valid = self.check_equal_value() and valid 

40 valid = self.check_pmu_curr() and valid 

41 valid = self.check_pmu_angle() and valid 

42 return valid 

43 

44 def check_timestamp(self): 

45 """ 

46 Check the timestamp header field 

47 The delta between two frames should not exceed the given threshold. 

48 

49 :return: `False` if any of the deltas exceeds the threshold. `True` otherwise. 

50 """ 

51 warnings = self.__check_delta(self.fb.header_field(self.config.base.timestamp_field), 

52 self.config.base.time_delta) 

53 if not warnings: 

54 return True 

55 logger.warning('Timestamp delta exceeds %s on: %s', self.config.base.time_delta, warnings) 

56 self.warnings['time'] = warnings 

57 return False 

58 

59 def check_equal_value(self): 

60 """ 

61 Check that all frames have the same values in the requested fields 

62 

63 :return: `False` if any of the entries differs. `True` otherwise. 

64 """ 

65 result = True 

66 for field_name in self.config.base.equal_header_values: 

67 warnings = self.__check_same_value(self.fb.header_field(field_name)) 

68 if warnings: 

69 logger.warning('Value for "%s" differs on: %s', field_name, warnings) 

70 self.warnings[field_name] = warnings 

71 result = False 

72 return result 

73 

74 def check_temp(self): 

75 """ 

76 Check that all frames are taken with more or less the same temperature. 

77 

78 :return: `False` if any of the entries exceeds threshold. `True` otherwise. 

79 """ 

80 temps = Collections.as_float_array(self.fb.header_field(self.config.base.temp_field)) 

81 if max(temps) - min(temps) > self.config.base.temp_threshold: 

82 logger.warning('Delta of "%s" exceeds %s', self.config.base.temp_field, self.config.base.temp_threshold) 

83 self.warnings['temp'] = True 

84 return False 

85 return True 

86 

87 def check_pmu_curr(self): 

88 """ 

89 Check the PMU current of all frames. 

90 

91 :return: `False` if any of the entries exceeds threshold. `True` otherwise. 

92 """ 

93 currents = self.fb.header_field(self.config.spol.pmu_cur_field) 

94 warnings = self.__check_threshold(currents, self.config.spol.pmu_cur_threshold) 

95 if warnings: 

96 logger.warning('"%s" exceeds threshold on %s', self.config.spol.pmu_cur_field, warnings) 

97 self.warnings[self.config.spol.pmu_cur_field] = warnings 

98 return False 

99 return True 

100 

101 def check_pmu_angle(self): 

102 """ 

103 Check the PMU error of all frames. 

104 

105 :return: `False` if any of the entries exceeds threshold. `True` otherwise. 

106 """ 

107 error_values = self.fb.header_field(self.config.spol.pmu_ang_error_field) 

108 warnings = self.__check_threshold(error_values, self.config.spol.pmu_ang_threshold) 

109 if warnings: 

110 logger.warning('"%s" exceeds threshold on %s', self.config.spol.pmu_ang_error_field, warnings) 

111 self.warnings[self.config.spol.pmu_ang_error_field] = warnings 

112 return False 

113 return True 

114 

115 def __check_threshold(self, data, threshold): 

116 warnings = [] 

117 for i in range(len(data)): 

118 float(data[i]) > threshold and warnings.append(self.fb.file_by(i)) 

119 return warnings 

120 

121 def __check_delta(self, data, threshold): 

122 warnings = [] 

123 for i in range(len(data) - 1): 

124 if abs(float(data[i + 1]) - float(data[i])) > threshold: 

125 warnings.append(self.fb.file_by(i + 1)) 

126 return warnings 

127 

128 def __check_same_value(self, data): 

129 warnings = [] 

130 for i in range(len(data) - 1): 

131 data[i + 1] != data[i] and warnings.append(self.fb.file_by(i + 1)) 

132 return warnings