postprocessing_seismo_lib

postprocessing_seismo_lib is a lightweight Python library for building and parsing structured API messages, especially for use with nested JSON structures used in event-based data systems. It also converts seismic pick and detection data between the SOA/SCSN JSON format and the ANSS format used by anss-formats.

This library is vetted and works against python 3.10.5. This library does not work with any python libraries below 3.10 (it has been specifically vetted against python 3.6.5 and python 3.8.10 and found not to work).

Table of Contents

• Features
• Use cases of this library
• Installation
• Core Utilities
  • Extraction of body
  • Creation of the request for a body object
  • Creation of the request for a body object using ANSS-format picks (wrap_data_anss, new)
  • Creation of full response format
  • Updating a DataRetrieval input file (new)
• Pick / Detection Format Conversions
  • SOA vs ANSS field mapping (anss-formats==0.1.3)
  • Core Pick conversion functions
  • How to Run the Conversion Examples
  • SOA-envelope → ANSS Pick / Detection conversions (new)
  • AQMS / pick-filter-spec Detection JSON writer (new)
  • Hypocenter JSON writer (new)
• Notes

Features

• Extract the body section from a structured JSON file using extract_body_from_file
• Create request for a body object using wrap_data, with provided associator or pickfilter files.
• Validates if input and output formats are to specification using wrap_data
• Create request for a body object from ANSS-format picks (e.g. Phasenet/NEIC output) using wrap_data_anss, the ANSS-format counterpart to wrap_data, for provided associator or pickfilter files.
• Builds a full message with status, headers and body using convert_file_to_json, with provided csv, arcout or quakeml files
• Converts seismic Pick and Detection data between the SOA/SCSN JSON format and the ANSS format (anss-formats==0.1.3), from several different upstream sources (a raw SOA pick list, the Phasenet lambda's response, pick-filter-spec's PickProcess/v2 response, or an AQMS-db / pandas-DataFrame pipeline)
• Round-trips a full SOA detection (pick-filter-style or association-style picks, enveloped or bare, with or without Magnitude) through anss-formats==0.1.3's Detection/Pick classes and back using soa_to_soa_detection_format_using_anss_libraries
• Updates a DataRetrieval-style JSON file's RetrieveParameters (Host/Method/Port/Window/SavePath/Url/directory, and optionally an event ID under inEvid) using update_dataretrieval_input

This library also provides utilities for converting between different seismic pick and detection data formats, including:

• SOA → SOA (ANSS-informed)
• SOA → ANSS (Pick)
• ANSS → SOA (Pick)
• ANSS → SCSN (Detection)
• PhaseNet CSV → SOA
• Phasenet-lambda SOA → ANSS (Pick)
• pick-filter-spec SOA → ANSS (Pick)
• pick-filter-spec SOA → ANSS (Detection)
• AQMS db rows / pandas DataFrame → ANSS (Detection)

These functions are useful for normalizing pick and detection data across different pipelines and tools. See Pick / Detection Format Conversions for details.

Use cases of this library

1. Individual users
2. Pipeline scripts

Installation

pip install postprocessing-seismo-lib
OR
pip install --upgrade postprocessing-seismo-lib

This installs anss-formats==0.1.3 as a dependency (declared in setup.py). If you have code written against the older anss-formats==0.1.0-shaped Pick objects (pickerType, qualityInfo, machineLearningInfo, anssformats.association.Association), see SOA vs ANSS field mapping below -- those classes/fields were removed in 0.1.3.

After installation, we have provided sample files that can be vetted against the library's core functions. Run the below script to analyze the contents of each file, and see if the outputs are generated locally:

import json, importlib.resources
from postprocessing_seismo_lib import wrap_data, extract_body_from_file, convert_file_to_json

pick_file = json.load(importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('xxxx_file_containing_picks.json').open('r'))
print("Picks")
print(pick_file)

filtered_pick_file = json.load(importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('xxxx_file_containing_filtered_picks.json').open('r'))
print("Filtered picks")
print(pick_file)


json_path = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('40584759_csv.json')
print("JSON file with body")
print(json_path)


#FIRST USE CASE: Extract body from a JSON file
body_data = extract_body_from_file(str(json_path))
print("Body extracted:")
print(body_data)


#SECOND USE CASE: Create RetrieveParameter wrapping around input data for various modules

## FOR THE ASSOCIATOR MODULE:

input_path = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('xxxx_file_containing_filtered_picks.json')

wrap_data(
    input_file_path=str(input_path),
    output_file_path='output_associator.json',
    evid='evid_filtered_picks',
    module='associator'
)

## FOR THE PICK FILTER MODULE:

### THE BELOW SCENARIO filters picks under the default conditions:
[1] mode='hypoPN'
[2] testType='local'
[3] logging='False'

input_path = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('xxxx_file_containing_picks.json')

wrap_data(
    input_file_path=str(input_path),
    output_file_path='output_pickfilter.json',
    evid='evid_picks',
    module='pickfilter'
)

### THE BELOW SCENARIO shows that we can adjust those conditions within the pickfilter:
[1] mode='st-proc'
[2] testType='local'
[3] logging='True'

wrap_data(
    input_file_path = [YOUR INPUT FILE PATH],
    output_file_path = [YOUR OUTPUT FILE PATH],
    module = 'pickfilter',
    evid = '[NAME OF EVID USED]',
    mode = 'st-proc',
    testType = 'local',
    logging = 'True'
)

#THIRD USE CASE: Create Response wrapping around known data
gamma_events = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('40584759_gamma_events.csv')
gamma_picks = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('40584759_gamma_picks.csv')

xml_file_nosignifier = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('40584759_events_testGOUA')
xml_file_signifier = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('40584759_events_test.xml')
arcout_file = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('z78966423_api_stproc_9999.arcout')

print("GaMMa events")
print(gamma_events)
print(type(gamma_events))

print("GaMMa picks")
print(gamma_picks)
print(type(gamma_picks))

# For CSV
convert_file_to_json(
    input_file="",  # not used for CSV
    output_file="response_csv.json",
    id="id_testing",
    event_file=str(gamma_events),
    pick_file=str(gamma_picks),
    error_log_file="csv_error_log.txt"
)

# For QuakeML XML (this input file has no XML signifiers but was parsed successfully as XML here)
convert_file_to_json(
    input_file=str(xml_file_nosignifier),
    output_file="response_quakeml_nosignifiers.json",
    id="id testing",
    error_log_file="quakeml_error_log_one.txt"
)

#Conventional QuakeML XML here
convert_file_to_json(
    input_file=str(xml_file_signifier),
    output_file="response_quakeml_signifiers.json",
    id="id testing",
    error_log_file="quakeml_error_log_two.txt"
)


# For ArcOut
convert_file_to_json(
    input_file=str(arcout_file),
    output_file="response_arcout.json",
    id="id testing",
    error_log_file="arcout_error_log.txt"
)

Core Utilities

Extraction of body

The below function allows for extracting out the body from an output response file:

from postprocessing_seismo_lib import extract_body_from_file

body_data = extract_body_from_file("output_response_association.json")
body_data = extract_body_from_file("output_response_pickfilter.json")

where as an example, output_response_association.json is:

{
  "status": 404,
  "headers": {
    "Content-Type": "application/json"
  },
  "body": {
    "id": "78604159",
    "format": "none.noeventsfound",
    "data": []
  }
}

Creation of the request for a body object

The below function creates the request from the body object, which can be extracted from the above function. All four variables listed below need to be specified:

from postprocessing_seismo_lib import wrap_data

#creating the request for the associator input
wrap_data(
    input_file_path='[xxxx_file_containing_filtered_picks].json',
    output_file_path='output_associator.json',
    evid='[Name of choice]',
    module='associator'
)


#creating the request for the pickfilter input

## Pickfilter default settings:
[1] mode='hypoPN'
[2] testType='local'
[3] logging='False'

wrap_data(
    input_file_path='[xxxx_file_containing_picks].json',
    output_file_path='output_pickfilter.json',
    evid='[Name of choice]',
    module='pickfilter'
)

### Pickfilter, adjusting various settings:
[1] mode='st-proc'
[2] testType='local'
[3] logging='True'

wrap_data(
    input_file_path = [YOUR INPUT FILE PATH],
    output_file_path = [YOUR OUTPUT FILE PATH],
    module = 'pickfilter',
    evid = '[NAME OF EVID USED]',
    mode = 'st-proc',
    testType = 'local',
    logging = 'True'
)

The request format will be different across each module. Currently, the module takes in 'associator' and 'pickfilter' but this will be expanded in future updates.

Specifically, this function reads a list of pick dictionaries from a JSON file specified by input_file_path, validates them against a schema, wraps the data into a module-specific JSON structure, validates the output, and writes it to a new file specified by output_file_path. Any errors are logged to a file named wrap_data_errors.log.

wrap_data's input schema does not require every SOA pick key to be present -- for example, Onset (present on pick-filter output, generally absent on raw Phasenet output) is accepted either way, and undeclared extra keys are never rejected. See How to Run the Conversion Examples below for a worked example using a real pick-filter response (which does carry Onset).

As an example, our input_file_path='[xxxx_file_containing_picks].json' might look like this (as a list of dictionaries):

[
    {
        "Amplitude": {
            "Amplitude": 1039.6302490234,
            "SNR": 11.074
        },
        "Filter": [
            {
                "HighPass": 1.0,
                "Type": "HighPass"
            }
        ],
        "Onset": "emergent",
        "Phase": "S",
        "Picker": "deep-learning",
        "Polarity": "no-result",
        "Quality": [
            {
                "Standard": "PhaseNet",
                "Value": 0.851
            },
            {
                "Standard": "hypoinverse",
                "Value": 2
            }
        ],
        "Site": {
            "Channel": "HHE",
            "Location": "",
            "Network": "CI",
            "Station": "WOR"
        },
        "Source": {
            "AgencyID": "CI",
            "Author": "hypoPN"
        },
        "Time": "2025-04-22T21:51:15.148Z",
        "Type": "Pick"
    },
    {
        ...
    }
]

and its output would be the necessary format to POST into the associator API endpoint:

{
  "RetrieveParameters": {
    "pickFile": "Ryan_testingAgainPicks_picks.json",
    "pickDataStr": [
      {
        "Amplitude": {
          "Amplitude": 1039.6302490234,
          "SNR": 11.074
        },
        "Filter": [
          {
            "HighPass": 1.0,
            "Type": "HighPass"
          }
        ],
        "Onset": "emergent",
        "Phase": "S",
        "Picker": "deep-learning",
        "Polarity": "no-result",
        "Quality": [
          {
            "Standard": "PhaseNet",
            "Value": 0.851
          },
          {
            "Standard": "hypoinverse",
            "Value": 2
          }
        ],
        "Site": {
          "Channel": "HHE",
          "Location": "",
          "Network": "CI",
          "Station": "WOR"
        },
        "Source": {
          "AgencyID": "CI",
          "Author": "hypoPN"
        },
        "Time": "2025-04-22T21:51:15.148Z",
        "Type": "Pick"
      },
      ...
    ]
  }
}

Creation of the request for a body object using ANSS-format picks (wrap_data_anss, new)

wrap_data validates picks against the SCSN/SOA pick schema (capitalized keys, e.g. Amplitude/Site/Source). If your picks are instead in ANSS format (lowerCamelCase keys, e.g. amplitudeInfo/channel/source) -- the format produced upstream by Phasenet/NEIC -- wrap_data will reject them:

wrap_data(
    input_file_path="picks_neic_body.json",
    output_file_path="picks_neic_body_pickfilter.json",
    module='pickfilter',
    evid='evid_picks',
)
# Error in wrap_data (module: pickfilter): Pick at index 0 is invalid: 'Amplitude' is a required property

wrap_data_anss(input_file_path, output_file_path, evid='stubvalue', module='associator', mode='hypoPN', testType='local', logging='False') is the ANSS-format counterpart: same signature, same output wrapping structure (RetrieveParameters with evid/pickFile/mode/testType/logging/pickDataStr depending on module), but it validates each pick against the ANSS pick schema instead. As with wrap_data, input_file_path must be a bare JSON list of picks -- use extract_body_from_file first if starting from an envelope-wrapped response file.

Pickfilter example, using example_data/picks_neic.json (raw Phasenet/NEIC picks, envelope-wrapped) and the body extracted from it, example_data/picks_neic_body.json:

import json
from postprocessing_seismo_lib import wrap_data_anss, extract_body_from_file

file_used = 'picks_neic.json'
body_data = extract_body_from_file(file_used)

# Save the list to a file
with open("picks_neic_body.json", "w", encoding="utf-8") as file:
    json.dump(body_data, file, indent=4)

wrap_data_anss(
    input_file_path="picks_neic_body.json",
    output_file_path="picks_neic_body_pickfilter.json",
    module='pickfilter',
    evid='evid_picks',
    mode='hypoPN',
    testType='local_input_anss_pick_output_anss_detection',
    logging='False'
)
# Data validated and written to 'picks_neic_body_pickfilter.json' for module 'pickfilter'.

Associator example, using example_data/61146748_filteredpicks_picks_NEIC_PICK.json (envelope-wrapped filtered NEIC-format picks -- the output of an upstream pick-filter step):

from postprocessing_seismo_lib import wrap_data_anss, extract_body_from_file
import json

body_data = extract_body_from_file('61146748_filteredpicks_picks_NEIC_PICK.json')
with open("61146748_filteredpicks_picks_NEIC_PICK_body.json", "w", encoding="utf-8") as file:
    json.dump(body_data, file, indent=4)

wrap_data_anss(
    input_file_path="61146748_filteredpicks_picks_NEIC_PICK_body.json",
    output_file_path="output_associator_anss.json",
    module='associator',
    evid='61146748'
)
# Data validated and written to 'output_associator_anss.json' for module 'associator'.

Like wrap_data's SCSN schema, the ANSS pick schema requires only the structurally load-bearing fields (type, id, channel, source, time, phase); the remaining fields observed on ANSS picks (polarity, onset, method, filterInfo, amplitudeInfo, associationInfo, analyticsInfo) are validated for shape when present but not required -- the two example files above disagree on which of those are populated on every pick (e.g. the associator-style file always carries polarity but only some picks carry onset/method/filterInfo/amplitudeInfo, while the pickfilter-style file is the reverse), so requiring all of them would make one of the two example files fail validation. Errors are logged to the same wrap_data_errors.log file used by wrap_data.

Creation of full response format

Below shows how to build out the Response format for provided files. In all cases below, you provide an ID and an output file name (of type json). Also, provide the error log file, in case any errors occur. If any errors exist, a file of the name you specified will be generated. If no errors exist, the output JSON file will be generated at the path where you run the python script.

If you are converting from csv to json, you provide the _events.csv and _picks.csv that are generated from pinging the associator API, and set them to event_file and pick_file. Leave the input_file blank. For quakeML or arcout conversion to json, specify the input_file.

from postprocessing_seismo_lib import convert_file_to_json

# For CSV
convert_file_to_json(
    input_file="",  # not used for CSV
    output_file="[Output file name].json",
    id="[Name of choice]",
    event_file="[xxxx]_gamma_events.csv",
    pick_file="[xxxx]_gamma_picks.csv",
    error_log_file="csv_error_log.txt"
)

# For QuakeML XML (this input file has no XML signifiers but was parsed successfully as XML here)
convert_file_to_json(
    input_file="[xxxx]_events_test",
    output_file="[xxxx]_quakeml.json",
    id="[Name of choice]",
    error_log_file="quakeml_error_log.txt"
)

#Conventional QuakeML XML here
convert_file_to_json(
    input_file="[xxxx]_events_test.xml",
    output_file="[xxxx]_quakeml.json",
    id="[Name of choice]",
    error_log_file="quakeml_error_log.txt"
)


# For ArcOut
convert_file_to_json(
    input_file="[xxxx]_api_stproc_9999.arcout",
    output_file="[Output file name].json",
    id="[Name of choice]",
    error_log_file="arcout_error_log.txt"
)

Updating a DataRetrieval input file (new)

update_dataretrieval_input(file_path, output_file_path, Host, Method, Port, Window, SavePath=None, Url=None, directory=None, evid=None) loads an existing DataRetrieval-style JSON file, overwrites/creates the mandatory RetrieveParameters keys (Host/Method/Port/Window), always sets SavePath (empty string if not given), and writes the result to output_file_path. Url and directory are added only if provided. evid works the same way: if supplied, it's added to RetrieveParameters under a new inEvid key; if omitted, no inEvid key is added at all (this matches the pre-existing behavior for Url/directory, and keeps the function backward compatible with callers that don't pass evid).

from postprocessing_seismo_lib import update_dataretrieval_input
import importlib.resources

data_choice_file = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('61077388_data_choice_updated.json')

updated = update_dataretrieval_input(
    file_path=data_choice_file,
    output_file_path="61077388_data_choice_with_evid.json",
    Host="gibbes.gps.caltech.edu",
    Method="TriNet",
    Port=9101,
    Window="event",
    evid="61077388"
)

print("RetrieveParameters after update:")
print(updated["RetrieveParameters"])
# {'Host': 'gibbes.gps.caltech.edu', 'Method': 'TriNet', 'Port': 9101, 'Window': 'event',
#  'SavePath': '', 'inEvid': '61077388'}

61077388_data_choice_updated.json already contains a Windows list (per-channel retrieval time windows) alongside its RetrieveParameters section; the Windows list passes through untouched -- only RetrieveParameters is modified. Calling update_dataretrieval_input again without evid leaves RetrieveParameters exactly as it was before this feature was added (no inEvid key present).

Pick / Detection Format Conversions

This library also allows for conversion between SOA and ANSS Pick and Detection formats, using anss-formats==0.1.3.

Note that whether you are converting to the ANSS or SOA format, the following keys: Polarity, Onset, Picker, Filter, Amplitude and Quality are all optional, with Type, Site, Source, Time and Phase mandatory.

SOA vs ANSS field mapping (anss-formats==0.1.3)

anss-formats==0.1.3 changed the Pick schema significantly from the 0.1.0 version this library previously targeted. If you have existing code or data built against 0.1.0, note the following:

Because Pick.method is now a free string, soa_to_anss_pick_format no longer needs to coerce unrecognized Picker values (like deep-learning or machine-learning) down to "other" -- they pass straight through.

The SOA Pick format itself (Site/Source/Time/Phase/Polarity/Picker/Onset/Filter/Amplitude/Quality, capitalized keys) has not changed.

Core Pick conversion functions

from postprocessing_seismo_lib import (
    anss_to_soa_pick_format,
    soa_to_anss_pick_format,
    phasenet_csv_to_soa_pick_format,
    soa_to_soa_pick_format_using_anss_libraries
)

soa_to_anss_pick_format(pick_file_path, station_csv_path=None) looks up each pick's station coordinates by (Network, Station, Channel). station_csv_path is optional: if omitted, the archive_stations_loc_dates.csv bundled with this library under postprocessing_seismo_lib/example_data/ is used as the default station list; pass your own CSV (same NET,STA,CHA,LOC,LAT,LON,ELEV,ONDATE,OFFDATE columns) to override it.

How to Run the Conversion Examples

The library includes example datasets under postprocessing_seismo_lib.example_data. You can use these to test each conversion workflow.

Start by importing dependencies:

import json
import importlib.resources
import pandas as pd

Example 1: SOA → SOA (ANSS-Informed)

Enhances SOA picks using ANSS-informed logic.

pick_file_one_contents = json.load(
    importlib.resources.files('postprocessing_seismo_lib.example_data')
    .joinpath('79765767_picks.json')
    .open('r')
)

print("Inspecting the pick contents, example one")
print(pick_file_one_contents)

pick_file_two_contents = json.load(
    importlib.resources.files('postprocessing_seismo_lib.example_data')
    .joinpath('60209491_picks.json')
    .open('r')
)

print("Inspecting the pick contents, example two")
print(pick_file_two_contents)

pick_file_one = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('79765767_picks.json')

print("Path for the pick file, example one")
print(pick_file_one)

pick_file_two = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('60209491_picks.json')

print("Path for the pick file, example one")
print(pick_file_two)

soa_picks_informed_by_anss = soa_to_soa_pick_format_using_anss_libraries(pick_file_one)

print("SOA picks (ANSS-informed):")
print(soa_picks_informed_by_anss)

Example 2: SOA → ANSS

Converts SOA-formatted picks into ANSS format using station metadata.

archive_stations = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('archive_stations_loc_dates.csv')

pick_file_two = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('60209491_picks.json')

print("Path for the archive station file")
print(archive_stations)

df = pd.read_csv(archive_stations)
print("Archive station contents:")
print(df)

formatted_anss_picks = soa_to_anss_pick_format(pick_file_two, archive_stations)

print("ANSS picks, from SOA")
print(formatted_anss_picks)

# List of picks in ANSS format
picks_anss = formatted_anss_picks["picks"]

string_dict_picks = json.dumps(picks_anss, indent=4, default=str)

# saved file
with open("picks_ANSS_from_library.json", "w") as f:
    f.write(string_dict_picks)

Example 3: PhaseNet CSV → SOA

Converts PhaseNet TensorFlow CSV output into SOA pick format.

picks_phasenet_tensorflow = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('picks_phasenet_tensorflow.csv')

print("Path for the picks_phasenet_tensorflow file")
print(picks_phasenet_tensorflow)

df = pd.read_csv(picks_phasenet_tensorflow)
print("Phasenet picks, tensorflow, contents:")
print(df)

highpass_filt = 1.0

pick_list = phasenet_csv_to_soa_pick_format(picks_phasenet_tensorflow, highpass_filt)

print("SOA picks, from Phasenet CSV")
print(pick_list)

Example 4: ANSS → SOA

Converts ANSS-formatted picks (anss-formats==0.1.3 shape) into SOA format.

anss_file_contents = json.load(importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('picks_ANSS.json').open('r'))
print("Inspecting the pick contents, ANSS format")
print(anss_file_contents)

anss_file_path = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('picks_ANSS.json')
print("Path for the pick file, ANSS format")
print(anss_file_path)

soa_picks=anss_to_soa_pick_format(anss_file_path)

# Access list of picks
print("SOA picks, from ANSS")
picks_soa=soa_picks["picks"]
print(picks_soa)

SOA-envelope → ANSS Pick / Detection conversions (new)

These four functions consume the SOA-format JSON response envelopes produced further upstream in the pipeline (the Phasenet lambda, or pick-filter-spec's PickProcess/v2 endpoint) and convert them directly into anss-formats==0.1.3 Pick/Detection objects (or, for soa_to_soa_detection_format_using_anss_libraries, back into SOA detection format after a detour through those ANSS objects). All accept either a file path to the full {"status", "headers", "body"} envelope, a path to just the bare body, or an already-loaded dict/list -- and all accept an optional station_csv_path, defaulting to the bundled archive_stations_loc_dates.csv if not given.

from postprocessing_seismo_lib import (
    phasenet_soa_to_anss_pick_format,
    pickfilter_soa_to_anss_pick_format,
    pickfilter_soa_to_anss_detection_format,
    soa_to_soa_detection_format_using_anss_libraries,
)

Phasenet lambda SOA pick list → ANSS Pick

phasenet_soa_to_anss_pick_format(soa_source, station_csv_path=None) converts the SOA picks returned by the Phasenet lambda into a list of anss-formats==0.1.3 Pick dicts.

phasenet_soa_response = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('picks_SOA_fromPhasenetLambda.json')

formatted = phasenet_soa_to_anss_pick_format(phasenet_soa_response)

print("ANSS picks, from Phasenet-lambda SOA response")
print(formatted["picks"])

Example input (picks_SOA_fromPhasenetLambda.json, one entry from the "body" list):

{
    "Type": "Pick",
    "Site": {"Station": "TKX", "Channel": "HNZ", "Network": "BC", "Location": ""},
    "Time": "2026-02-24T17:47:47.410Z",
    "Source": {"AgencyID": "BC", "Author": "hypoPN"},
    "Phase": "P",
    "Polarity": "up",
    "Picker": "deep-learning",
    "Filter": [{"Type": "HighPass", "HighPass": 1.0}],
    "Amplitude": {"Amplitude": 3.380613327026367, "SNR": 10.485},
    "Quality": [{"Standard": "PhaseNet", "Value": 0.906}]
}

pick-filter-spec SOA pick list → ANSS Pick

pickfilter_soa_to_anss_pick_format(pickfilter_source, station_csv_path=None) converts the SOA picks returned by pick-filter-spec's PickProcess/v2 endpoint (a "localpick"-style request) into a list of anss-formats==0.1.3 Pick dicts. Unlike a raw Phasenet pick, a pick-filter pick typically carries Onset and a two-entry Quality list (a PhaseNet probability and a hypoinverse pick-weight code 0-4) -- both are preserved (Onset -> Pick.onset, Quality -> Pick.analyticsInfo.predictions, one Prediction per Quality entry).

pickfilter_picks_response = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('61146748_filteredpicks_picksToCONVSPECFORMAT_PICKK.json')

formatted = pickfilter_soa_to_anss_pick_format(pickfilter_picks_response)

print("ANSS picks, from pick-filter-spec SOA response")
print(formatted["picks"])

postprocessing_seismo_lib/example_data/61146748_picksSPECFORMAT_ANSS_from_library_0.1.3.json is the expected output for this fixture (bundled alongside the input so you can diff your own run against it -- every field matches except each pick's randomly-generated id).

pick-filter-spec SOA detection → ANSS Detection

pickfilter_soa_to_anss_detection_format(detection_source, station_csv_path=None) converts the SOA detection returned by pick-filter-spec's PickProcess/v2 endpoint (a "localdetection"-style request) into a single anss-formats==0.1.3 Detection dict.

pickfilter_detection_response = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('61146748_filteredpicks_detectionSPECFORMAT_TEST_2026CHECK_DETECT.json')

formatted = pickfilter_soa_to_anss_detection_format(pickfilter_detection_response)

print("ANSS detection, from pick-filter-spec SOA response")
print(formatted["detection"])

postprocessing_seismo_lib/example_data/61146748_detectionSPECFORMAT_ANSS_from_library_0.1.3.json is the expected output for this fixture (again, matches except for pick ids).

SOA detection → ANSS Detection → SOA detection round trip

soa_to_soa_detection_format_using_anss_libraries(detection_source, station_csv_path=None) is the detection-level analogue of soa_to_soa_pick_format_using_anss_libraries: it takes an SOA-format detection (the loaded JSON contents, not just a file path), builds the anss-formats==0.1.3 Detection/Pick objects, and converts the result back into SOA detection format. Use it to validate/normalize an SOA detection through the ANSS library without needing the raw anss-formats Detection dict as your output.

It auto-detects, independently:

• Envelope vs bare body -- a {"status", "headers", "body": {...}} response envelope, or an already-unwrapped detection body dict.
• Optional Magnitude -- a Magnitude list of {Type, Value, Author}, or absent entirely.
• Per-pick shape -- pick-filter-style picks (Filter/Amplitude/Quality, no AssociationInfo) and association-style picks (AssociationInfo with Phase/Distance/Azimuth/Residual, no Filter/Amplitude) both round-trip correctly, since filterInfo/amplitudeInfo and locationInfo are independent, optional fields on the ANSS Pick object.

from postprocessing_seismo_lib import soa_to_soa_detection_format_using_anss_libraries

# 1) Envelope-wrapped, pick-filter-style picks (Filter/Amplitude/Quality)
pickfilter_detection_response = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('61146748_filteredpicks_detectionSPECFORMAT_TEST_2026CHECK_DETECT.json')
result_1 = soa_to_soa_detection_format_using_anss_libraries(pickfilter_detection_response)
print("SOA detection (ANSS-informed), envelope + pick-filter-style picks")
print(result_1["detection"])

# 2) Bare body, pick-filter-style picks (Filter/Amplitude/Quality), with Magnitude
detection_body_file = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('1204954_detection.json')
result_2 = soa_to_soa_detection_format_using_anss_libraries(detection_body_file)
print("SOA detection (ANSS-informed), bare body + pick-filter-style picks")
print(result_2["detection"])

# 3) Bare body, association-style picks (AssociationInfo w/ Distance/Azimuth/Residual)
association_detection_file = importlib.resources.files('postprocessing_seismo_lib.example_data').joinpath('association_detection_wrapped_data.json')
result_3 = soa_to_soa_detection_format_using_anss_libraries(association_detection_file)
print("SOA detection (ANSS-informed), bare body + association-style picks")
print(result_3["detection"])

All three calls return {"success", "detection", "num_picks", "num_errors", "errors"}, where "detection" is the reconstructed SOA-format detection dict (Type/ID/Source/Hypocenter/Data, plus Magnitude when present in the input).

AQMS / pick-filter-spec Detection JSON writer (new)

A separate module, postprocessing_seismo_lib.json_writers, builds anss-formats==0.1.3 Detection JSON directly from two other, distinct data sources -- not the SOA JSON envelopes used by the functions above:

• AQMS database rows, passed as (event, origin, netmag, arrivals) dicts.
• a pandas DataFrame of already-processed picks plus an evt dict, as produced by pick-filter-spec's pick_processing.py.

from postprocessing_seismo_lib import dumps_detection_anss, dump_detection_anss, map_arrival_quality_to_score, detection_anss_to_scsn

dumps_detection_anss(*data, agencyID=None, author=None) -> dict builds and returns the ANSS detection JSON. dump_detection_anss(*data, fp, agencyID=None, author=None, **json_kwargs) does the same and writes it to a file-like object fp (mirroring json.dump). map_arrival_quality_to_score(quality, channel) maps an ML confidence score to a 0-4 Hypoinverse pick-weight code, for callers building AQMS-style arrivals.

Each pick's channel.geometry in the output is the station's lon/lat/elev, not the detection's hypocenter -- for AQMS input this comes from arrivals[i]['site'], which always carries real coordinates. For the (picks_df, evt) input, no station-CSV lookup is performed: if a row's Site dict carries Lon/Lat/Elev, they're passed straight through to the pick's Channel geometry; if it doesn't (the common case, since pick-filter-spec's Site normally has no coordinates), that pick's channel.geometry is simply omitted from the output rather than filled with a fabricated or looked-up location. A partial set (e.g. Lon/Lat with no Elev) is also omitted -- all three or none.

Without station coordinates (the common case for pick-filter-spec's DataFrame output):

import pandas as pd
from datetime import datetime
from types import SimpleNamespace

picks_df = pd.DataFrame([
    {
        "id": "p1",
        "Phase": "P",
        "Time": datetime(2026, 7, 19, 12, 0, 0),
        "Site": {"Station": "ABL", "Channel": "HHZ", "Network": "CI", "Location": ""},
        "Source": {"AgencyID": "CI", "Author": "hypoPN"},
        "Picker": "deep-learning",
        "Onset": "impulsive",
        "Polarity": "up",
        "AssociationInfo": {"Distance": 0.5, "Residual": -0.02},
        "Quality": [{"Standard": "hypoinverse", "Value": 1}],
    },
])

evt = {
    "id": 99999,
    "loc": SimpleNamespace(longitude=-118.0, latitude=34.0, z=8.0),
    "time": datetime(2026, 7, 19, 11, 59, 55),
    "mag": {"Type": "l", "Value": 2.5},
}

anss_detection = dumps_detection_anss(picks_df, evt, agencyID="CI", author="hypoPN")
print(anss_detection["hypocenter"]["geometry"]["coordinates"])   # event location
print("geometry" in anss_detection["pickData"][0]["channel"])    # False -- no station coords for DataFrame input
'''
[-118.0, 34.0, 8.0]
False
'''

With station coordinates -- add Lon/Lat/Elev to the row's Site dict and they pass straight through to channel.geometry (everything else identical to the example above):

picks_df = pd.DataFrame([
    {
        "id": "p1",
        "Phase": "P",
        "Time": datetime(2026, 7, 19, 12, 0, 0),
        "Site": {
            "Station": "ABL", "Channel": "HHZ", "Network": "CI", "Location": "",
            "Lon": -119.22496, "Lat": 34.84843, "Elev": 1974.0,
        },
        "Source": {"AgencyID": "CI", "Author": "hypoPN"},
        "Picker": "deep-learning",
        "Onset": "impulsive",
        "Polarity": "up",
        "AssociationInfo": {"Distance": 0.5, "Residual": -0.02},
        "Quality": [{"Standard": "hypoinverse", "Value": 1}],
    },
])

anss_detection = dumps_detection_anss(picks_df, evt, agencyID="CI", author="hypoPN")
print(anss_detection["pickData"][0]["channel"])
'''
{'type': 'Feature',
 'geometry': {'type': 'Point', 'coordinates': [-119.22496, 34.84843, 1974.0]},
 'properties': {'station': 'ABL', 'channel': 'HHZ', 'network': 'CI', 'location': ''}}
'''

The reverse direction, detection_anss_to_scsn(detection) -> dict (importable from postprocessing_seismo_lib, defined in utils.py), converts an ANSS Detection dict back into the legacy SCSN detection JSON format (Type/ID/Source/Hypocenter/Data/Magnitude, capitalized keys):

scsn_detection = detection_anss_to_scsn(anss_detection)
print(scsn_detection)

These two functions were ported from a separate contribution (pick-filter-spec / postprocessing-library commit 8cdb808, "writer for anss detection json initial") and are kept in their own json_writers module specifically because their input contract (AQMS rows / DataFrame) differs from the SOA-JSON-envelope converters documented above.

Hypocenter JSON writer (new)

A third json_writers module, hypocenter_json.py, builds anss-formats==0.1.3 Hypocenter JSON (a GeoJSON Feature with Point geometry). Two input paths, and an id wins when both are supplied since the database is authoritative:

• orid or evid -- looks the origin up in the AQMS origin table (joined through event.prefor for evid) via db_name/db_hostname/db_user/db_password/db_port.
• lat, lon, depth, origin_time -- builds the Hypocenter directly, no database access. origin_time accepts either an epoch number (as the origin table stores it) or an ISO8601 string.

from postprocessing_seismo_lib import dumps_hypocenter_anss, dump_hypocenter_anss

dumps_hypocenter_anss(*, orid=None, evid=None, lat=None, lon=None, depth=None, origin_time=None, db_name=None, db_hostname=None, db_user=None, db_password=None, db_port=5432) -> dict builds and returns the ANSS hypocenter JSON. dump_hypocenter_anss(*, fp, ...same keyword arguments..., **json_kwargs) does the same and writes it to a file-like object fp (mirroring json.dump).

Building directly from coordinates (no database access):

hypo = dumps_hypocenter_anss(lat=34.0, lon=-118.0, depth=8.0, origin_time=1784548800.5)
print(hypo)
'''
{'type': 'Feature', 'geometry': {'type': 'Point', 'coordinates': [-118.0, 34.0, 8.0]},
 'properties': {'originTime': '2026-07-20T12:00:00.500Z'}}
'''

Looking an origin up by orid (or evid) from the AQMS database:

hypo = dumps_hypocenter_anss(
    orid=12345,
    db_name="mydb", db_hostname="db.example.org", db_user="reader", db_password="...",
)

Both functions raise ValueError when neither input path has complete input (an id, or all of lat/lon/depth/origin_time), and LookupError when an orid/evid matches no row (or a matching row has a NULL coordinate/time).

This module also ships a CLI entry point (python -m postprocessing_seismo_lib.json_writers.hypocenter_json --orid ... --cfg db.yaml, or --lat/--lon/--depth/--time for the direct path) for ad hoc use outside the library API.

Notes

• All Pick conversion functions return a dictionary containing a "picks" key (a list); the Detection conversion functions (pickfilter_soa_to_anss_detection_format, soa_to_soa_detection_format_using_anss_libraries) return a dictionary containing a "detection" key (a single dict).
• You can directly manipulate or save this list/dict depending on your workflow.
• Ensure input files match expected formats (JSON or CSV as required).
• These utilities are especially useful for integrating:
  • PhaseNet outputs
  • ANSS datasets
  • SOA-based processing pipelines
  • pick-filter-spec / AQMS detection pipelines
  • AQMS origin-table hypocenter lookups (dumps_hypocenter_anss/dump_hypocenter_anss)