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
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 async 

iter_batches(
    window_increment: timedelta,
    window_padding: timedelta,
    start_time: datetime | None = None,
    min_length: timedelta | None = None,
    min_stations: int = 0,
) -> AsyncIterator[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[NSL] = set()

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

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 async 

iter_images(
    batch_iterator: AsyncIterator[WaveformBatch],
) -> AsyncIterator[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 AsyncIterator[WaveformBatch]

Async iterator over batches.

 required

Yields:

Type Description
AsyncIterator[tuple[WaveformImages, WaveformBatch]]

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

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[NSL] = []

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(
    squirrel: Squirrel, 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.