Source code for obspy.db.db

# -*- coding: utf-8 -*-
"""
SQLAlchemy ORM definitions (database layout) for obspy.db.

:copyright:
    The ObsPy Development Team (devs@obspy.org)
:license:
    GNU Lesser General Public License, Version 3
    (https://www.gnu.org/copyleft/lesser.html)
"""
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)
from future.builtins import *  # NOQA

import pickle

from sqlalchemy import (Boolean, Column, DateTime, Float, ForeignKey, Integer,
                        PickleType, String)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relation
from sqlalchemy.schema import UniqueConstraint
import numpy as np

from obspy import Trace, UTCDateTime


Base = declarative_base()


[docs]class WaveformPath(Base): """ DB table containing file directories. """ __tablename__ = 'default_waveform_paths' __table_args__ = (UniqueConstraint('path'), {}) id = Column(Integer, primary_key=True) path = Column(String, nullable=False, index=True) archived = Column(Boolean, default=False) files = relation("WaveformFile", order_by="WaveformFile.id", backref="path", cascade="all, delete, delete-orphan")
[docs] def __init__(self, data={}): self.path = data.get('path')
[docs] def __repr__(self): return "<WaveformPath('%s')>" % self.path
[docs]class WaveformFile(Base): """ DB table containing waveform files. """ __tablename__ = 'default_waveform_files' __table_args__ = (UniqueConstraint('file', 'path_id'), {}) id = Column(Integer, primary_key=True) file = Column(String, nullable=False, index=True) size = Column(Integer, nullable=False) mtime = Column(Integer, nullable=False, index=True) format = Column(String, nullable=False, index=True) path_id = Column(Integer, ForeignKey('default_waveform_paths.id')) channels = relation("WaveformChannel", order_by="WaveformChannel.id", backref="file", cascade="all, delete, delete-orphan")
[docs] def __init__(self, data={}): self.file = data.get('file') self.size = data.get('size') self.mtime = int(data.get('mtime')) self.format = data.get('format')
[docs] def __repr__(self): return "<WaveformFile('%s')>" % (self.id)
[docs]class WaveformChannel(Base): """ DB table containing channels. """ __tablename__ = 'default_waveform_channels' __table_args__ = (UniqueConstraint('network', 'station', 'location', 'channel', 'file_id'), {}) id = Column(Integer, primary_key=True) file_id = Column(Integer, ForeignKey('default_waveform_files.id'), index=True) network = Column(String(2), nullable=False, index=True) station = Column(String(5), nullable=False, index=True) location = Column(String(2), nullable=False, index=True) channel = Column(String(3), nullable=False, index=True) starttime = Column(DateTime, nullable=False, index=True) endtime = Column(DateTime, nullable=False, index=True) calib = Column(Float, nullable=False) sampling_rate = Column(Float, nullable=False) npts = Column(Integer, nullable=False) preview = Column(PickleType(protocol=0), nullable=True) gaps = relation("WaveformGaps", order_by="WaveformGaps.id", backref="channel", cascade="all, delete, delete-orphan") features = relation("WaveformFeatures", order_by="WaveformFeatures.id", backref="channel", cascade="all, delete, delete-orphan")
[docs] def __init__(self, data={}): self.update(data)
[docs] def update(self, data): self.network = data.get('network', '') self.station = data.get('station', '') self.location = data.get('location', '') self.channel = data.get('channel', '') self.starttime = data.get('starttime') self.endtime = data.get('endtime') self.calib = data.get('calib', 1.0) self.npts = data.get('npts', 0) self.sampling_rate = data.get('sampling_rate', 1.0) self.preview = data.get('preview', None)
[docs] def __repr__(self): return "<WaveformChannel('%s')>" % (self.id)
[docs] def get_preview(self, apply_calibration=False): try: data = np.loads(self.preview) except Exception: data = np.array([]) if apply_calibration: data = data * self.calib tr = Trace(data=data) tr.stats.starttime = UTCDateTime(self.starttime) tr.stats.delta = 30.0 tr.stats.network = self.network tr.stats.station = self.station tr.stats.location = self.location tr.stats.channel = self.channel tr.stats.calib = self.calib tr.stats.preview = True return tr
[docs]class WaveformGaps(Base): """ DB table containing gaps. """ __tablename__ = 'default_waveform_gaps' id = Column(Integer, primary_key=True) channel_id = Column(Integer, ForeignKey('default_waveform_channels.id'), index=True) gap = Column(Boolean, nullable=False, index=True) starttime = Column(DateTime, nullable=False, index=True) endtime = Column(DateTime, nullable=False, index=True) samples = Column(Integer, nullable=False)
[docs] def __init__(self, data={}): self.gap = data.get('gap', True) self.starttime = data.get('starttime') self.endtime = data.get('endtime') self.samples = data.get('samples', 0)
[docs] def __repr__(self): return "<WaveformGaps('%s')>" % (self.id)
[docs]class WaveformFeatures(Base): """ DB table containing optional features created during indexing. """ __tablename__ = 'default_waveform_features' __table_args__ = (UniqueConstraint('channel_id', 'key'), {}) id = Column(Integer, primary_key=True) channel_id = Column(Integer, ForeignKey('default_waveform_channels.id'), index=True) key = Column(String, nullable=False, index=True) value = Column(PickleType, nullable=True)
[docs] def __init__(self, data={}): self.key = data.get('key') self.value = pickle.dumps(data.get('value', None))
[docs] def __repr__(self): return "<WaveformFeatures('%s')>" % (self.id)