Source code for instaseis.database_interfaces.remote_instaseis_db

#!/usr/bin/env python
# -*- coding: utf-8 -*-
Instaseis database class for remote access over HTTP.

    Lion Krischer (, 2020
    GNU Lesser General Public License, Version 3 [non-commercial/academic use]
import io
import numpy as np
import obspy
from urllib.parse import urlencode, urlparse
import requests
import warnings

from .base_instaseis_db import BaseInstaseisDB, DEFAULT_MU
from .. import (

[docs]class RemoteInstaseisDB(BaseInstaseisDB): """ Remote Instaseis database interface. """ def __init__(self, url, *args, **kwargs): """ :param url: URL to the remote Instaseis server. :type db_path: str """ self.url = url self._scheme, self._netloc, self._path = urlparse(url)[:3] self._path = self._path.strip("/") # Parse the root message of the server. try: root = self._download_url(self._get_url(path="")) except Exception as e: raise InstaseisError( "Failed to connect to remote Instaseis " "server due to: %s" % (str(e)) ) # XXX: Add Instaseis version checks! Make sure server and client are # on the same version! if "type" not in root or root["type"] != "Instaseis Remote Server": raise InstaseisError( "Instaseis server responded with invalid " "response: %s" % (str(root)) ) if root["version"] != __version__: msg = ( "Instaseis versions on server (%s) and on your local " "client (%s) differ and thus things might not work as " "expected." % (root["version"], __version__) ) warnings.warn(msg, InstaseisWarning) self._get_info() def _get_seismograms(self, source, receiver, components=("Z", "N", "E")): """ Extract seismograms for a moment tensor point source from the AxiSEM database. :param source: instaseis.Source or instaseis.ForceSource object :type source: :class:`instaseis.source.Source` or :class:`instaseis.source.ForceSource` :param receiver: instaseis.Receiver object :type receiver: :class:`instaseis.source.Receiver` :param components: a tuple containing any combination of the strings ``"Z"``, ``"N"``, ``"E"``, ``"R"``, and ``"T"`` """ # Collect parameters. params = {"components": "".join(components).upper()} # Start with the receiver. params["receiverlatitude"] = receiver.latitude params["receiverlongitude"] = receiver.longitude if receiver.depth_in_m is not None: params["receiverdepthinmeters"] = receiver.depth_in_m if params["networkcode"] = if receiver.station: params["stationcode"] = receiver.station # Do the source. params["sourcelatitude"] = source.latitude params["sourcelongitude"] = source.longitude if source.depth_in_m is not None: params["sourcedepthinmeters"] = source.depth_in_m if isinstance(source, ForceSource): params["fr"] = source.f_r params["ft"] = source.f_t params["fp"] = source.f_p elif isinstance(source, Source): params["mrr"] = source.m_rr params["mtt"] = source.m_tt params["mpp"] = source.m_pp params["mrt"] = source.m_rt params["mrp"] = source.m_rp params["mtp"] = source.m_tp else: raise NotImplementedError url = self._get_url(path="seismograms_raw", **params) r = requests.get(url) if "Instaseis-Mu" not in r.headers: # pragma: no cover warnings.warn( "Mu is not passed via the HTTP headers. Maybe some " "proxy removed it? Mu is now always the default mu.", InstaseisWarning, ) mu = DEFAULT_MU else: mu = float(r.headers["Instaseis-Mu"]) with io.BytesIO(r.content) as fh:, 0) st = # Convert back to dictionary of numpy arrays...this is a bit # redundant but plays nice with the rest of Instaseis and still # enables a REST API that serves MiniSEED files. data = {"mu": mu} for tr in st: data[[-1].upper()] = return data def _get_url(self, path, **kwargs): # Not tested in the test-suite as it would be awkward to do with the # current setup. But manually vetted and should be good. if self._path: # pragma: no cover path = "/" + self._path + "/" + path url = "%s://%s" % (self._scheme, self._netloc) if path: url += "/%s" % path if kwargs: url += "?%s" % urlencode(kwargs) return url def _download_url(self, url): """ Helper function downloading data from a URL. """ r = requests.get(url) # Not tested in test suite as it would be awkward to do. Manually # tested and should be good. if r.status_code != 200: # pragma: no cover raise InstaseisError( "Status code %i when downloading '%s'" % (r.status_code, url) ) return r.json() def _get_info(self): """ Returns a dictionary with information about the currently loaded database. """ info = self._download_url(self._get_url(path="info")) info["directory"] = self.url # Convert types lost in the translation to JSON. info["datetime"] = obspy.UTCDateTime(info["datetime"]) info["slip"] = np.array(info["slip"], dtype=np.float64) info["sliprate"] = np.array(info["sliprate"], dtype=np.float64) return info