Coverage for pesummary/gw/file/formats/GWTC1.py: 87.3%

1# Licensed under an MIT style license -- see LICENSE.md 

2 

3import os 

4 

5import h5py 

6import numpy as np 

7from pesummary.gw.file.formats.base_read import GWSingleAnalysisRead 

8from pesummary.utils.utils import logger 

9 

10__author__ = ["Charlie Hoy <charlie.hoy@ligo.org>"] 

11 

12 

13def open_GWTC1(path, path_to_samples=None, **kwargs): 

14 """Grab the parameters and samples in a bilby file 

15 

16 Parameters 

17 ---------- 

18 path: str 

19 path to the result file you wish to read in 

20 path_to_samples: str, optional 

21 path to the group containing the posterior samples you wish to load 

22 """ 

23 f = h5py.File(path, 'r') 

24 keys = list(f.keys()) 

25 if path_to_samples is not None: 

26 data = f[path_to_samples] 

27 elif "Overall_posterior" in keys or "overall_posterior" in keys: 

28 data = \ 

29 f["overall_posterior"] if "overall_posterior" in keys else \ 

30 f["Overall_posterior"] 

31 else: 

32 f.close() 

33 raise Exception( 

34 "Failed to read in result file because there was no group " 

35 "called 'Overall_posterior' or 'overall_posterior'" 

36 ) 

37 

38 parameters = list(data.dtype.names) 

39 samples = [list(i) for i in data] 

40 extra_kwargs = GWTC1.grab_extra_kwargs(path) 

41 extra_kwargs["sampler"]["nsamples"] = len(samples) 

42 prior_samples = GWTC1.grab_priors(f) 

43 version = None 

44 f.close() 

45 data = { 

46 "parameters": parameters, 

47 "samples": samples, 

48 "injection": None, 

49 "version": version, 

50 "kwargs": extra_kwargs 

51 } 

52 if len(prior_samples): 

53 data["prior"] = {"samples": prior_samples} 

54 return data 

55 

56 

57class GWTC1(GWSingleAnalysisRead): 

58 """PESummary wrapper of the GWTC1 sample release 

59 

60 Attributes 

61 ---------- 

62 path_to_results_file: str 

63 path to the results file you wish to load in with `GWTC1` 

64 pe_algorithm: str 

65 name of the algorithm used to generate the posterior samples 

66 """ 

67 def __init__(self, path_to_results_file, injection_file=None, **kwargs): 

68 super(GWTC1, self).__init__(path_to_results_file, **kwargs) 

69 self.load(self._grab_data_from_GWTC1_file) 

70 

71 @classmethod 

72 def load_file(cls, path, injection_file=None, **kwargs): 

73 if injection_file and not os.path.isfile(injection_file): 

74 raise IOError("%s does not exist" % (path)) 

75 return super(GWTC1, cls).load_file( 

76 path, injection_file=injection_file, **kwargs 

77 ) 

78 

79 @staticmethod 

80 def grab_extra_kwargs(path): 

81 """ 

82 """ 

83 return {"sampler": {}, "meta_data": {}} 

84 

85 @staticmethod 

86 def grab_priors(obj): 

87 """ 

88 """ 

89 from pesummary.utils.samples_dict import SamplesDict 

90 

91 keys = list(obj.keys()) 

92 if "prior" in keys or "priors" in keys: 

93 data = obj["prior"] if "prior" in keys else obj["priors"] 

94 parameters = list(data.dtype.names) 

95 samples = [list(i) for i in data] 

96 return SamplesDict(parameters, np.array(samples).T) 

97 logger.warning( 

98 "Failed to draw prior samples because there is not an entry for " 

99 "'prior' or 'priors' in the result file" 

100 ) 

101 return {} 

102 

103 @staticmethod 

104 def _grab_data_from_GWTC1_file(path, path_to_samples=None, **kwargs): 

105 """ 

106 """ 

107 return open_GWTC1(path, path_to_samples=path_to_samples, **kwargs) 

108 

109 @property 

110 def calibration_data_in_results_file(self): 

111 """ 

112 """ 

113 return None