Coverage for /usr/local/lib/python3.7/dist-packages/pyrocko/client/geofon.py : 22%

# http://pyrocko.org - GPLv3 

# 

# The Pyrocko Developers, 21st Century 

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

from __future__ import absolute_import, division 

try: 

from future.moves.urllib.request import urlopen 

except ImportError: 

from urllib2 import urlopen 

import time 

import re 

import logging 

import json 

from pyrocko import model, util 

from pyrocko.moment_tensor import MomentTensor, symmat6 

from .base_catalog import EarthquakeCatalog, NotFound 

logger = logging.getLogger('pyrocko.client.geofon') 

km = 1000. 

class Geofon(EarthquakeCatalog): 

'''Access the Geofon earthquake catalog ''' 

def __init__(self, get_moment_tensors=True): 

self.events = {} 

self._get_moment_tensors = get_moment_tensors 

def flush(self): 

self.events = {} 

def iter_event_names( 

self, 

time_range=None, 

nmax=1000, 

magmin=None, 

latmin=-90., 

latmax=90., 

lonmin=-180., 

lonmax=180.): 

logger.debug('In Geofon.iter_event_names(...)') 

dmin = time.strftime('%Y-%m-%d', time.gmtime(time_range[0])) 

dmax = time.strftime('%Y-%m-%d', time.gmtime(time_range[1]+24*60*60)) 

if magmin is None: 

magmin = '' 

else: 

magmin = '%g' % magmin 

ipage = 1 

while True: 

url = ('http://geofon.gfz-potsdam.de/eqinfo/list.php?' + '&'.join([ 

'page=%i' % ipage, 

'datemin=%s' % dmin, 

'datemax=%s' % dmax, 

'latmin=%g' % latmin, 

'latmax=%g' % latmax, 

'lonmin=%g' % lonmin, 

'lonmax=%g' % lonmax, 

'magmin=%s' % magmin, 

'fmt=geojson', 

'nmax=%i' % nmax])) 

logger.debug('Opening URL: %s' % url) 

page = urlopen(url).read() 

logger.debug('Received page (%i bytes)' % len(page)) 

events = self._parse_events_page(page) 

for ev in events: 

if ev.moment_tensor is True: 

ev.moment_tensor = self.get_mt(ev) 

if not events: 

break 

for ev in events: 

if time_range[0] <= ev.time and ev.time <= time_range[1]: 

self.events[ev.name] = ev 

yield ev.name 

ipage += 1 

def get_event(self, name): 

logger.debug('In Geofon.get_event("%s")' % name) 

if name not in self.events: 

url = 'http://geofon.gfz-potsdam.de/eqinfo/event.php' \ 

'?id=%s&fmt=geojson' % name 

logger.debug('Opening URL: %s' % url) 

page = urlopen(url).read() 

logger.debug('Received page (%i bytes)' % len(page)) 

try: 

ev = self._parse_event_page(page) 

self.events[name] = ev 

except NotFound: 

raise NotFound(url) # reraise with url 

ev = self.events[name] 

if ev.moment_tensor is True: 

ev.moment_tensor = self.get_mt(ev) 

return ev 

def get_mt(self, ev): 

syear = time.strftime('%Y', time.gmtime(ev.time)) 

url = 'http://geofon.gfz-potsdam.de/data/alerts/%s/%s/mt.txt' % ( 

syear, ev.name) 

logger.debug('Opening URL: %s' % url) 

page = urlopen(url).read() 

logger.debug('Received page (%i bytes)' % len(page)) 

return self._parse_mt_page(page) 

def _parse_mt_page(self, page): 

d = {} 

for k in 'Scale', 'Mrr', 'Mtt', 'Mpp', 'Mrt', 'Mrp', 'Mtp': 

r = k.encode('ascii')+br'\s*=?\s*(\S+)' 

m = re.search(r, page) 

if m: 

s = m.group(1).replace(b'10**', b'1e') 

d[k.lower()] = float(s) 

m = symmat6(*(d[x] for x in 'mrr mtt mpp mrt mrp mtp'.split())) 

m *= d['scale'] 

mt = MomentTensor(m_up_south_east=m) 

return mt 

def _parse_events_page(self, page, limit=None): 

j = json.loads(page.decode('utf-8')) 

events = [] 

for ifeature, feature in enumerate(j['features']): 

ev = self._json_feature_to_event(feature) 

events.append(ev) 

if limit and ifeature + 1 == limit: 

break 

return events 

def _parse_event_page(self, page): 

return self._parse_events_page(page, limit=1)[0] 

def _json_feature_to_event(self, feature): 

name = feature['id'] 

lon, lat, depth = feature['geometry']['coordinates'] 

depth *= 1000. 

properties = feature['properties'] 

magnitude = properties['mag'] 

magnitude_type = properties['magType'] 

region = properties['place'] 

tevent = util.str_to_time(properties['time'].replace('T', ' ')) 

if 'hasMT' in properties and properties['hasMT'] == 'yes' \ 

and self._get_moment_tensors: 

moment_tensor = True # flag for caller to query MT 

else: 

moment_tensor = None 

status = properties['status'][:1] 

tags = [] 

if status in 'AMC': 

tags.append('geofon_status:%s' % status) 

category = properties.get('evtype', '') 

if re.match(r'^[a-zA-Z0-9]+$', category): 

tags.append('geofon_category:%s' % category) 

ev = model.Event( 

lat=float(lat), 

lon=float(lon), 

time=tevent, 

name=name, 

depth=float(depth), 

magnitude=float(magnitude), 

magnitude_type=str(magnitude_type), 

region=str(region), 

moment_tensor=moment_tensor, 

catalog='GEOFON', 

tags=tags) 

return ev