Source code for postprocessing_seismo_lib.conversions
import json
import os
import random
import uuid
import importlib.resources
import traceback
from datetime import datetime
import pandas as pd
from anssformats.pick import Pick
from anssformats.channel import Channel, ChannelProperties
from anssformats.geojson import PointGeometry
from anssformats.source import Source
from anssformats.filter import Filter
from anssformats.amplitude import Amplitude
from anssformats.associationpickinfo import AssociationPickInfo
from anssformats.analytics import Analytics, Prediction
from anssformats.detection import Detection
from anssformats.hypocenter import Hypocenter, HypocenterProperties
from anssformats.magnitude import Magnitude
'''
This module targets anss-formats==0.1.3.
anss-formats==0.1.3 removed several classes/fields that anss-formats==0.1.0 relied on:
0.1.0 field/class 0.1.3 replacement
-------------------------------------- --------------------------------------------------
Pick.pickerType Pick.method (now a free string, no enum)
Pick.qualityInfo / Pick.machineLearningInfo removed; folded into Pick.analyticsInfo
(Analytics / Prediction)
anssformats.association.Association anssformats.associationpickinfo.AssociationPickInfo
(phase / source / method only)
polarity: "up" / "down" only polarity: "up" / "down" / "no-result"
(none) new optional "onset": "impulsive" / "emergent" /
"questionable"
(none) new Detection / Hypocenter / HypocenterProperties /
Magnitude / LocationPickInfo
The SOA pick format itself (Site/Source/Time/Phase/Polarity/Picker/Onset/Filter/Amplitude/
Quality, all capitalized keys) has not changed.
'''
# =============================================================================
# Shared helpers
# =============================================================================
def _default_station_csv_path():
return importlib.resources.files(
'postprocessing_seismo_lib.example_data'
).joinpath('archive_stations_loc_dates.csv')
def _load_station_lookup(station_csv_path=None):
"""Build a (NET, STA, CHA) -> (LON, LAT, ELEV) lookup.
If station_csv_path is not provided, falls back to the archive_stations_loc_dates.csv
bundled with the library under postprocessing_seismo_lib/example_data/.
"""
if station_csv_path is None:
station_csv_path = _default_station_csv_path()
station_metadata_df = pd.read_csv(station_csv_path)
if station_metadata_df is None or station_metadata_df.empty:
raise RuntimeError("Station metadata is empty or not loaded. Cannot continue conversion.")
return {
(row.NET, str(row.STA), row.CHA): (row.LON, row.LAT, row.ELEV)
for _, row in station_metadata_df.iterrows()
}
# SOA producers disagree on how they stringify pick times: the legacy archive dump uses a
# space-separated, non-ISO format, while the Phasenet lambda and pick-filter-spec emit proper
# ISO8601 with a "Z" suffix.
_SOA_TIME_FORMATS = (
"%Y-%m-%dT%H:%M:%S.%fZ",
"%Y-%m-%d %H:%M:%S.%f",
"%Y-%m-%dT%H:%M:%SZ",
)
def _parse_soa_time(time_str):
for fmt in _SOA_TIME_FORMATS:
try:
return datetime.strptime(time_str, fmt)
except ValueError:
continue
raise ValueError(f"Unrecognized SOA time format: {time_str!r}")
def _build_channel(site, station_lookup):
lon, lat, elev = station_lookup.get(
(site["Network"], str(site["Station"]), site["Channel"]),
(None, None, None),
)
if lon is None or lat is None or elev is None:
print(f"Missing coordinates for {site}, defaulting to 0.0")
lon = lon if lon is not None else 0.0
lat = lat if lat is not None else 0.0
elev = elev if elev is not None else 0.0
return Channel(
geometry=PointGeometry(coordinates=[float(lon), float(lat), float(elev)]),
properties=ChannelProperties(
station=site["Station"],
channel=site["Channel"],
network=site["Network"],
location=site["Location"],
),
)
def _build_analytics(quality_entries, phase, source):
"""Fold an SOA "Quality" list into a 0.1.3 Analytics/Prediction object.
PhaseNet-standard entries carry a [0,1] confidence, so they map onto Prediction.probability.
Other standards (e.g. hypoinverse pick-weight codes 0-4) don't fit that range, so they're
carried as the prediction's value instead, under label "pickWeight".
"""
predictions = []
for q in quality_entries or []:
standard = q.get("Standard")
value = q.get("Value")
if standard == "PhaseNet":
predictions.append(
Prediction(
label="phase",
value=phase,
probability=value,
modelID=standard,
source=source,
)
)
else:
predictions.append(
Prediction(
label="pickWeight",
value=value,
modelID=standard,
source=source,
)
)
if not predictions:
return None
return Analytics(predictions=predictions)
def _soa_pick_to_pick_obj(pick_used, station_lookup, id_):
"""Build a single anss-formats==0.1.3 Pick object from one SOA-format pick dict."""
site = pick_used["Site"]
source = Source(
agencyID=pick_used["Source"]["AgencyID"],
author=pick_used["Source"]["Author"],
)
time_val = _parse_soa_time(pick_used["Time"])
phase = pick_used["Phase"]
polarity = pick_used.get("Polarity")
if polarity is not None:
polarity = str(polarity).lower()
if polarity not in ("up", "down", "no-result"):
polarity = None
onset = pick_used.get("Onset")
if onset not in ("impulsive", "emergent", "questionable"):
onset = None
# "method" replaces 0.1.0's "pickerType" and is a free string in 0.1.3 (no enum), so unlike
# the old soa_to_anss_pick_format there is no "deep-learning"/"machine-learning" -> "other"
# fallback needed here.
method = pick_used.get("Picker")
if method is not None and not str(method).strip():
method = None
filter_info = [
Filter(
type=f["Type"],
highPass=f.get("HighPass"),
lowPass=f.get("LowPass"),
units="Hertz",
)
for f in (pick_used.get("Filter") or [])
] or None
amplitude_info = None
amp = pick_used.get("Amplitude")
if isinstance(amp, dict) and amp:
amp_val = amp.get("Amplitude")
snr_val = amp.get("SNR")
amplitude_info = Amplitude(
amplitude=amp_val,
snr=max(0.0, snr_val) if snr_val is not None else None,
)
association_info = AssociationPickInfo(
phase=phase,
source=source,
method=method,
)
analytics_info = _build_analytics(pick_used.get("Quality"), phase, source)
return Pick(
type="Pick",
id=str(id_),
channel=_build_channel(site, station_lookup),
source=source,
time=time_val,
phase=phase,
polarity=polarity,
onset=onset,
method=method,
filterInfo=filter_info,
amplitudeInfo=amplitude_info,
associationInfo=association_info,
analyticsInfo=analytics_info,
)
def _soa_pick_to_anss_pick(pick_used, station_lookup, id_):
"""Build a single anss-formats==0.1.3 Pick dict from one SOA-format pick dict.
model_dump_json() (not model_dump()) is used so the CustomDT serializer runs and "time"
comes out as ISO8601 "YYYY-MM-DDTHH:MM:SS.SSSZ" per the format spec, rather than a raw
datetime repr.
"""
pick = _soa_pick_to_pick_obj(pick_used, station_lookup, id_)
return json.loads(pick.model_dump_json())
def _extract_pick_list(source):
"""Unwrap a lambda/API response envelope ({"body": [...]}) down to a bare pick list.
Accepts a file path (to either an envelope or a bare list), an already-loaded envelope dict,
or an already-loaded bare list.
"""
if isinstance(source, (str, os.PathLike)):
with open(source, 'r') as f:
source = json.load(f)
if isinstance(source, dict) and "body" in source:
return source["body"]
return source
# =============================================================================
# Core conversion functions
# =============================================================================
[docs]
def anss_to_soa_pick_format(anss_file_path):
"""
Convert ANSS-formatted (anss-formats==0.1.3) picks back to SOA format.
Parameters:
-----------
anss_file_path : str
Path to the JSON file containing ANSS picks.
Returns:
--------
dict
{
"success": True/False,
"num_picks": int,
"num_errors": int,
"picks": list of SOA-format picks,
"errors": list of errors with traceback
}
"""
soa_picks = []
errors = []
# --- Load ANSS pick file ---
try:
with open(anss_file_path, 'r') as f:
anss_picks = json.load(f)
except Exception as e:
return {
"success": False,
"stage": "load_anss_file",
"error": str(e),
"traceback": traceback.format_exc()
}
# --- Process each ANSS pick ---
for idx, pick in enumerate(anss_picks):
try:
# =========================================================
# Channel info
# =========================================================
channel = pick.get("channel", {})
properties = channel.get("properties", {})
station = properties.get("station", "")
network = properties.get("network", "")
channel_code = properties.get("channel", "")
location = properties.get("location", "")
# =========================================================
# Source info
# =========================================================
source = pick.get("source", {})
agencyID = source.get("agencyID", "")
author = source.get("author", "")
# =========================================================
# Time formatting
# =========================================================
time_val = pick.get("time")
if isinstance(time_val, str):
time_str = time_val
elif hasattr(time_val, "isoformat"):
time_str = time_val.isoformat() + "Z"
else:
time_str = str(time_val) + "Z"
# =========================================================
# Required SOA fields
# =========================================================
phase = pick.get("phase", "")
soa_pick = {
"Type": "Pick",
"Site": {
"Station": station,
"Channel": channel_code,
"Network": network,
"Location": location
},
"Time": time_str,
"Source": {
"AgencyID": agencyID,
"Author": author
},
"Phase": phase,
}
# =========================================================
# Optional fields
# =========================================================
# --- Polarity --- (0.1.3 allows "up", "down", or "no-result")
polarity = pick.get("polarity")
if polarity is not None:
polarity = str(polarity).lower()
if polarity in ("up", "down", "no-result"):
soa_pick["Polarity"] = polarity
# --- Onset --- (new in 0.1.3; has no 0.1.0 equivalent)
onset = pick.get("onset")
if onset in ("impulsive", "emergent", "questionable"):
soa_pick["Onset"] = onset
# --- Picker (was pick.get("pickerType") in 0.1.0; "method" in 0.1.3) ---
method = pick.get("method")
if method is not None and str(method).strip():
soa_pick["Picker"] = str(method)
# --- Filter ---
filter_info_raw = pick.get("filterInfo", [])
filter_info = []
if isinstance(filter_info_raw, list):
for f in filter_info_raw:
filter_dict = {}
if "type" in f and f["type"] is not None:
filter_dict["Type"] = f["type"]
if "highPass" in f and f["highPass"] is not None:
filter_dict["HighPass"] = f["highPass"]
if "lowPass" in f and f["lowPass"] is not None:
filter_dict["LowPass"] = f["lowPass"]
if filter_dict:
filter_info.append(filter_dict)
if filter_info:
soa_pick["Filter"] = filter_info
# --- Amplitude ---
amp_info = pick.get("amplitudeInfo")
if isinstance(amp_info, dict):
amplitude_dict = {}
amplitude = amp_info.get("amplitude")
snr = amp_info.get("snr")
if amplitude is not None:
amplitude_dict["Amplitude"] = amplitude
if snr is not None:
amplitude_dict["SNR"] = snr
if amplitude_dict:
soa_pick["Amplitude"] = amplitude_dict
# --- Quality --- (0.1.0 read this from qualityInfo/machineLearningInfo; those
# classes are gone in 0.1.3 -- the same information now lives in
# analyticsInfo.predictions, so reconstruct SOA Quality entries from there instead)
quality_info = []
analytics_info = pick.get("analyticsInfo") or {}
for prediction in analytics_info.get("predictions") or []:
model_id = prediction.get("modelID")
label = prediction.get("label")
if label == "phase":
value = prediction.get("probability")
elif label == "pickWeight":
value = prediction.get("value")
else:
value = prediction.get("probability", prediction.get("value"))
if model_id is not None and value is not None:
quality_info.append({"Standard": model_id, "Value": value})
if quality_info:
soa_pick["Quality"] = quality_info
soa_picks.append(soa_pick)
except Exception as e:
errors.append({
"index": idx,
"pick": pick,
"error": str(e),
"traceback": traceback.format_exc()
})
return {
"success": True,
"num_picks": len(soa_picks),
"num_errors": len(errors),
"picks": soa_picks,
"errors": errors
}
[docs]
def create_channel(CHANNEL_GEOMETRY, CHANNEL_PROPERTIES):
newChannel = Channel(
geometry=CHANNEL_GEOMETRY, properties=CHANNEL_PROPERTIES
)
return newChannel
[docs]
def soa_to_anss_pick_format(pick_file_path, station_csv_path=None):
"""
Convert a list of SOA-format picks (in a JSON file) into anss-formats==0.1.3 Pick dicts.
Parameters:
-----------
pick_file_path : str
Path to the JSON file containing a bare list of SOA-format picks.
station_csv_path : str, optional
Path to a station metadata CSV (NET, STA, CHA, LON, LAT, ELEV columns). If not provided,
the archive_stations_loc_dates.csv bundled with this library is used.
Returns:
--------
dict
{"success", "num_picks", "num_errors", "picks", "errors"}
"""
modified_anss_picks = []
errors = []
# --- Load pick file ---
try:
with open(pick_file_path, 'r') as f:
pick_list = json.load(f)
except Exception as e:
return {
"success": False,
"stage": "load_pick_file",
"error": str(e),
"traceback": traceback.format_exc()
}
# --- Load station metadata ---
try:
station_lookup = _load_station_lookup(station_csv_path)
except Exception as e:
return {
"success": False,
"stage": "load_station_csv",
"error": str(e),
"traceback": traceback.format_exc()
}
# --- Process picks ---
for idx, pick_used in enumerate(pick_list):
try:
rand_num = random.randint(1, 10000)
id_ = f"PhasenetPick_{rand_num}"
modified_anss_picks.append(_soa_pick_to_anss_pick(pick_used, station_lookup, id_))
except Exception as e:
errors.append({
"index": idx,
"pick": pick_used,
"error": str(e),
"traceback": traceback.format_exc()
})
return {
"success": True,
"num_picks": len(modified_anss_picks),
"num_errors": len(errors),
"picks": modified_anss_picks,
"errors": errors
}
[docs]
def phasenet_soa_to_anss_pick_format(soa_source, station_csv_path=None):
"""
[FIRST] Convert SOA picks from the Phasenet lambda's response into anss-formats==0.1.3
Pick dicts.
Parameters:
-----------
soa_source :
One of:
- a path to a lambda HTTP response envelope JSON file
({"status", "headers", "body": [<SOA pick>, ...]})
- a path to a JSON file containing a bare list of SOA picks
- an already-loaded envelope dict, or an already-loaded bare list of SOA picks
station_csv_path : str, optional
Path to a station metadata CSV. If not provided, the archive_stations_loc_dates.csv
bundled with this library is used.
Returns:
--------
dict
{"success", "num_picks", "num_errors", "picks", "errors"}
"""
try:
soa_picks = _extract_pick_list(soa_source)
except Exception as e:
return {
"success": False,
"stage": "load_pick_file",
"error": str(e),
"traceback": traceback.format_exc()
}
try:
station_lookup = _load_station_lookup(station_csv_path)
except Exception as e:
return {
"success": False,
"stage": "load_station_csv",
"error": str(e),
"traceback": traceback.format_exc()
}
anss_picks = []
errors = []
for idx, pick_used in enumerate(soa_picks):
try:
id_ = f"PhasenetPick_{idx}_{random.randint(1, 10000)}"
anss_picks.append(_soa_pick_to_anss_pick(pick_used, station_lookup, id_))
except Exception as e:
errors.append({
"index": idx,
"pick": pick_used,
"error": str(e),
"traceback": traceback.format_exc()
})
return {
"success": True,
"num_picks": len(anss_picks),
"num_errors": len(errors),
"picks": anss_picks,
"errors": errors
}
[docs]
def pickfilter_soa_to_anss_pick_format(pickfilter_source, station_csv_path=None):
"""
[SECOND] Convert a list of SOA picks returned by pick-filter-spec's PickProcess/v2
endpoint (a "localpick"-style request) into anss-formats==0.1.3 Pick dicts.
Unlike a Phasenet-only pick, a pick-filter pick typically carries an "Onset" and a
2-entry "Quality" list (PhaseNet probability + Hypoinverse pick-weight code), both of
which are preserved here (onset -> Pick.onset, Quality -> Pick.analyticsInfo.predictions).
Parameters:
-----------
pickfilter_source :
One of:
- a path to the PickProcess/v2 response envelope JSON file
({"status", "headers", "body": [<SOA pick>, ...]})
- a path to a JSON file containing a bare list of SOA picks
- an already-loaded envelope dict, or an already-loaded bare list of SOA picks
station_csv_path : str, optional
Path to a station metadata CSV. If not provided, the archive_stations_loc_dates.csv
bundled with this library is used.
Returns:
--------
dict
{"success", "num_picks", "num_errors", "picks", "errors"}
"""
try:
pick_entries = _extract_pick_list(pickfilter_source)
except Exception as e:
return {
"success": False,
"stage": "load_pick_file",
"error": str(e),
"traceback": traceback.format_exc()
}
try:
station_lookup = _load_station_lookup(station_csv_path)
except Exception as e:
return {
"success": False,
"stage": "load_station_csv",
"error": str(e),
"traceback": traceback.format_exc()
}
anss_picks = []
errors = []
for idx, entry in enumerate(pick_entries):
try:
anss_picks.append(_soa_pick_to_anss_pick(entry, station_lookup, str(uuid.uuid4())))
except Exception as e:
errors.append({
"index": idx,
"pick": entry,
"error": str(e),
"traceback": traceback.format_exc()
})
return {
"success": True,
"num_picks": len(anss_picks),
"num_errors": len(errors),
"picks": anss_picks,
"errors": errors
}
[docs]
def pickfilter_soa_to_anss_detection_format(detection_source, station_csv_path=None):
"""
[THIRD] Convert an SOA-format detection returned by pick-filter-spec's PickProcess/v2
endpoint (a "localdetection"-style request) into a single anss-formats==0.1.3
Detection dict.
Parameters:
-----------
detection_source :
One of:
- a path to the PickProcess/v2 response envelope JSON file
({"status", "headers", "body": {Type, ID, Source, Hypocenter, Magnitude, Data}})
- a path to a JSON file containing the bare detection body dict
- an already-loaded envelope dict, or an already-loaded bare detection body dict
station_csv_path : str, optional
Path to a station metadata CSV. If not provided, the archive_stations_loc_dates.csv
bundled with this library is used.
Returns:
--------
dict
{"success", "detection", "num_picks", "num_errors", "errors"}
"""
try:
if isinstance(detection_source, (str, os.PathLike)):
with open(detection_source, 'r') as f:
detection_source = json.load(f)
body = detection_source["body"] if isinstance(detection_source, dict) and "body" in detection_source else detection_source
except Exception as e:
return {
"success": False,
"stage": "load_detection_file",
"error": str(e),
"traceback": traceback.format_exc()
}
try:
station_lookup = _load_station_lookup(station_csv_path)
except Exception as e:
return {
"success": False,
"stage": "load_station_csv",
"error": str(e),
"traceback": traceback.format_exc()
}
errors = []
pick_data = []
for idx, entry in enumerate(body.get("Data", [])):
try:
pick_data.append(_soa_pick_to_pick_obj(entry, station_lookup, str(uuid.uuid4())))
except Exception as e:
errors.append({
"index": idx,
"pick": entry,
"error": str(e),
"traceback": traceback.format_exc()
})
try:
source_entry = body["Source"]
source = Source(agencyID=source_entry["AgencyID"], author=source_entry["Author"])
hypo_entry = body["Hypocenter"]
hypocenter = Hypocenter(
geometry=PointGeometry(
coordinates=[
float(hypo_entry["Longitude"]),
float(hypo_entry["Latitude"]),
float(hypo_entry["Depth"]),
]
),
properties=HypocenterProperties(
originTime=_parse_soa_time(hypo_entry["Time"]),
),
)
magnitude_data = [
Magnitude(
value=m["Value"],
type=m["Type"],
source=Source(agencyID=source_entry["AgencyID"], author=m["Author"]),
)
for m in body.get("Magnitude", [])
] or None
detection = Detection(
type="Detection",
id=body["ID"],
source=source,
hypocenter=hypocenter,
pickData=pick_data,
magnitudeData=magnitude_data,
)
detection_dict = json.loads(detection.model_dump_json())
except Exception as e:
return {
"success": False,
"stage": "build_detection",
"error": str(e),
"traceback": traceback.format_exc()
}
return {
"success": True,
"detection": detection_dict,
"num_picks": len(pick_data),
"num_errors": len(errors),
"errors": errors
}
[docs]
def phasenet_csv_to_soa_pick_format(picks_file, highpass_filt):
try:
if not os.path.exists(picks_file):
raise FileNotFoundError(f"File not found: {picks_file}")
picks_df = pd.read_csv(picks_file)
except Exception as e:
return {
"status": "error",
"stage": "file_read",
"message": str(e),
"traceback": traceback.format_exc()
}
pick_list = []
errors = []
for idx, row in picks_df.iterrows():
try:
# --- ID parsing ---
id = row['station_id']
id_split = id.split('.')
if len(id_split) < 4:
raise ValueError(f"Invalid station_id format: {id}")
net = id_split[0]
sta = id_split[1]
loc = id_split[2]
inst = id_split[-1]
# =========================================================
# Required fields
# =========================================================
phase_time_str = row['phase_time'] + 'Z'
phase_type = row['phase_type']
pick = {
"Type": "Pick",
"Site": {
"Station": sta,
"Channel": inst,
"Network": net,
"Location": loc
},
"Time": phase_time_str,
"Source": {
"AgencyID": net,
"Author": "hypoPN"
},
"Phase": phase_type,
}
# =========================================================
# Optional fields
# =========================================================
# --- Polarity ---
polarity = row.get('phase_polarity', None)
if pd.notna(polarity):
if polarity == 'U':
polarity_val = "up"
elif polarity == 'D':
polarity_val = "down"
else:
polarity_val = "no-result"
if polarity_val is not None:
pick["Polarity"] = polarity_val
# --- Picker ---
picker = row.get('picker', "deep-learning")
if pd.notna(picker) and str(picker).strip():
pick["Picker"] = str(picker)
# --- Filter ---
if highpass_filt is not None and pd.notna(highpass_filt):
pick["Filter"] = [{
"Type": "HighPass",
"HighPass": highpass_filt,
}]
# --- Amplitude ---
amplitude_dict = {}
amp = row.get('phase_amp', None)
snr = row.get('phase_snr_db', None)
if pd.notna(amp):
amplitude_dict["Amplitude"] = float(amp)
if pd.notna(snr):
amplitude_dict["SNR"] = float(snr)
if amplitude_dict:
pick["Amplitude"] = amplitude_dict
# --- Quality ---
prob = row.get('phase_score', None)
if pd.notna(prob):
pick["Quality"] = [{
"Standard": "PhaseNet",
"Value": float(prob)
}]
pick_list.append(pick)
except Exception as e:
errors.append({
"row_index": idx,
"row_data": row.to_dict(),
"message": str(e),
"traceback": traceback.format_exc()
})
return {
"status": "success" if not errors else "partial_success",
"picks": pick_list,
"errors": errors
}
[docs]
def capitalize_keys(obj):
"""Recursively capitalize the first letter of every key in a nested structure."""
if isinstance(obj, dict):
return {k[0].upper() + k[1:] if k else k: capitalize_keys(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [capitalize_keys(i) for i in obj]
else:
return obj
[docs]
def soa_to_soa_pick_format_using_anss_libraries(pick_file_used):
"""
Normalize/enhance a list of SOA picks using anss-formats==0.1.3 classes (ChannelProperties,
Source, Filter, Amplitude) for field validation, then capitalize the resulting keys back to
SOA's convention.
Note: anss-formats==0.1.3 removed anssformats.quality.Quality (folded into
Pick.analyticsInfo elsewhere in this module), so Quality here is passed through as a plain
dict -- the SOA Quality shape itself (list of {Standard, Value}) hasn't changed.
"""
TYPE = "Pick"
pick_list = []
errors = []
# --- Load file safely ---
try:
with open(pick_file_used, 'r') as f:
pick_list_modded = json.load(f)
except Exception as e:
return {
"success": False,
"stage": "load_pick_file",
"error": str(e),
"traceback": traceback.format_exc()
}
# --- Process picks ---
for idx, pick_used in enumerate(pick_list_modded):
try:
CHANNEL_PROPERTIES = ChannelProperties(
station=pick_used["Site"]["Station"],
channel=pick_used["Site"]["Channel"],
network=pick_used["Site"]["Network"],
location=pick_used["Site"]["Location"]
)
SOURCE = Source(agencyID="net", author="hypoPN")
# --- Parse time safely ---
TIME = _parse_soa_time(pick_used["Time"])
PHASE = pick_used["Phase"]
# --- Base required pick fields ---
pick = {
"Type": TYPE,
"Site": CHANNEL_PROPERTIES.model_dump(),
"Time": TIME,
"Source": SOURCE.model_dump(),
"Phase": PHASE,
}
# =========================================================
# Optional fields
# =========================================================
# --- Polarity --- (0.1.3 allows up / down / no-result)
if "Polarity" in pick_used:
polarity = pick_used.get("Polarity")
if polarity is not None:
polarity = polarity.lower()
if polarity in ["up", "down", "no-result"]:
pick["Polarity"] = polarity
# --- Onset --- (new, no 0.1.0 equivalent)
if "Onset" in pick_used:
onset = pick_used.get("Onset")
if onset in ["impulsive", "emergent", "questionable"]:
pick["Onset"] = onset
# --- Picker ---
if "Picker" in pick_used:
picker = pick_used.get("Picker")
if picker:
if picker in ["manual", "raypicker", "filterpicker", "sta-lta", "deep-learning", "machine-learning", "other"]:
pick["Picker"] = picker
# --- Filter ---
if (
"Filter" in pick_used
and isinstance(pick_used["Filter"], list)
and len(pick_used["Filter"]) > 0
):
filter_obj = pick_used["Filter"][0]
FILTER = [
Filter(
type=filter_obj["Type"],
highPass=filter_obj.get("HighPass"),
lowPass=filter_obj.get("LowPass"),
)
]
pick["Filter"] = [f.model_dump() for f in FILTER]
# --- Amplitude ---
if "Amplitude" in pick_used and isinstance(pick_used["Amplitude"], dict):
amp_val = pick_used["Amplitude"].get("Amplitude")
snr_val = pick_used["Amplitude"].get("SNR")
amplitude_kwargs = {}
if amp_val is not None:
amplitude_kwargs["amplitude"] = amp_val
if snr_val is not None:
amplitude_kwargs["snr"] = max(0.0, snr_val)
if amplitude_kwargs:
AMPLITUDE = Amplitude(**amplitude_kwargs)
pick["Amplitude"] = AMPLITUDE.model_dump()
# --- Quality --- (anssformats.quality.Quality no longer exists in 0.1.3; SOA's
# Quality shape hasn't changed, so pass it through as-is)
if (
"Quality" in pick_used
and isinstance(pick_used["Quality"], list)
and len(pick_used["Quality"]) > 0
):
pick["Quality"] = [
{"Standard": q["Standard"], "Value": q["Value"]}
for q in pick_used["Quality"]
]
pick_list.append(pick)
except Exception as e:
errors.append({
"index": idx,
"pick": pick_used,
"error": str(e),
"traceback": traceback.format_exc()
})
# --- Final formatting step ---
try:
modified_anss_picks = capitalize_keys(pick_list)
except Exception as e:
return {
"success": False,
"stage": "capitalize_keys",
"error": str(e),
"traceback": traceback.format_exc(),
"partial_picks": pick_list,
"errors": errors
}
# --- Final return ---
return {
"success": True,
"num_picks": len(modified_anss_picks),
"num_errors": len(errors),
"picks": modified_anss_picks,
"errors": errors
}