121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
|
import pandas as pd
|
||
|
|
|
||
|
|
from obspy.core.event import Catalog
|
||
|
|
from obspy.core.event import Event
|
||
|
|
from obspy.core.event import Pick
|
||
|
|
from obspy.core.event import Origin
|
||
|
|
from obspy.core.event import OriginQuality
|
||
|
|
|
||
|
|
from obspy import UTCDateTime
|
||
|
|
from obspy.core.event.base import WaveformStreamID, Comment
|
||
|
|
|
||
|
|
|
||
|
|
class CatalogConverter:
|
||
|
|
"""
|
||
|
|
Class for converting SeisBench and pyOcto detection results to ObsPy Catalog, which can be saved as QuakeML.
|
||
|
|
"""
|
||
|
|
def __init__(self, config, picks, catalog_df, assignments_df, name_of_algorithm):
|
||
|
|
self._catalog_df = catalog_df
|
||
|
|
self._assignments_df = assignments_df
|
||
|
|
self._picks = picks
|
||
|
|
self._config = config
|
||
|
|
self._name_of_algorithm = name_of_algorithm
|
||
|
|
self.catalog = None
|
||
|
|
|
||
|
|
def _convert_pick(self, p):
|
||
|
|
"""
|
||
|
|
Function converts picks from SeisBench to ObsPy format
|
||
|
|
:param p: SeisBench pick
|
||
|
|
:return: ObsPy pick
|
||
|
|
"""
|
||
|
|
pick = Pick()
|
||
|
|
pick.time = UTCDateTime(p.peak_time.datetime)
|
||
|
|
pick.waveform_id = WaveformStreamID(network_code=p.trace_id.split(".")[0],
|
||
|
|
station_code=p.trace_id.split(".")[1],
|
||
|
|
channel_code=p.trace_id.split(".")[2])
|
||
|
|
if p.phase == 'P':
|
||
|
|
pick.phase_hint = self._config["P_hint"]
|
||
|
|
elif p.phase == 'S':
|
||
|
|
pick.phase_hint = self._config["S_hint"]
|
||
|
|
pick.evaluation_mode = 'automatic'
|
||
|
|
pick.evaluation_status = 'preliminary'
|
||
|
|
return pick
|
||
|
|
|
||
|
|
def _convert_origin(self, origin_sb, list_of_picks_sb):
|
||
|
|
origin = Origin()
|
||
|
|
origin.time = UTCDateTime(pd.to_datetime(origin_sb.time, unit='s').to_pydatetime())
|
||
|
|
origin.latitude = origin_sb.latitude # float
|
||
|
|
origin.longitude = origin_sb.longitude # float
|
||
|
|
origin.depth = origin_sb.depth # float in kilometers (SWIP5 origin version) down the see level
|
||
|
|
origin.depth_type = 'operator assigned'
|
||
|
|
# TODO: make sure that region is not necessary
|
||
|
|
# origin.region = self._config["region"]
|
||
|
|
origin.evaluation_mode = "automatic"
|
||
|
|
origin.evaluation_status = 'preliminary'
|
||
|
|
origin.comments.append(Comment(text=f"Localized by: {self._name_of_algorithm}", force_resource_id=False))
|
||
|
|
origin.quality = OriginQuality(used_phase_count=len(list_of_picks_sb))
|
||
|
|
return origin
|
||
|
|
|
||
|
|
def _convert_event(self, origin_sb, list_of_picks_sb):
|
||
|
|
"""
|
||
|
|
Function convert GaMMa detection to ObsPy Event
|
||
|
|
:param origin_sb:
|
||
|
|
:param list_of_picks_sb:
|
||
|
|
:return:
|
||
|
|
"""
|
||
|
|
event = Event()
|
||
|
|
for p in list_of_picks_sb:
|
||
|
|
pick = self._convert_pick(p)
|
||
|
|
event.picks.append(pick)
|
||
|
|
origin = self._convert_origin(origin_sb, list_of_picks_sb)
|
||
|
|
event.origins.append(origin)
|
||
|
|
return event
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _append_pick_trace_id(pick, stream):
|
||
|
|
"""
|
||
|
|
Function assigns channel to pick - it is useful for work with SWIP
|
||
|
|
:param pick:
|
||
|
|
:param stream:
|
||
|
|
:return:
|
||
|
|
"""
|
||
|
|
channel = stream[0].stats.channel
|
||
|
|
if pick.phase == "P":
|
||
|
|
pick.trace_id = pick.trace_id + channel[:-1] + "Z"
|
||
|
|
if pick.phase == "S":
|
||
|
|
pick.trace_id = pick.trace_id + channel[:-1] + "E"
|
||
|
|
return pick
|
||
|
|
|
||
|
|
def catalog2obspy(self):
|
||
|
|
"""
|
||
|
|
Function convert GaMMa catalog and SeisBench picks
|
||
|
|
:return: ObsPy Catalog object
|
||
|
|
"""
|
||
|
|
# TODO: make sure that resource id is necessary
|
||
|
|
#cat = Catalog(resource_id=self._config["resource_id"])
|
||
|
|
cat = Catalog()
|
||
|
|
for j, row in self._catalog_df.iterrows():
|
||
|
|
event = self._catalog_df.iloc[j]
|
||
|
|
event_picks = [self._picks[i] for i in
|
||
|
|
self._assignments_df[self._assignments_df["event_idx"] ==
|
||
|
|
event["idx"]]["pick_idx"]]
|
||
|
|
event_obspy = self._convert_event(event, event_picks)
|
||
|
|
cat.append(event_obspy)
|
||
|
|
self.catalog = cat
|
||
|
|
|
||
|
|
def save_catalog_to_file(self, file_path):
|
||
|
|
"""
|
||
|
|
Save ObsPy catalog to a file.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
file_path (str): The file path where the catalog will be saved.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
None
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
self.catalog.write(file_path, format="QUAKEML")
|
||
|
|
print(f"Catalog saved successfully to {file_path}")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error occurred while saving catalog: {e}")
|