import json
from anssformats.geojson import PointGeometry
from anssformats.detection import Detection
from anssformats.source import Source
from anssformats.magnitude import Magnitude
from anssformats.pick import Pick
from anssformats.associationpickinfo import AssociationPickInfo
from anssformats.locationpickinfo import LocationPickInfo
from anssformats.channel import Channel, ChannelProperties
from anssformats.hypocenter import Hypocenter, HypocenterProperties
"""
Library for building ANSS detection JSON.
Two data sources are supported:
* AQMS db rows, passed as ``(event, origin, netmag, arrivals)`` dicts.
Every reference of Event, Origin, Netmag, Arrival, etc here is to the
equivalent AQMS db row.
* a pandas DataFrame of already-processed picks plus an ``evt`` dict, as
produced by pick-filter-spec's ``pick_processing.py``.
Both sources are forward-mapped to a common intermediate representation
(``_assemble_json``) so the anssformats object assembly lives in one place.
Ported from postprocessing-library commit 8cdb808 ("writer for anss detection
json initial", Aparna Bhaskaran) -- this is a separate, distinct entry point
from the SOA-JSON-envelope converters in ``postprocessing_seismo_lib.conversions``:
those consume the SOA JSON produced by the Phasenet lambda / pick-filter-spec's
PickProcess/v2 endpoint, while this module consumes AQMS database rows or an
already-processed pandas DataFrame.
"""
# --------------------------------------------------------------------------- #
# Public API
# --------------------------------------------------------------------------- #
[docs]
def dumps_detection_anss(*data, agencyID=None, author=None):
"""Build ANSS detection JSON and return it as a dict.
Accepts either data source:
* ``dumps_detection_anss(event, origin, netmag, arrivals)``
-- the AQMS dicts.
* ``dumps_detection_anss(picks_df, evt, agencyID=..., author=...)``
-- a pandas DataFrame of picks and an ``evt`` dict. ``agencyID`` and
``author`` set the detection-level Source (the DataFrame carries no
event agency/author).
"""
det_meta, hypo, mag, picks = _normalize(data, agencyID, author)
return _assemble_json(det_meta, hypo, mag, picks)
[docs]
def dump_detection_anss(*data, fp, agencyID=None, author=None, **json_kwargs):
"""Build ANSS detection JSON and write it to ``fp``.
Mirrors ``json.dump``: ``fp`` is a ``.write()``-supporting file-like object
and ``json_kwargs`` (e.g. ``indent``) pass through to ``json.dump``. Accepts
the same data sources as :func:`dumps_detection_anss`; ``fp`` is
keyword-only.
"""
d = dumps_detection_anss(*data, agencyID=agencyID, author=author)
json.dump(d, fp, **json_kwargs)
return d
# --------------------------------------------------------------------------- #
# Input dispatch
# --------------------------------------------------------------------------- #
def _is_dataframe(obj):
"""True if ``obj`` is a pandas DataFrame, without importing pandas eagerly."""
try:
import pandas as pd
except ImportError:
return False
return isinstance(obj, pd.DataFrame)
def _normalize(data, agencyID, author):
"""Route the positional data to the matching front-end mapper."""
if data and _is_dataframe(data[0]):
if len(data) != 2:
raise TypeError(
"DataFrame input requires (picks_df, evt); "
f"got {len(data)} positional argument(s)"
)
return _from_dataframe(data[0], data[1], agencyID, author)
if len(data) == 4:
return _from_aqms(*data)
raise TypeError(
"pass either (event, origin, netmag, arrivals) or (picks_df, evt)"
)
# --------------------------------------------------------------------------- #
# Front-end mappers: source-specific -> common intermediate
#
# Intermediate shapes:
# det_meta = {id, agencyID, author}
# hypo = {lon, lat, depth, time}
# mag = {value, type, agencyID, author}
# picks = list of dicts with keys: id, phase, time, station, channel,
# network, location, source_agencyID, source_author, method,
# onset, polarity, distance, residual, weight, azimuth, quality
# (missing optional values are None and get dropped on dump)
# quality is a list of {Standard, Value} dicts (or None)
# --------------------------------------------------------------------------- #
[docs]
def map_arrival_quality_to_score(quality, channel):
"""Map a pick's ML quality/probability to an integer hypoinverse score (0-4).
Copied verbatim from db-detection's ``db2detectionjson.py`` so the SCSN
quality thresholds live alongside the builder; behavior is intentionally
identical.
"""
thresholds = [0.97, 0.93, 0.6, 0] # quality 0,1,2,3 cutoffs
thresholds_DAS = [1, 0.97, 0.9, 0] # DAS-specific 0,1,2,3 cutoffs
maxqualB = 2 # max quality B?? channels can get
score = 4 # default value
score = 0 if quality >= thresholds[0] else score
score = 1 if quality < thresholds[0] and quality >= thresholds[1] else score
score = 2 if quality < thresholds[1] and quality >= thresholds[2] else score
score = 3 if quality < thresholds[2] and quality >= thresholds[3] else score
# now do DAS thresholds
if thresholds_DAS:
if channel[1] == 'S':
score = 0 if quality >= thresholds_DAS[0] else score
score = 1 if quality < thresholds_DAS[0] and quality >= thresholds_DAS[1] else score
score = 2 if quality < thresholds_DAS[1] and quality >= thresholds_DAS[2] else score
score = 3 if quality < thresholds_DAS[2] and quality >= thresholds_DAS[3] else score
if maxqualB > 0:
score = maxqualB if channel.startswith('B') and quality < maxqualB else score
return score
def _from_aqms(event, origin, netmag, arrivals):
det_meta = {
'id': str(event['evid']),
'agencyID': event['auth'],
'author': 'hypoPN',
}
hypo = {
'lon': origin['lon'],
'lat': origin['lat'],
'depth': origin['depth'],
'time': origin['time'],
}
mag = {
'value': netmag['magnitude'],
'type': netmag['magtype'],
'agencyID': netmag['auth'],
'author': netmag['subsource'] if netmag['subsource'] else 'NA',
}
picks = []
for arr in arrivals:
if arr['subsource']:
source_author = arr['subsource']
match arr['subsource'].lower():
case 'gamma' | 'hypopn':
method = 'deep-learning'
case 'jiggle':
method = 'manual'
case _:
method = 'sta-lta'
else:
source_author = 'NA'
method = 'sta-lta'
if arr['qual']:
match arr['qual']:
case 'e':
onset = 'emergent'
case 'i':
onset = 'impulsive'
case _:
onset = 'questionable'
else:
onset = None
match arr['fm']:
case 'c.':
polarity = 'up'
case 'd.':
polarity = 'down'
case '..':
polarity = 'no-result'
case _:
polarity = 'no-result'
picks.append({
'id': arr['id'],
'phase': arr['phase'],
'time': arr['time'],
'station': arr['site']['station'],
'channel': arr['site']['channel'],
'network': arr['site']['network'],
'location': arr['site']['location'],
'source_agencyID': arr['auth'],
'source_author': source_author,
'method': method,
'onset': onset,
'polarity': polarity,
'distance': arr['distance'],
'residual': arr['timeres'],
'weight': arr['wgt'],
'azimuth': arr['azimuth'],
'quality': (
[{'Standard': 'hypoinverse',
'Value': map_arrival_quality_to_score(arr['quality'], arr['site']['channel'])}]
if arr.get('quality') is not None else None
),
})
return det_meta, hypo, mag, picks
def _from_dataframe(picks_df, evt, agencyID, author):
"""Map pick-filter-spec's ``(picks_df, evt)`` to the common intermediate.
``picks_df`` rows are already-processed ANSS picks (capitalized keys, values
already mapped to their final strings). ``evt`` is
``{'id', 'loc'(.latitude/.longitude/.z), 'time', 'mag'({Type, Value})}``.
"""
import pandas as pd
def clean(v):
# dict-valued cells (Site/Source/AssociationInfo) are never NaN-scalars
if isinstance(v, dict):
return v
return None if (v is None or (pd.api.types.is_scalar(v) and pd.isna(v))) else v
loc = evt['loc']
det_meta = {
'id': str(evt['id']),
'agencyID': agencyID,
'author': author,
}
hypo = {
'lon': loc.longitude,
'lat': loc.latitude,
'depth': loc.z,
'time': evt['time'],
}
mag = evt['mag']
mag = {
'value': mag['Value'],
'type': mag['Type'],
'agencyID': agencyID,
'author': 'NA',
}
has_id = 'id' in picks_df.columns or 'ID' in picks_df.columns
id_col = 'id' if 'id' in picks_df.columns else 'ID'
picks = []
for idx, row in picks_df.iterrows():
site = clean(row.get('Site')) or {}
src = clean(row.get('Source')) or {}
assoc = clean(row.get('AssociationInfo')) or {}
quality = clean(row.get('Quality')) # already [{Standard, Value}, ...]
# Pick.id is required by the schema; fall back to the row index.
pick_id = clean(row.get(id_col)) if has_id else None
if pick_id is None:
pick_id = str(idx)
picks.append({
'id': str(pick_id),
'phase': clean(row.get('Phase')),
'time': clean(row.get('Time')),
'station': site.get('Station'),
'channel': site.get('Channel'),
'network': site.get('Network'),
'location': site.get('Location'),
'source_agencyID': src.get('AgencyID'),
'source_author': src.get('Author'),
'method': clean(row.get('Picker')),
'onset': clean(row.get('Onset')),
'polarity': clean(row.get('Polarity')),
'distance': assoc.get('Distance'),
'residual': assoc.get('Residual'),
'quality': quality if quality is not None else None,
# not present in the pick DataFrame -> omitted from output
'weight': None,
'azimuth': None,
})
return det_meta, hypo, mag, picks
# --------------------------------------------------------------------------- #
# Common core: intermediate -> anssformats objects -> JSON
# --------------------------------------------------------------------------- #
def _assemble_json(det_meta, hypo, mag, pick_dicts):
source = Source(
agencyID=det_meta['agencyID'],
author=det_meta['author'],
)
point_geometry = PointGeometry(coordinates=[hypo['lon'], hypo['lat'], hypo['depth']])
hypocenter = Hypocenter(
geometry=point_geometry,
properties=HypocenterProperties(originTime=hypo['time']),
)
magnitude = Magnitude(
value=mag['value'],
type=mag['type'],
source=Source(agencyID=mag['agencyID'], author=mag['author']),
)
picks = []
for p in pick_dicts:
channel_geometry = PointGeometry(coordinates=[hypo['lon'], hypo['lat'], hypo['depth']])
channel_properties = ChannelProperties(
station=p['station'],
channel=p['channel'],
network=p['network'],
location=p['location'],
)
pick_source = Source(
agencyID=p['source_agencyID'],
author=p['source_author'] if p['source_author'] else 'NA',
)
association_info = AssociationPickInfo(
phase=p['phase'],
source=pick_source,
)
# weight/azimuth (and distance) left as None are dropped by exclude_none.
location_pick_info = LocationPickInfo(
phase=p['phase'],
distance=p['distance'],
weight=p['weight'],
residual=p['residual'],
azimuth=p['azimuth'],
source=pick_source,
)
picks.append(Pick(
type="Pick",
id=p['id'],
phase=p['phase'],
source=pick_source,
channel=Channel(geometry=channel_geometry, properties=channel_properties),
time=p['time'],
method=p['method'],
associationInfo=association_info,
locationInfo=location_pick_info,
onset=p['onset'],
polarity=p['polarity'],
))
detection = Detection(
type="Detection",
id=det_meta['id'],
source=source,
hypocenter=hypocenter,
magnitudeData=[magnitude],
pickData=picks,
)
# model_dump() leaves datetimes as objects; only the JSON serializer renders
# them as the ISO8601 Z-suffixed strings the Detection spec requires.
d = json.loads(detection.model_dump_json())
# anssformats' Pick schema has no quality field and silently drops unknown
# keys, so inject Quality into the serialized picks. pickData preserves the
# order of pick_dicts.
for pick_json, p in zip(d.get('pickData', []), pick_dicts):
if p.get('quality'):
pick_json['quality'] = p['quality']
return d