Source code for

# - GPLv3
# The Pyrocko Developers, 21st Century
# ---|P------/S----------~Lg----------

Event data access interface (*deprecated*).

from pyrocko import model, response

import logging
import copy
import pickle

logger = logging.getLogger('')

class NoRestitution(Exception):

class FileNotFound(Exception):

    def __init__(self, s):
        self.s = s

    def __str__(self):
        return 'File not found: %s' % self.s

class Problems(object):
    def __init__(self):
        self._problems = {}

    def add(self, kind, nslct):
        if kind not in self._problems:
            self._problems[kind] = set()
        problem = self._problems[kind]

    def dump(self, fn):
        f = open(fn, 'wb')
        pickle.dump(self._problems, f)

    def load(self, fn):
        f = open(fn, 'rb')
        self._problems = pickle.load(f)

    def mapped(self, mapping=lambda nslct: nslct[:3]):
        p = {}
        for kind, problem in self._problems.items():
            nsl = set()
            for nslct in problem:
            p[kind] = nsl

        return p

[docs]class EventDataAccess(object): ''' Abstract base class for event data access (see ''' def __init__(self, events=None, stations=None, datapile=None): self._pile = datapile self._events = events if stations is None: self._stations = None else: self._stations = {} for station in stations: self._stations[station.nsl()] = station self._problems = Problems() def get_pile(self): return self._pile
[docs] def get_pyrocko_events(self): ''' Extract :py:class:`~pyrocko.model.event.Event` instances from the volume. ''' if not self._events: self._events = self._get_events_from_file() return self._events
[docs] def get_pyrocko_station(self, tr, relative_event=None): ''' Get station information for a given trace. :param tr: :py:class:`~pyrocko.trace.Trace` instance :returns: :py:class:`~pyrocko.model.station.Station` objects. ''' self._update_stations() s = copy.deepcopy(self._stations[tr.nslc_id[:3]]) if relative_event is not None: s.set_event_relative_data(relative_event) return s
[docs] def get_pyrocko_channel(self, tr): ''' Get channel information for a given trace. :param tr: :py:class:`~pyrocko.trace.Trace` instance :returns: :py:class:`~pyrocko.model.station.Channel` objects. ''' sta = self.get_station(tr) return sta.get_channel(
[docs] def get_pyrocko_stations(self): ''' Exctract a list of :py:class:`~pyrocko.model.station.Station` instances. ''' return list(self._get_stations().values())
def _get_stations(self, relative_event=None): self._update_stations() stations = copy.deepcopy(self._stations) if relative_event is not None: for s in stations.values(): s.set_event_relative_data(relative_event) return stations def _update_stations(self): if not self._stations: self._stations = {} for station in self._get_stations_from_file(): self._stations[station.nsl()] = station self._insert_channel_descriptions(self._stations) def _insert_channel_descriptions(self, stations): pile = self.get_pile() nslc_ids = pile.gather_keys( lambda tr: (, tr.station, tr.location, for nslc in nslc_ids: if nslc[:3] not in stations: logger.warning( 'No station description for trace %s.%s.%s.%s' % nslc) continue sta = stations[nslc[:3]] try: sta.add_channel(self._get_channel_description_from_file(nslc)) except FileNotFound: logger.warning( 'No channel description for trace %s.%s.%s.%s' % nslc) def _get_channel_description_from_file(self, nslc): return model.Channel(nslc[3], None, None, 1.) def iter_traces(self, group_selector=None, trace_selector=None): for traces in self.get_pile().chopper_grouped( gather=lambda tr: (, tr.station, tr.location), group_selector=group_selector, trace_selector=trace_selector): yield traces def problems(self): return self._problems def _redundant_channel_weeder(self, redundant_channel_priorities, nslcs): if redundant_channel_priorities is None: return [] # n=network,s=station,l=location,c=channel # channels by station by_nsl = {} for nslc in nslcs: nsl = nslc[:3] if nsl not in by_nsl: by_nsl[nsl] = [] by_nsl[nsl].append(nslc) # figure out what items to remove to_delete = [] for ((h1, h2), (l1, l2)) in redundant_channel_priorities: for nsl, nslcs in by_nsl.items(): channels = [nslc[3] for nslc in nslcs] if h1 in channels and \ h2 in channels and \ l1 in channels and \ l2 in channels: to_delete.append(nslc[:3] + (l1,)) to_delete.append(nslc[:3] + (l2,)) return to_delete def get_restitution(self, tr, allowed_methods): if 'integration' in allowed_methods: response.IntegrationResponse() else: raise Exception('only "integration" restitution method is allowed')