Skip to content

Module base classes

Every module of a search, e.g. a ray tracer or an image function, subclasses one of these base classes. Subclass them to extend Qseek with your own modules.

Base class Configured in Discriminator field
WaveformProvider data_provider provider
BatchPreProcessing pre_processing process
ImageFunction image_function image
Picker picker of the image function
RayTracer ray_tracers tracer
TravelTimeCorrections station_corrections corrections
Trigger trigger trigger
EventMagnitudeCalculator magnitudes magnitude
FeatureExtractor features feature
Callback callbacks callback

WaveformProvider pydantic-model

Bases: Model

Fields:

provider pydantic-field 

provider: Literal['WaveformProvider'] = 'WaveformProvider'

channel_selector pydantic-field 

channel_selector: list[Annotated[str, StringConstraints(to_upper=True, max_length=2, min_length=2)]] | None = None

Channel selector for waveforms, e.g. ['HH', 'EN'].

get_subclasses classmethod 

get_subclasses() -> tuple[type[WaveformProvider], ...]

get_squirrel 

get_squirrel() -> Squirrel

available_nsls 

available_nsls() -> set[NSL]

prepare async 

prepare(stations: StationInventory) -> None

iter_batches 

iter_batches(window_increment: timedelta, window_padding: timedelta, start_time: datetime | None = None, min_length: timedelta | None = None, min_stations: int = 0) -> AsyncGenerator[WaveformBatch]

get_traces async 

get_traces(nsls: Sequence[NSL], start_time: datetime, end_time: datetime, channel_priorities: Sequence[str] | None = None, want_incomplete: bool = False) -> list[Trace]

BatchPreProcessing pydantic-model

Bases: BaseModel

Fields:

Validators:

process pydantic-field 

process: Literal['BasePreProcessing'] = 'BasePreProcessing'

stations pydantic-field 

stations: set[NSLType] = set()

List of station codes to process. E.g. ['6E.BFO', '6E.BHZ']. If empty, all stations are processed.

n_threads pydantic-field 

n_threads: PositiveInt = 8

The number of threads processing the traces in parallel. Ignored by the DeepDenoiser.

thread_pool property 

thread_pool: ThreadPoolExecutor

The thread pool of the module, created on first use.

validate_stations pydantic-validator 

validate_stations(v) -> set[NSL]

get_subclasses classmethod 

get_subclasses() -> tuple[type[BatchPreProcessing], ...]

Returns a tuple of all the subclasses of BasePreProcessing.

filter_traces 

filter_traces(batch: WaveformBatch) -> list[Trace]

Selects traces from the given list based on the stations specified.

Parameters:

Name Type Description Default
batch WaveformBatch

The batch of traces to select from.

 required

Returns:

Type Description
list[Trace]

list[Trace]: The selected traces.

prepare async 

prepare() -> None

Prepare the pre-processing module.

process_batch async 

process_batch(batch: WaveformBatch) -> WaveformBatch

Process a list of traces.

Parameters:

Name Type Description Default
batch WaveformBatch

The batch of traces to process.

 required

Returns:

Type Description
WaveformBatch

list[Trace]: The processed list of traces.

ImageFunction pydantic-model

Bases: Model

Fields:

image pydantic-field 

image: Literal['base'] = 'base'

picker pydantic-field 

picker: Picker

name property 

name: str

get_subclasses classmethod 

get_subclasses() -> tuple[type[ImageFunction], ...]

Returns a tuple of all the subclasses of ImageFunction.

prepare async 

prepare() -> None

process_traces async 

process_traces(traces: list[Trace]) -> list[WaveformImage]

Process traces to generate image functions.

Parameters:

Name Type Description Default
traces list[Trace]

List of traces to process.

 required

Returns:

Type Description
list[WaveformImage]

list[WaveformImage]: List of image functions.

get_blinding 

get_blinding() -> timedelta

Blinding duration for the image function. Added to padded waveforms.

Returns:

Name Type Description
timedelta timedelta

The blinding duration for the image function.

get_phases 

get_phases() -> tuple[PhaseDescription, ...]

Get the phases provided by the image function.

Returns:

Type Description
tuple[PhaseDescription, ...]

tuple[PhaseDescription, ...]: The phases provided by the image function.

get_images async 

get_images(batch: WaveformBatch) -> WaveformImages

Calculate the images of a waveform batch.

Images without traces are skipped.

Parameters:

Name Type Description Default
batch WaveformBatch

Batch of waveforms.

 required

Returns:

Name Type Description
WaveformImages WaveformImages

Images of the batch.

Raises:

Type Description
ValueError

If no image has traces.

iter_images 

iter_images(batch_iterator: AsyncGenerator[WaveformBatch]) -> AsyncGenerator[tuple[WaveformImages, WaveformBatch]]

Iterate over images from batches.

The images are calculated in a background task, ahead of the consumer. Batches whose images cannot be calculated due to a ValueError are skipped, other errors are raised to the consumer. The background task is cancelled when the consumer stops iterating.

Parameters:

Name Type Description Default
batch_iterator AsyncGenerator[WaveformBatch]

Async generator of batches.

 required

Returns:

Type Description
AsyncGenerator[tuple[WaveformImages, WaveformBatch]]

AsyncGenerator[tuple[WaveformImages, WaveformBatch]]: Images and their batch.

Picker pydantic-model

Bases: Model

pick_trace 

pick_trace(trace: Trace, phase: PhaseDescription, event_time: datetime, modelled_arrival: datetime) -> ObservedArrival | None

Pick a phase arrival in a single image function trace.

Parameters:

Name Type Description Default
trace Trace

Image function trace of a station.

 required
phase PhaseDescription

Phase of the observed arrival.

 required
event_time datetime

Time of the event, picks before it are rejected.

 required
modelled_arrival datetime

Modelled arrival time to search around.

 required

Returns:

Type Description
ObservedArrival | None

ObservedArrival | None: Picked arrival, None if none found.

add_picks 

add_picks(detections: Sequence[EventDetection], images: WaveformImages) -> None

Pick the observed arrivals of the detections' receivers.

Picks are searched around the modelled arrival of each receiver's phase detection and attached as its observed arrival.

Parameters:

Name Type Description Default
detections Sequence[EventDetection]

Detections with modelled arrivals.

 required
images WaveformImages

Images the detections were located from.

 required

RayTracer pydantic-model

Bases: Model

Fields:

tracer pydantic-field 

tracer: Literal['RayTracer'] = 'RayTracer'

get_subclasses classmethod 

get_subclasses() -> tuple[type[RayTracer], ...]

prepare async 

prepare(octree: Octree, stations: StationInventory, rundir: Path | None = None)

get_available_phases 

get_available_phases() -> tuple[str, ...]

get_travel_time_location 

get_travel_time_location(phase: str, source: Location, receiver: Location) -> float

get_travel_times_locations 

get_travel_times_locations(phase: str, source: Location, receivers: Sequence[Location]) -> ndarray

get_travel_times async 

get_travel_times(phase: str, nodes: Sequence[Node], stations: Sequence[Station]) -> ndarray

Get travel times for a phase from a source to a set of stations.

Parameters:

Name Type Description Default
phase str

Phase name.

 required
nodes Sequence[Node]

Nodes to get traveltime for.

 required
stations Sequence[Station]

Stations to calculate travel times to.

 required

Returns:

Type Description
ndarray

Travel times in seconds.

get_arrivals 

get_arrivals(phase: str, event_time: datetime, source: Location, receivers: Sequence[Location]) -> list[ModelledArrival | None]

ModelledArrival dataclass

Attributes:

Name Type Description
phase str

Name of the phase
time datetime

Time of the arrival
tracer str

phase instance-attribute 

phase: str

Name of the phase

time instance-attribute 

time: datetime

Time of the arrival

tracer class-attribute instance-attribute 

tracer: str = ''

TravelTimeCorrections pydantic-model

Bases: Model

Fields:

corrections pydantic-field 

corrections: Literal['TravelTimeCorrections'] = 'TravelTimeCorrections'

n_stations property 

n_stations: int

get_subclasses classmethod 

get_subclasses() -> tuple[type[TravelTimeCorrections], ...]

Get the subclasses of this class.

Returns:

Type Description
tuple[type[TravelTimeCorrections], ...]

tuple[type]: The subclasses of this class.

get_delay 

get_delay(station_nsl: NSL, phase: PhaseDescription, node: Node | None = None) -> float

Get the traveltime delay for a station and phase.

The delay is the difference between the observed and predicted traveltime.

Parameters:

Name Type Description Default
station_nsl NSL

The station NSL.

 required
phase PhaseDescription

The phase description.

 required
node Node | None

The node to get the delay for. If None, the delay for the station is returned. Defaults to None.

 None

Returns:

Name Type Description
float float

The traveltime delay in seconds.

get_delays async 

get_delays(station_nsls: Sequence[NSL], phase: PhaseDescription, nodes: Sequence[Node]) -> ndarray

Get the traveltime delays for a set of stations and a phase.

Parameters:

Name Type Description Default
station_nsls Sequence[NSL]

The stations to get the delays for.

 required
phase PhaseDescription

The phase to get the delays for.

 required
nodes Sequence[Node]

The nodes to get the delays for.

 required

Returns:

Type Description
ndarray

np.ndarray: The traveltime delays for the given stations and phase.

prepare async 

prepare(stations: StationInventory, octree: Octree, phases: Iterable[PhaseDescription], rundir: Path) -> None

Prepare the station for the corrections.

Parameters:

Name Type Description Default
stations Stations

The station to prepare.

 required
octree Octree

The octree to use for the preparation.

 required
phases Iterable[PhaseDescription]

The phases to prepare the station for.

 required
rundir Path

The run directory of the search.

 required

Trigger pydantic-model

Bases: Model

Base class of the triggers.

A trigger detects events as peaks of the detection function, the maximum semblance over all nodes. A peak is a detection when its height and its prominence exceed the thresholds of the trigger. Subclasses set the thresholds with get_threshold.

Fields:

trigger pydantic-field 

trigger: Literal['Trigger'] = 'Trigger'

blinding pydantic-field 

blinding: timedelta = timedelta(seconds=1.0)

Minimum time between two detections. Peaks of the detection function closer than this count as one detection. Prevents detecting the same event twice.

get_subclasses classmethod 

get_subclasses() -> tuple[type[Trigger], ...]

Get the subclasses of this class.

Returns:

Type Description
tuple[type[Trigger], ...]

tuple[type[Trigger], ...]: The subclasses of this class.

get_threshold 

get_threshold(detection_function: ndarray) -> tuple[float, float]

Get the thresholds for a window of the detection function.

Parameters:

Name Type Description Default
detection_function ndarray

Detection function of the window, without padding.

 required

Returns:

Type Description
tuple[float, float]

tuple[float, float]: Minimum height and minimum prominence of a detection.

detect async 

detect(detection_function: ndarray, sampling_rate: float, padding: int = 0) -> tuple[ndarray, ndarray]

Detect events in a window of the detection function.

Peaks in the padding are discarded. The padding gives the peaks at the edges of the window their prominence.

Parameters:

Name Type Description Default
detection_function ndarray

Detection function of the window, with padding.

 required
sampling_rate float

Sampling rate of the detection function in Hz.

 required
padding int

Number of padding samples at both ends. Defaults to 0.

 0

Returns:

Type Description
tuple[ndarray, ndarray]

tuple[np.ndarray, np.ndarray]: Sample indices of the detections in the window without padding, and the detection function at the detections.

EventMagnitudeCalculator pydantic-model

Bases: Model

Fields:

magnitude pydantic-field 

magnitude: Literal['MagnitudeCalculator'] = 'MagnitudeCalculator'

min_stations pydantic-field 

min_stations: PositiveInt = 3

Minimum number of station magnitudes required to calculate the network magnitude.

exclude_stations pydantic-field 

exclude_stations: list[NSLType] = []

List of station NSLs to exclude from magnitude calculation.

get_subclasses classmethod 

get_subclasses() -> tuple[type[EventMagnitudeCalculator], ...]

Get the subclasses of this class.

Returns:

Type Description
tuple[type[EventMagnitudeCalculator], ...]

list[type]: The subclasses of this class.

has_magnitude 

has_magnitude(event: EventDetection) -> bool

Check if the given event has a magnitude.

Parameters:

Name Type Description Default
event EventDetection

The event to check.

 required

Returns:

Name Type Description
bool bool

True if the event has a magnitude, False otherwise.

get_magnitude async 

get_magnitude(waveform_provider: WaveformProvider, stations: StationInventory, event: EventDetection) -> EventMagnitude

Calculates the magnitude for the given event.

Parameters:

Name Type Description Default
waveform_provider WaveformProvider

The waveform provider.

 required
stations StationInventory

The station inventory.

 required
event EventDetection

The event detection object.

 required

Raises:

Type Description
NotImplementedError

This method is not implemented in the base class.

prepare async 

prepare(octree: Octree, stations: StationInventory) -> None

Prepare the magnitudes calculation by initializing necessary data structures.

Parameters:

Name Type Description Default
octree Octree

The octree containing seismic event data.

 required
stations Stations

The stations containing seismic station data.

 required

Raises:

Type Description
NotImplementedError

This method must be implemented by subclasses.

csv_header 

csv_header() -> list[str]

Get the CSV header for the magnitude data.

Returns:

Type Description
list[str]

list[str]: The CSV header as a list of column names.

FeatureExtractor pydantic-model

Bases: BaseModel

Fields:

feature pydantic-field 

feature: Literal['FeatureExtractor'] = 'FeatureExtractor'

get_subclasses classmethod 

get_subclasses() -> tuple[type[FeatureExtractor], ...]

Get the subclasses of this class.

Returns:

Type Description
tuple[type[FeatureExtractor], ...]

list[type]: The subclasses of this class.

add_features async 

add_features(waveform_provider: WaveformProvider, stations: StationInventory, event: EventDetection) -> None

Callback pydantic-model

Bases: Model

Base class for search lifecycle callbacks.

Fields:

callback pydantic-field 

callback: Literal['Callback'] = 'Callback'

on_start async 

on_start(search: Search) -> None

Called once after the search has been prepared.

on_stop async 

on_stop(search: Search) -> None

Called once after the search has finished.

on_batch_start async 

on_batch_start(batch: WaveformBatch) -> None

Called before a waveform batch is processed.

on_batch_end async 

on_batch_end(batch: WaveformBatch) -> None

Called after a waveform batch has been processed.

on_new_detection async 

on_new_detection(detection: EventDetection) -> None

Called for every new event detection.

get_subclasses classmethod 

get_subclasses() -> tuple[type[Callback], ...]

Get the subclasses of this class.

Returns:

Type Description
tuple[type[Callback], ...]

tuple[type[Callback], ...]: The subclasses of this class.