Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/dataset/volcanoes.py: 95%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

4# ---|P------/S----------~Lg---------- 

5 

6''' 

7Volcanoes of the World dataset from the Smithsonian Institution (`VOTW 

8<https://volcano.si.edu/gvp_votw.cfm>`_). 

9''' 

10 

11import csv 

12import logging 

13import numpy as num 

14from os import path as op 

15from collections import OrderedDict 

16 

17from pyrocko import config, util 

18from .util import get_download_callback 

19 

20logger = logging.getLogger('volcanoes') 

21 

22citation = ''' 

23Global Volcanism Program, 2013. Volcanoes of the World, 

24v. 4.8.5. Venzke, E (ed.). Smithsonian Institution. Downloaded 29 Jan 2020. 

25https://doi.org/10.5479/si.GVP.VOTW4-2013 ''' 

26 

27 

28__doc__ += ''' 

29 

30.. rubric:: Citation 

31''' + citation 

32 

33 

34class Volcano(object): 

35 age = None 

36 

37 __fields__ = OrderedDict() 

38 __slots__ = list(__fields__.keys()) 

39 

40 def __init__(self, *args): 

41 for (attr, attr_type), value in zip(self.__fields__.items(), args): 

42 if attr_type in (int, float): 

43 if not value or value == '?': 

44 value = 0. 

45 try: 

46 setattr(self, attr, attr_type(value)) 

47 except ValueError as e: 

48 print(list(zip(self.__fields__.keys(), args))) 

49 raise e 

50 

51 def __str__(self): 

52 d = {attr: getattr(self, attr) for attr in self.__fields__.keys()} 

53 return '\n'.join(['%s: %s' % (attr, val) for attr, val in d.items()]) 

54 

55 

56class VolcanoHolocene(Volcano): 

57 age = 'holocene' 

58 

59 __fields__ = OrderedDict([ 

60 ('fid', str), 

61 ('volcano_number', int), 

62 ('volcano_name', str), 

63 ('primary_volcano_type', str), 

64 ('last_eruption_year', int), 

65 ('country', str), 

66 ('geological_summary', str), 

67 ('region', str), 

68 ('subregion', str), 

69 ('lat', float), 

70 ('lon', float), 

71 ('elevation', float), 

72 ('tectonic_setting', str), 

73 ('geologic_epoch', str), 

74 ('evidence_category', str), 

75 ('primary_photo_link', str), 

76 ('primary_photo_caption', str), 

77 ('primary_photo_credit', str), 

78 ('major_rock_type', str), 

79 ('geolocation', str), 

80 ]) 

81 

82 

83class VolcanoPleistocene(Volcano): 

84 age = 'pleistocene' 

85 

86 __fields__ = OrderedDict([ 

87 ('fid', str), 

88 ('volcano_number', int), 

89 ('volcano_name', str), 

90 ('primary_volcano_type', str), 

91 ('country', str), 

92 ('geological_summary', str), 

93 ('region', str), 

94 ('subregion', str), 

95 ('lat', float), 

96 ('lon', float), 

97 ('elevation', float), 

98 ('geologic_epoch', str), 

99 ('geolocation', str), 

100 ]) 

101 

102 

103class Volcanoes(object): 

104 ''' 

105 Volcanoes of the World. 

106 ''' 

107 URL_HOLOCENE = 'https://mirror.pyrocko.org/smithsonian/smithsonian-holocene.csv' # noqa 

108 URL_PLEISTOCENE = 'https://mirror.pyrocko.org/smithsonian/smithsonian-pleistocene.csv' # noqa 

109 

110 def __init__(self): 

111 self.fname_holocene = op.join( 

112 config.config().volcanoes_dir, 'smithsonian-holocene.csv') 

113 self.fname_pleistocene = op.join( 

114 config.config().volcanoes_dir, 'smithsonian-pleistocene.csv') 

115 

116 if not op.exists(self.fname_holocene) \ 

117 or not op.exists(self.fname_pleistocene): 

118 self.download() 

119 

120 self.volcanoes = [] 

121 self._load_volcanoes(self.fname_holocene, VolcanoHolocene) 

122 self._load_volcanoes(self.fname_pleistocene, VolcanoPleistocene) 

123 

124 def _load_volcanoes(self, fname, cls): 

125 with open(fname, 'r', encoding='utf8') as f: 

126 next(f) # skip header 

127 reader = csv.reader(f, dialect='unix') 

128 for row in reader: 

129 volcano = cls(*row) 

130 self.volcanoes.append(volcano) 

131 logger.debug('loaded %d volcanoes', self.nvolcanoes) 

132 

133 def download(self): 

134 logger.info('Downloading Holocene volcanoes...') 

135 util.download_file( 

136 self.URL_HOLOCENE, self.fname_holocene, 

137 status_callback=get_download_callback( 

138 'Downloading Holocene volcanoe database...')) 

139 

140 logger.info('Downloading Pleistocene volcanoes...') 

141 util.download_file( 

142 self.URL_PLEISTOCENE, self.fname_pleistocene, 

143 status_callback=get_download_callback( 

144 'Downloading Pleistocene volcanoe database...')) 

145 

146 @property 

147 def volcanoes_holocene(self): 

148 return [v for v in self.volcanoes if isinstance(v, VolcanoHolocene)] 

149 

150 @property 

151 def volcanoes_pleistocene(self): 

152 return [v for v in self.volcanoes if isinstance(v, VolcanoPleistocene)] 

153 

154 @property 

155 def nvolcanoes(self): 

156 return len(self.volcanoes) 

157 

158 @property 

159 def nvolcanoes_holocene(self): 

160 return len(self.volcanoes_holocene) 

161 

162 @property 

163 def nvolcanoes_pleistocene(self): 

164 return len(self.volcanoes_pleistocene) 

165 

166 def get_coords(self): 

167 return num.array([(v.lat, v.lon) for v in self.volcanoes]) 

168 

169 def get_names(self): 

170 return [v.volcano_name for v in self.volcanoes] 

171 

172 

173if __name__ == '__main__': 

174 logging.basicConfig(level=logging.DEBUG) 

175 volc = Volcanoes() 

176 print(volc.volcanoes[0])