Coverage for pygeodyn/inout/priors.py: 33%

1#-*- coding: utf-8 -*- 

2""" Contains the reading methods for prior types """ 

3import h5py 

4import os.path 

5import numpy as np 

6 

7def read_0_path(data_directory, prior_type, dt_sampling, measures=None): 

8 """ 

9 Reads prior data under the 0 path format. 

10 

11 :param data_directory: directory where the prior data is stored 

12 :type data_directory: str 

13 :param prior_type: type of the prior data 

14 :type prior_type: str 

15 :param dt_sampling: smoothing window duration in second 

16 :type dt_sampling: float 

17 :param measures: sequence of string saying which measures should be extracted. Allowed 

18 values are 'times', 'B', 'U' and 'ER', order do not matter. 

19 :return: MF, U, ER, times, dt_samp, tag 

20 :rtype: lists 

21 """ 

22 allowed_measures = ('times', 'B', 'U', 'ER') 

23 if measures is None: 

24 measures = allowed_measures 

25 

26 err_mess = """Invalid measure given in measures, {0}, allowed values are {1}. 

27 """.format(measures, allowed_measures) 

28 assert any([m in measures for m in allowed_measures]), err_mess 

29 

30 # Read the prior data  

31 MF, U, ER, times, dt_samp, tag = [], [], [], [], [], [] 

32 for m in measures: 

33 if m == 'times': 

34 times.append(None) 

35 measure_fn = 'Real' + m + '.dat' 

36 if m == 'U': 

37 # Need to change the sign convention for U prior if read from Coupled-Earth 

38 U.append((-1) * np.genfromtxt(os.path.join(data_directory, measure_fn))) 

39 elif m == 'B': 

40 MF.append(np.genfromtxt(os.path.join(data_directory, measure_fn))) 

41 elif m == 'ER': 

42 ER.append(np.genfromtxt(os.path.join(data_directory, measure_fn))) 

43 

44 dt_samp.append(None) 

45 tag.append(prior_type) 

46 return MF, U, ER, times, dt_samp, tag 

47 

48 

49def read_50_and_71_path(data_directory, prior_type, dt_sampling, measures=None): 

50 """ 

51 Reads prior data under the 50% or 71% path format. 

52 

53 :param data_directory: directory where the prior data is stored 

54 :type data_directory: str 

55 :param prior_type: type of the prior data 

56 :type prior_type: str 

57 :param dt_sampling: smoothing window duration in second 

58 :type dt_sampling: float 

59 :param measures: sequence of string saying which measures should be extracted. Allowed 

60 values are 'times', 'B', 'U' and 'ER', order do not matter. 

61 :return: MF, U, ER, times, dt_samp, tag 

62 :rtype: lists 

63 """ 

64 if measures is None: 

65 measures = ('times', 'B', 'U', 'ER') 

66 

67 err_mess = """Invalid measure given in measures, {0}, allowed values are {1}. 

68 """.format(measures, ('times', 'B', 'U', 'ER')) 

69 assert any([m in measures for m in ('times', 'B', 'U', 'ER')]), err_mess 

70 

71 measures = ['MF' if m == 'B' else m for m in measures] 

72 

73 MF, U, ER, times, dt_samp, tag = [], [], [], [], [], [] 

74 with h5py.File(os.path.join(data_directory, 'Real.hdf5'), 'r') as hdf_file: 

75 times.append(np.array(hdf_file["times"])) 

76 if prior_type == "71path": 

77 MF.append(-np.array(hdf_file["MF"])) 

78 else: 

79 MF.append(np.array(hdf_file["MF"])) 

80 U.append(np.array(hdf_file["U"])) 

81 ER.append(np.array(hdf_file["ER"])) 

82 tag.append(prior_type) 

83 dt_samp.append(dt_sampling) 

84 return MF, U, ER, times, dt_samp, tag 

85 

86 

87def read_100_path(data_directory, prior_type, dt_sampling, measures=None): 

88 """ 

89 Reads prior data under 100% path format. 

90 

91 :param data_directory: directory where the prior data is stored 

92 :type data_directory: str 

93 :param prior_type: type of the prior data 

94 :type prior_type: str 

95 :param dt_sampling: smoothing window duration in second 

96 :type dt_sampling: float 

97 :param measures: sequence of string saying which measures should be extracted. Allowed 

98 values are 'times', 'B', 'U' and 'ER', order do not matter. 

99 :return: MF, U, ER, times, dt_samp, tag 

100 :rtype: lists 

101 """ 

102 if measures is None: 

103 measures = ('times', 'B', 'U', 'ER') 

104 

105 err_mess = """Invalid measure given in measures, {0}, allowed values are {1}. 

106 """.format(measures, ('times', 'B', 'U', 'ER')) 

107 assert any([m in measures for m in ('times', 'B', 'U', 'ER')]), err_mess 

108 

109 measures = ['MF' if m == 'B' else m for m in measures] 

110 

111 # list of 100path time series 

112 R = ["Real_A1.hdf5", 

113 "Real_A2.hdf5", 

114 "Real_A3.hdf5", 

115 "Real_A4.hdf5", 

116 "Real_B1.hdf5", 

117 "Real_B2.hdf5", 

118 "Real_B3_clean_1.hdf5", 

119 "Real_B3_clean_2.hdf5", 

120 "Real_B4.hdf5"] 

121 

122 #loop over time series 

123 MF, U, ER, times, dt_samp, tag = [], [], [], [], [], [] 

124 for i in range(9): 

125 with h5py.File(os.path.join(data_directory, R[i]), 'r') as hdf_file: 

126 times.append(np.array(hdf_file["times"][50:])) 

127 MF.append(np.array(hdf_file["MF"][50:])) 

128 U.append(np.array(hdf_file["U"][50:])) 

129 ER.append(np.array(hdf_file["ER"][50:])) 

130 dt = hdf_file["times"][1] - hdf_file["times"][0] 

131 tag.append(prior_type) 

132 dt_samp.append(dt) 

133 

134 #add 71path prior data 

135 MF_71p, U_71p, ER_71p, times_71p, dt_samp_71p, tag_71p = read_50_and_71_path(os.path.join(data_directory, '../71path'),'71path', dt_sampling) 

136 

137 #concatenate lists 

138 MF = MF_71p + MF 

139 U = U_71p + U 

140 ER = ER_71p + ER 

141 times = times_71p + times 

142 tag = tag_71p + tag 

143 dt_samp = dt_samp_71p + dt_samp 

144 

145 return MF, U, ER, times, dt_samp, tag