SeisBench

https://seisbench.readthedocs.io/en/latest/
Admin
SeisBench is an open-source Python toolbox for machine learning in seismology that provides unified...
Tokens:108,809
Snippets:713
Trust Score:9.6
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install SeisBench from Source

Source: https://seisbench.readthedocs.io/en/latest/_sources/pages/installation_and_usage.rst.txt

Install SeisBench by cloning the repository and installing locally. Append '[das]' to include DAS dependencies.

```bash
pip install .
```

--------------------------------

### Install SeisBench from source with DAS support

Source: https://seisbench.readthedocs.io/en/latest/pages/installation_and_usage.html

Install SeisBench from local source, including support for DAS data by appending `[das]` to the installation command.

```bash
pip install .[das]
```

--------------------------------

### Install PyTorch CPU Version and SeisBench

Source: https://seisbench.readthedocs.io/en/latest/_sources/pages/installation_and_usage.rst.txt

Install a pure CPU version of PyTorch first, then install SeisBench. This is useful if CUDA is not available or not needed.

```bash
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
pip install seisbench
```

--------------------------------

### Install Pyrocko and PyQt5

Source: https://seisbench.readthedocs.io/en/latest/pages/data/dataset_inspection.html

Install Pyrocko and PyQt5 for interactive waveform exploration with Snuffler. This is a prerequisite for using the `pyrocko_snuffle_event` functionality.

```bash
pip install pyrocko PyQt5

```

--------------------------------

### Get Pyrocko Picks from Sample

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/inspection.html

Converts a dataset sample's picks into a list of Pyrocko PhaseMarker objects. Requires Pyrocko to be installed. Handles missing pick data and polarity.

```python
def get_pyrocko_picks(self, sample_idx: int) -> list[PhaseMarker]:
        """
        Returns the picks of a sample as a list of pyrocko Pick objects.

        Requires pyrocko to be installed.

        :param idx: Idx of sample to return picks for
        :return: List of pyrocko Pick objects
        """
        metadata = self._metadata

        metadata = metadata.iloc[sample_idx].to_dict()
        sampling_rate = metadata["trace_sampling_rate_hz"]
        phases = [
            key.strip("trace_").strip("_arrival_sample")
            for key in metadata.keys()
            if key.endswith("_arrival_sample")
        ]
        location_code = metadata.get("station_location_code")
        location_code = "" if np.isnan(location_code) else location_code
        nsl = (
            metadata["station_network_code"],
            metadata["station_code"],
            location_code,
        )

        trace_start = datetime.fromisoformat(metadata["trace_start_time"])

        picks = []
        for phase in phases:
            if np.isnan(metadata[f"trace_{phase}_arrival_sample"]):
                continue

            pick_delay = metadata[f"trace_{phase}_arrival_sample"] / sampling_rate
            pick_time = trace_start.timestamp() + pick_delay
            automatic = metadata.get(f"trace_{phase}_status", "manual") == "automatic"

            pick = PhaseMarker(
                tmin=pick_time,
                tmax=pick_time,
                nslc_ids=[nsl + ("*",)],
                automatic=automatic,
                phasename=phase,
                polarity=_PYROCKO_POLARITY_MAP.get(
                    metadata.get(f"trace_{phase}_polarity", "Undecideable")
                ),
            )

```

--------------------------------

### Get Pyrocko Event from Sample

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/inspection.html

Converts a dataset sample's event information into a Pyrocko Event object. Requires Pyrocko to be installed. Uses metadata for event details.

```python
def get_pyrocko_event(self, sample_idx: int) -> Event:
        """
        Returns the event of a sample as a pyrocko Event object.

        Requires pyrocko to be installed.

        :param idx: Idx of sample to return event for, or event source_id
        :return: pyrocko Event object
        """

        metadata = self._metadata

        metadata = metadata.iloc[sample_idx].to_dict()
        return Event(
            name=f"{metadata['index']} ({metadata['split']})",
            time=metadata["source_origin_time"].timestamp(),
            lat=metadata["source_latitude_deg"],
            lon=metadata["source_longitude_deg"],
            depth=metadata["source_depth_km"] * 1e3,
            magnitude=metadata.get("source_magnitude", 0.0),
            magnitude_type=metadata.get("source_magnitude_type"),
            extras={
                "id": metadata["source_id"],
                "split": metadata["split"],
            },
        )
```

--------------------------------

### Install SeisBench via Pip

Source: https://seisbench.readthedocs.io/en/latest/_sources/pages/installation_and_usage.rst.txt

Use this command to install the base SeisBench package. Consider using a virtual environment.

```bash
pip install seisbench
```

--------------------------------

### Get Pyrocko Traces from Sample

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/inspection.html

Converts a dataset sample's waveforms into a list of Pyrocko Trace objects. Requires Pyrocko to be installed. Handles missing location codes.

```python
def get_pyrocko_traces(self, sample_idx: int) -> list[Trace]:
        """
        Returns the waveforms of a sample as a list of pyrocko Trace objects.

        Requires pyrocko to be installed.

        :param idx: Idx of sample to return traces for
        :return: List of pyrocko Trace objects
        """

        ds = self._dataset

        data, metadata = ds.get_sample(sample_idx)
        traces = []

        location_code = str(metadata.get("station_location_code"))
        location_code = "" if location_code == "nan" else location_code

        for ichannel, trace_data in enumerate(data):
            channel = metadata["trace_component_order"][ichannel]
            trace = Trace(
                ydata=trace_data,
                tmin=datetime.fromisoformat(metadata["trace_start_time"]).timestamp(),
                deltat=1.0 / metadata["trace_sampling_rate_hz"],
                network=metadata["station_network_code"],
                station=metadata["station_code"],
                location=location_code,
                channel=f"{metadata['trace_channel']}{channel}",
            )
            traces.append(trace)
        return traces
```

--------------------------------

### Install SeisBench with DAS Support

Source: https://seisbench.readthedocs.io/en/latest/_sources/pages/installation_and_usage.rst.txt

Install SeisBench with additional dependencies for DAS data support. This avoids installing them by default to reduce the number of dependencies.

```bash
pip install seisbench[das]
```

--------------------------------

### SteeredWindow Class

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/generate.html

Selects a window based on start and end samples defined in the 'control' dictionary, with options to specify window length or use the provided start and end samples.

```APIDOC
## SteeredWindow Class

### Description
Window selection that relies on the “control” dict from the SteeredGenerator. Selects a window of given length with the window defined by start and end sample in the middle. If no length is given, the window between start and end sample is returned. If there are insufficient samples on one side of the target window, the window will be moved. If the total number of samples is insufficient, the window will start at the earliest possible sample. The behavior in this case depends on the chosen strategy for FixedWindow.

### Parameters

#### Constructor Parameters
- **windowlen** (int or None) - Length of the window to be returned. If None, will be determined using the start and end samples from the “control” dict.
- **start_key** (str) - Key of the start sample in the “control” dict
- **end_key** (str) - Key of the end sample in the “control” dict
- **window_output_key** (str) - The sample start and end will be written as numpy array to this key in the state_dict
- **kwargs** - Parameters passed to the init method of FixedWindow.
```

--------------------------------

### Get Train, Dev, and Test Datasets

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Convenience method for returning training, development, and test sets.

```python
train_dev_test()
```

--------------------------------

### Get Train, Dev, and Test Datasets

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Returns the training, development, and test sets. This is equivalent to calling `self.train()`, `self.dev()`, and `self.test()` sequentially.

```python
self.train(), self.dev(), self.test()
```

--------------------------------

### Get Sample from Dataset

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Wraps `WaveformDataset.get_sample()`. Passes arguments to the parent function.

```python
get_sample(_idx_ , _* args_, _** kwargs_)
```

--------------------------------

### Load and Access PiSDL Subsets

Source: https://seisbench.readthedocs.io/en/latest/pages/data/benchmark_datasets.html

Demonstrates how to load the PiSDL dataset and access specific regional subsets. Ensure the seisbench library is installed.

```python
import seisbench.data as sbd

dataset = sbd.PiSDL()

# Data from the Dawson-Septimus area
dawson = dataset.get_dawson_septimus_subset()
# Data from Insehim
insheim = dataset.get_insheim_subset()
# Data from St. Gallen
st_gallen = dataset.get_st_gallen_subset()
# Data from Switzerland
switzerland = dataset.get_switzerland_subset()
# Data from the Ruhr area
floodrisk = dataset.get_floodrisk_subset()
# Data from Vendenheim
vendenheim = dataset.get_vendenheim_subset()
```

--------------------------------

### GET /plot_map

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Plots the dataset onto a map using the Mercator projection. Requires a cartopy installation.

```APIDOC
## GET /plot_map

### Description
Plots the dataset onto a map using the Mercator projection. Requires a cartopy installation.

### Parameters
#### Query Parameters
- **res** (str) - Optional - Resolution for cartopy features, defaults to 110m.
- **connections** (bool) - Optional - If true, plots lines connecting sources and stations. Defaults to false.
- **kwargs** (dict) - Optional - Plotting kwargs passed to matplotlib. Args must be prefixed with `sta_`, `ev_`, or `conn_` to address stations, events, or connections respectively.

### Response
#### Success Response (200)
- **figure** (handle) - A figure handle for the created figure.
```

--------------------------------

### Download Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/bohemia.html

Initializes client and downloads catalog and inventory data if not already present locally.

```python
    def _download_dataset(
        self,
        writer,
        time_before: float = 60.0,
        time_after: float = 60.0,
        **kwargs,
    ):
        logger.info(
            "No pre-downloaded dataset available, downloading from BGR and Geofon. "
            "This may take a while.",
        )
        if not self._eida_token:
            logger.warning(
                "No EIDA token provided. "
                "Restricted WB network data will not be accessible.",
            )
        self._init_client()
        writer.data_format = {
            "dimension_order": "CW",
            "component_order": COMPONENT_ORDER,
            "measurement": "velocity",
            "unit": "counts",
            "instrument_response": "not restituted",
        }

        cat = self.get_catalog()
        inv = self.get_inventory(cat, force_download=True)

        cat = fixup_catalog(cat, inv)
```

--------------------------------

### Initialize Iquique Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/iquique.html

Initializes the Iquique Benchmark Dataset. Includes citation and a warning about dataset size. Use when accessing the Iquique dataset for seismic event analysis.

```python
import seisbench

from .base import WaveformBenchmarkDataset




[docs]
class Iquique(WaveformBenchmarkDataset):
    """
    Iquique Benchmark Dataset of local events used for training in Woollam (2019)
    study (see citation).

    Splits are set using standard random sampling of :py:class:`seisbench.data.base.BenchmarkDataset`.
    """

    def __init__(self, **kwargs):
        citation = (
            "Woollam, J., Rietbrock, A., Bueno, A. and De Angelis, S., 2019. "
            "Convolutional neural network for seismic phase classification, "
            "performance demonstration over a local seismic network. "
            "Seismological Research Letters, 90(2A), pp.491-502. "
            "https://doi.org/10.1785/0220180312"
        )

        seisbench.logger.warning(
            "Check available storage and memory before downloading and general use "
            "of Iquique dataset. "
            "Dataset size: waveforms.hdf5 ~5Gb, metadata.csv ~2.6Mb"
        )

        super().__init__(citation=citation, repository_lookup=True, **kwargs)

    def _download_dataset(**kwargs):
        pass

```

--------------------------------

### Initialize SeisBench Dataset Writer

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Constructor for the writer class, setting up paths and default bucketing parameters.

```python
def __init__(self, metadata_path, waveforms_path):
        self.metadata_path = Path(metadata_path)
        self.waveforms_path = Path(waveforms_path)
        self.metadata_dict = {}
        self.data_format = {}

        self.metadata_path.parent.mkdir(parents=True, exist_ok=True)
        self.waveforms_path.parent.mkdir(parents=True, exist_ok=True)

        self._metadata = []
        self._waveform_file = None
        self._pbar = None

        self._bucketer = GeometricBucketer()
        self._bucket_size = 1024
        self._cache = defaultdict(list)  # Cache used for the bucketing
        self._bucket_counter = 0  # Bucket counter to generate bucket names
```

--------------------------------

### Initialize BohemiaSaxony Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/bohemia.html

CLI entry point for configuring and initializing the BohemiaSaxony dataset with an optional EIDA token.

```python
if __name__ == "__main__":
    from argparse import ArgumentParser

    parser = ArgumentParser()
    parser.add_argument(
        "--eida-token",
        type=str,
        default=None,
        help="Path to EIDA token file.",
    )
    args = parser.parse_args()
    logger.root.setLevel("INFO")
    if args.eida_token is not None:
        path = Path(args.eida_token)
        if not path.exists():
            parser.error(f"EIDA token file {path} does not exist.")
        args.eida_token = str(path.expanduser().resolve())

    ds = BohemiaSaxony(eida_token=args.eida_token, output_order="ZNE")
```

--------------------------------

### Get Test Dataset

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Convenience method for getting the test split of the dataset.

```python
test()
```

--------------------------------

### Get Training Dataset

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Convenience method for getting the training split of the dataset.

```python
train()
```

--------------------------------

### Initialize GEOFON Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/geofon.html

Initializes the GEOFON dataset. The dataset will be downloaded from the SeisBench repository on first usage. Ensure 'repository_lookup' is set to True for automatic downloading.

```python
from seisbench.data import GEOFON

dataset = GEOFON()

```

--------------------------------

### List and Load Pretrained Models

Source: https://seisbench.readthedocs.io/en/latest/pages/models/pretrained_models.html

Use these commands to discover available pretrained models and load a specific version into your environment.

```python
import seisbench.models as sbm

sbm.PhaseNet.list_pretrained()                  # Get available models
model = sbm.PhaseNet.from_pretrained("geofon")  # Load the model trained on GEOFON
```

--------------------------------

### GEOFON Dataset Initialization with Custom Parameters

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/geofon.html

Initializes the GEOFON dataset, allowing for custom download parameters. The 'basepath' must be set in 'download_kwargs' to specify the download location and start the conversion process.

```python
from seisbench.data import GEOFON

dataset = GEOFON(download_kwargs={'basepath': '/path/to/download'})

```

--------------------------------

### Get FDSN Client for ETHZ

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/ethz.html

Provides a class method to get an FDSN client instance specifically for the ETHZ data source. This client is used to query and download waveform data.

```python
from obspy.clients.fdsn import Client

@classmethod
def _fdsn_client(cls):
    return Client("ETH")

```

--------------------------------

### GET /group/waveforms

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves only the waveforms for all members of a specific group.

```APIDOC
## GET /group/waveforms

### Description
Returns the waveforms for each member of a group.

### Method
GET

### Parameters
#### Path Parameters
- **idx** (int) - Required - Group index

### Response
#### Success Response (200)
- **waveforms** (list) - List of waveforms
```

--------------------------------

### GET /available_chunks

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Determines the chunks of the dataset in the given path.

```APIDOC
## GET /available_chunks

### Description
Determines the chunks of the dataset in the given path.

### Parameters
#### Query Parameters
- **path** (str) - Required - Dataset path

### Response
#### Success Response (200)
- **chunks** (list[str]) - List of chunks
```

--------------------------------

### Implement Dataset Download and Conversion

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/lendb.html

The _download_dataset method handles fetching the original HDF5 file and iterating through earthquake and noise entries to populate the SeisBench writer.

```python
def _download_dataset(self, writer, cleanup=False):
        """
        Downloads and converts the dataset from the original publication

        :param writer: WaveformWriter
        :param cleanup: If true, delete the original hdf5 file after conversion. Defaults to false.
        :return:
        """
        path = self.path
        path.mkdir(parents=True, exist_ok=True)

        path_original = path / "LEN-DB.hdf5"

        # Uses callback_if_uncached only to be able to utilize the cache mechanism
        # Concurrent accesses are anyhow already controlled by the callback_if_uncached call wrapping _download_dataset
        # It's therefore considered save to set force=True
        def callback_download_original(path):
            seisbench.util.download_http(
                "https://zenodo.org/record/3648232/files/LEN-DB.hdf5?download=1",
                path,
                desc="Downloading original dataset",
            )

        seisbench.util.callback_if_uncached(
            path_original, callback_download_original, force=True
        )

        writer.data_format = {
            "dimension_order": "CW",
            "component_order": "ZNE",
            "measurement": "velocity",
            "sampling_rate": 20,
            "unit": "km/s",
            "instrument_response": "restituted",
        }

        with h5py.File(path_original, "r") as f:
            # Set total number of traces for progress bar
            writer.set_total(len(f["AN"].keys()) + len(f["EQ"].keys()))

            # Write EQs (Earthquakes)
            for eq_name, eq_data in f["EQ"].items():
                network, station, _ = eq_name.split("_")
                eq_attributes = dict(eq_data.attrs)

                starttime = str(UTCDateTime(eq_attributes["starttime"]))
                otime = str(UTCDateTime(eq_attributes["otime"]))

                metadata = {
                    "trace_name": eq_name,
                    "trace_start_time": starttime,
                    "trace_category": "earthquake",
                    "trace_p_arrival_sample": 80,
                    "trace_p_status": "estimated",
                    "station_code": station,
                    "station_network_code": network,
                    "station_latitude_deg": eq_attributes["stla"],
                    "station_longitude_deg": eq_attributes["stlo"],
                    "station_elevation_m": eq_attributes["stel"],
                    "source_magnitude": eq_attributes["mag"],
                    "source_latitude_deg": eq_attributes["evla"],
                    "source_longitude_deg": eq_attributes["evlo"],
                    "source_depth_km": eq_attributes["evdp"] / 1e3,
                    "source_origin_time": otime,
                    "path_ep_distance_km": eq_attributes["dist"] / 1e3,
                    "path_azimuth_deg": eq_attributes["az"],
                    "path_back_azimuth_deg": eq_attributes["baz"],
                    "split": self._get_split_from_time(starttime),
                }

                writer.add_trace(metadata, eq_data[()])

            # Write ANs (Noise)
            for an_name, an_data in f["AN"].items():
                network, station, _ = an_name.split("_")
                an_attributes = dict(an_data.attrs)

                starttime = str(UTCDateTime(an_attributes["starttime"]))

                metadata = {
                    "trace_name": an_name,
                    "trace_start_time": starttime,
                    "trace_category": "noise",
                    "station_code": station,
                    "station_network_code": network,
                    "station_latitude_deg": an_attributes["stla"],
                    "station_longitude_deg": an_attributes["stlo"],
                    "station_elevation_m": an_attributes["stel"],
                    "split": self._get_split_from_time(starttime),
                }

                writer.add_trace(metadata, an_data[()])

        if cleanup:
            # Remove original dataset
            os.remove(path_original)
```

--------------------------------

### GET /group/samples

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves both waveforms and metadata for all members of a specific group.

```APIDOC
## GET /group/samples

### Description
Returns the waveforms and metadata for each member of a group.

### Method
GET

### Parameters
#### Path Parameters
- **idx** (int) - Required - Group index

### Response
#### Success Response (200)
- **waveforms** (list) - List of waveforms
- **metadata** (list) - List of metadata dicts
```

--------------------------------

### GET /get_event_sample_indices

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/data/base.html

Returns the indices of all samples associated with a given event.

```APIDOC
## GET /get_event_sample_indices

### Description
Returns the indices of all samples associated with a given event. Requires that source_id is part of the metadata.

### Parameters
#### Query Parameters
- **event_id** (str) - Required - Event identifier

### Response
#### Success Response (200)
- **indices** (list[int]) - List of indices associated with the event
```

--------------------------------

### Germany Class

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/util.html

Example usage of how to create more complex region geometries.

```APIDOC
## Germany Class

### Description
Example usage of how to create more complex region geometries.

### Methods
#### get_query_parameters
Return the domain specific query parameters for the `get_stations()` method as a dictionary.

Returns:
- dict - Possibilities keys for rectangular queries are `minlatitude`, `maxlatitude`, `minlongitude`, `maxlongitude`. For circular queries: `latitude`, `longitude`, `minradius`, `maxradius`.

#### is_in_domain
Checks whether a point is within the domain.

Parameters:
- **latitude** (float) - Latitude of query point
- **longitude** (float) - Longitude of query point

Returns:
- bool - True if point is within the domain, false otherwise
```

--------------------------------

### GET get_sample

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves a single sample from the dataset based on the provided index.

```APIDOC
## GET get_sample

### Description
Retrieves a specific sample from the dataset by its index.

### Parameters
#### Query Parameters
- **idx** (int) - Required - Index of the sample to retrieve.

### Response
#### Success Response (200)
- **sample** (object) - The requested waveform sample.
```

--------------------------------

### Configure FDSN Clients

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/bohemia.html

Internal method to initialize multi-client support for BGR, GEOFON, and LMU data sources.

```python
def _init_client(self):
        self._client = MultiClient()
        self._client.add_client(Client("BGR"))
        self._client.add_client(Client("GEOFON", eida_token=self._eida_token))
        self._client.add_client(Client("LMU"))
```

--------------------------------

### GET /get_event_source_id

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves the source ID for a specific event based on its index.

```APIDOC
## GET /get_event_source_id

### Description
Gets the source_id of an event. The idx refers to the integer index of the event in order of appearance in the metadata after removing duplicates.

### Method
GET

### Parameters
#### Query Parameters
- **idx** (int) - Required - Event index

### Response
#### Success Response (200)
- **source_id** (str) - Source ID of the event
```

--------------------------------

### Configure Progress Tracking and Finalize HDF5

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Methods to initialize progress bars and finalize HDF5 file writing by flushing the cache and saving metadata.

```python
    def set_total(self, n):
        """
        Set the total number of traces to write. Only used for correct progress calculation

        :param n: Number of traces
        :type n: int
        :return: None
        """
        if self._pbar is None:
            self._pbar = tqdm(desc="Traces converted")
        self._pbar.total = n

    def _finalize(self):
        """
        Finalizes the dataset, by flushing the remaining traces to hdf5 and writing metadata and data format.
        """
        self.flush_hdf5()

        if len(self._metadata) == 0:
            return

        if len(self.data_format) > 0:
            g_data_format = self._waveform_file.create_group("data_format")
            for key in self.data_format.keys():
                g_data_format.create_dataset(key, data=self.data_format[key])
        else:
            seisbench.logger.warning("No data format options specified.")

        metadata = pd.DataFrame(self._metadata)
        if self.metadata_dict is not None:
            metadata.rename(columns=self.metadata_dict, inplace=True)

        metadata.to_csv(self.metadata_path, index=False)

    def flush_hdf5(self):
        """
        Writes out all traces currently in the cache to the hdf5 file.
        Should be called if no more traces for the existing buckets will be added, e.g., after finishing a split.
        Does not write the metadata to csv.
        """
        buckets = list(self._cache.keys())

        for bucket in buckets:
            self._write_bucket(self._cache[bucket])
            del self._cache[bucket]
```

--------------------------------

### GET /n_events

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves the total number of unique events present in the dataset.

```APIDOC
## GET /n_events

### Description
Returns the number of unique events in the dataset. Requires that source_id is part of the metadata.

### Method
GET

### Response
#### Success Response (200)
- **count** (int) - Number of unique events
```

--------------------------------

### SeisBench Models Overview

Source: https://seisbench.readthedocs.io/en/latest/pages/documentation/models/overview.html

Overview of the model architecture and available implementations within the SeisBench library.

```APIDOC
## SeisBench Models Reference

### Description
The `seisbench.models` module provides a collection of pre-trained and base models for seismic data analysis, including waveform-based detection and phase picking, as well as DAS-specific models.

### Model Categories
- **Base classes**: General and WaveformModels, DASModels
- **Waveform Models**: BasicPhaseAE, CRED, DeepDenoiser, SeisDAE, Depth phase models, EQTransformer, EQTP, GPD, LFEDetect, OBSTransformer, PhaseNet, PickBlue, Skynet
- **DAS Models**: DASWaveformModelWrapper, DeepSubDAS
```

--------------------------------

### Initialize ETHZ Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/ethz.html

Initializes the ETHZ dataset, providing citation information and a warning about storage requirements. The dataset can be downloaded from a remote repository or directly from the source via an FDSN client.

```python
import seisbench
from seisbench.data.ethz import ETHZ

# Initialize the dataset
dataset = ETHZ()

```

--------------------------------

### GET /seisbench/data/crew/available_chunks

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/crew.html

Retrieves a list of all available data chunks for the CREW dataset.

```APIDOC
## GET /seisbench/data/crew/available_chunks

### Description
Returns a list of strings representing the available data chunks for the CREW dataset.

### Method
GET

### Endpoint
/seisbench/data/crew/available_chunks

### Response
#### Success Response (200)
- **chunks** (list of strings) - A list of chunk identifiers (e.g., "000", "001", ... "156").

#### Response Example
[
  "000",
  "001",
  "002"
]
```

--------------------------------

### GET /group/size

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves the number of samples contained within a specific group index.

```APIDOC
## GET /group/size

### Description
Returns the number of samples in a specified group.

### Method
GET

### Parameters
#### Path Parameters
- **idx** (int) - Required - Group index

### Response
#### Success Response (200)
- **size** (int) - Size of the group
```

--------------------------------

### Initialize BohemiaSaxony Dataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/bohemia.html

The class constructor accepts an optional EIDA token for accessing restricted network data.

```python
def __init__(self, eida_token: str | None = None, **kwargs):
        citation = """
The data is compiled from data of the networks from Saxony Seismic Network (SX),
West Bohemia Local Seismic Network (WEBNET; WB), Thuringia Seismological Network (TH),
and Czech Regional Seismic Network (CZ) and is provided by the BGR (Bundesanstalt für
Geowissenschaften und Rohstoffe) and GFZ GEOFON.

The dataset includes restricted data from the WB network, which requires an EIDA token
for access. Please provide the path to your EIDA token file when initializing the
dataset via the `eida_token` argument.

Catalog and Picks:
* Earthquakes in Saxony (Germany) and surroundings
  from 2006 to 2024 -- onsets and locations,
  https://opara.zih.tu-dresden.de/items/5387886f-25f2-4faf-8dca-33981d898ab9

Seismic Networks:
* SXNET Saxon Seismic Network, https://doi.org/10.7914/SN/SX
* West Bohemia Local Seismic Network (WEBNET), https://doi.org/10.7914/SN/WB
* Thüringer Seismologisches Netz, https://doi.org/10.7914/SN/TH
* Czech Regional Seismic Network, https://doi.org/10.7914/SN/CZ
"""
        self._eida_token = eida_token

        super().__init__(
            citation=citation,
            license="CC0-1.0",
            repository_lookup=False,
            **kwargs,
        )
```

--------------------------------

### Get Group Index from Parameters

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves the index of a group based on provided parameters.

```python
    def get_group_idx_from_params(self, params):
        """
        Returns the index of the group identified by the params.

        :param params: The parameters identifying the group. For a single grouping parameter, this argument will be a
                       single value. Otherwise this argument needs to be a tuple of keys.
        :return: Index of the group
        :rtype: int
        """
        self._verify_grouping_defined()

        if params in self._groups_to_group_idx:
            return self._groups_to_group_idx[params]
        else:
            raise KeyError("The dataset does not contain the requested group.")
```

--------------------------------

### Getting Number of Metadata Entries

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Returns the total number of metadata entries available in the dataset.

```python
return len(self.metadata)
```

--------------------------------

### Handle dataset download wrapper

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Provides downward compatibility for datasets with varying download signatures.

```python
    def _download_dataset_wrapper(self, files: list[Path], chunk: str, **kwargs):
        """
        Function only exists for downward compatibility, allowing flexibility in two cases:

        - Datasets without a ``chunk`` argument in ``download_dataset``.
        - Datasets with a different signature for ``download_dataset``, such as the ``WaveformDataWriter`` passed to
          ``BenchmarkDataset``.

        """
        tmp_download_args = self.add_chunk_to_download_args(chunk, kwargs)

        self._download_dataset(files, **tmp_download_args)
```

--------------------------------

### Get Dataset Path

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Returns the immutable path of the dataset. Raises ValueError if the path is not set.

```python
@property
    def path(self):
        """
        Path of the dataset (immutable)
        """
        if self._path is None:
            raise ValueError("Path is None. Can't create data set without a path.")
        return Path(self._path)
```

--------------------------------

### Method: get_sample

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/das_base.html

Loads a specific sample from the dataset by index, allowing control over whether data is loaded into memory or returned as virtual pointers.

```APIDOC
## Method: get_sample

### Description
Load the sample with the given index. Use the record_virtual and annotations_virtual arguments to control whether the record and annotations are loaded into memory or only pointers are returned.

### Parameters
#### Path Parameters
- **idx** (int) - Required - Index of the sample to load.

#### Query Parameters
- **record_virtual** (bool) - Optional - If true, the record is returned as a virtual array. Otherwise, the record is loaded into memory. Default: True.
- **annotations_virtual** (bool) - Optional - If true, the annotations are returned as virtual arrays. Otherwise, the annotations are loaded into memory. Default: False.

### Response
- **Returns** (tuple[dict[str, Any], NPArrayOrHDF5Dataset, dict[str, NPArrayOrHDF5Dataset]]) - A tuple containing metadata, the record data, and annotations.
```

--------------------------------

### Get Sample from MultiDASDataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/das_base.html

Retrieves a sample from the MultiDASDataset by resolving the index across its constituent datasets.

```python
def get_sample(self, idx: int, *args, **kwargs):
        dataset_idx, local_idx = self._resolve_idx(idx)
        return self.datasets[dataset_idx].get_sample(local_idx, *args, **kwargs)
```

--------------------------------

### Handle Dataset Download

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/isc_ehb.html

Raises a NotImplementedError, indicating that downloading the dataset directly from the source is not implemented. Users should access the dataset via the SeisBench repository.

```python
    def _download_dataset(self, *args, **kwargs):
        raise NotImplementedError(
            "Downloading from source is not implemented. "
            "However, this dataset is available in the SeisBench repository. "
            "Please verify you can access the repository at: "
            f"{self._remote_path()}."
        )
```

--------------------------------

### Get Dataset License

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves the license associated with the dataset. This property indicates the terms under which the dataset can be used.

```python
@property
def license(self):
    """
    The license attached to this dataset
    """
    return self._license
```

--------------------------------

### Configure Backup Remote Repository in Config File

Source: https://seisbench.readthedocs.io/en/latest/_sources/pages/installation_and_usage.rst.txt

Add this line to your SeisBench config file (e.g., .seisbench/config.json) to permanently use the backup remote repository for datasets and model weights.

```json
"remote_root": "https://seisbench.gfz.de/mirror/"
```

--------------------------------

### Download ETHZ event metadata

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/ethz.html

Fetches event data from the ETHZ FDSN service and saves it as a QuakeML file.

```python
    def _download_ethz_events_xml(
        self,
        starttime=obspy.UTCDateTime(2013, 1, 1),
        endtime=obspy.UTCDateTime(2021, 1, 1),
        minmagnitude=1.5,
    ):
        """
        Download QuakeML data from FDSN for all events satisfying defined parameters.
        QuakeML structure is also stored in cache.

        :param starttime: Define start time of window to select events, defaults to obspy.UTCDateTime(2013, 1, 1)
        :type starttime: obspy.core.UTCDateTime, optional
        :param endtime: Define start time of window to select events, defaults to obspy.UTCDateTime(2021, 1, 1)
        :type endtime: obspy.core.UTCDateTime, optional
        :param minmagnitude: Select all events with event mag > minmagnitude, defaults to 1.5
        :type minmagnitude: float, optional
        :return: Catalog containing all selected events
        :rtype: obspy.core.Catalog

        """
        query = (
            f"http://arclink.ethz.ch/fdsnws/event/1/query?"
            f"starttime={starttime.isoformat()}&endtime={endtime.isoformat()}"
            f"&minmagnitude={minmagnitude}&format=text"
        )
        resp = requests.get(query)
        ev_ids = [
            line.decode(sys.stdout.encoding).split("|")[0]
            for line in resp._content.splitlines()[1:]
        ]

        catalog = obspy.Catalog(events=[])
        with tqdm(
            desc="Downloading quakeml event meta from FDSNWS", total=len(ev_ids)
        ) as pbar:
            for ev_id in ev_ids:
                catalog += self.client.get_events(eventid=ev_id, includearrivals=True)
                pbar.update()
        catalog.write(str(self.path / "ethz_events.xml"), format="QUAKEML")

        return catalog
```

--------------------------------

### Get Internal Group

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Retrieves waveform data for a specific group without relying on metadata lookup.

```python
def _get_group_internal(self, idx, return_metadata, sampling_rate=None):
    """
    This function does *not* use the metadata_lookup.
    """
    self._verify_grouping_defined()

    group = self._groups[idx]
    idx = self._groups_to_trace_idx[group]

    metadata = self.metadata.iloc[idx].to_dict("list")

    sampling_rate = self._get_sample_unify_sampling_rate(metadata, sampling_rate)

    lookups = defaultdict(list)
    for query_idx, i in enumerate(idx):
        dataset_idx, _ = self._resolve_idx(i)
        lookups[dataset_idx].append(query_idx)

    waveforms_pre = {}
    for dataset_idx, query_idx in lookups.items():
        sub_metadata = {k: np.asarray(v)[query_idx] for k, v in metadata.items()}
        waveforms_pre[dataset_idx] = self.datasets[
            dataset_idx
        ]._get_waveforms_from_load_metadata(sub_metadata, sampling_rate, pack=False)
    waveforms_pre = {k: iter(v) for k, v in waveforms_pre.items()}

    waveforms = []
    for i in idx:
        dataset_idx, local_idx = self._resolve_idx(i)
        waveforms.append(next(waveforms_pre[dataset_idx]))

    self._calculate_trace_npts_group(metadata, waveforms)

    if return_metadata:
        return waveforms, metadata
    else:
        return waveforms
```

--------------------------------

### Retrieve train, dev, and test sets

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/das_base.html

Convenience method to return the training, development, and test datasets simultaneously.

```python
>>> self.train(), self.dev(), self.test()
```

--------------------------------

### Get Index from Trace Name

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Finds the index of a trace using its name, chunk, and dataset identifiers.

```python
    def get_idx_from_trace_name(self, trace_name, chunk=None, dataset=None):
        """
        Returns the index of a trace with given trace_name, chunk and dataset.
        Chunk and dataset parameters are optional, but might be necessary to uniquely identify traces for
        chunked datasets or for :py:class:`MultiWaveformDataset`.
        The method will issue a warning *the first time* a non-uniquely identifiable trace is requested.
        If no matching key is found, a `KeyError` is raised.

        :param trace_name: Trace name as in metadata["trace_name"]
        :type trace_name: str
        :param chunk: Trace chunk as in metadata["trace_chunk"]. If None this key will be ignored.
        :type chunk: None
        :param dataset: Trace dataset as in metadata["trace_dataset"]. Only for :py:class:`MultiWaveformDataset`.
                        If None this key will be ignored.
        :type dataset: None
        :return: Index of the sample
        """
        dict_key = "name"
        search_key = [trace_name]
        if chunk is not None:
            dict_key += "_chunk"
            search_key.append(chunk)
        if dataset is not None:
            dict_key += "_dataset"
            search_key.append(dataset)

        search_key = tuple(search_key)

        if not self._trace_identification_warning_issued and len(
            self._trace_name_to_idx[dict_key]
        ) != len(self.metadata):
            seisbench.logger.warning(
```

--------------------------------

### Create README for Dataset Dump

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/inspection.html

Generates a README.md file for a dataset dump, including information on data organization and how to open it with Pyrocko Snuffler.

```python
readme.write_text(
            f"""
# SeisBench MiniSEED dump of dataset {ds.name}

This directory contains a MiniSEED dump of the SeisBench dataset '{ds.name}'.
The dataset contains {len(ds)} earthquake records associated with {n_events} events.

The data is organized as follows:
- `mseed/`: Contains MiniSEED files, one file per day.
- `picks/`: Contains pick files in pyrocko .picks format, one file per day.
- `events.yaml`: Contains all events in pyrocko .yaml format.
- `stations.yaml`: Contains all stations in pyrocko .yaml format.
- `all_picks.picks`: Contains all picks in pyrocko .picks format

## Open Pyrocko Snuffler

You can open the data in pyrocko's snuffler using the following command:
```bash
squirrel snuffler -a mseed/
```

Use `N` to navigate between events.

## Dataset Citation
{getattr(ds, "_citation") if hasattr(ds, "_citation") else "No citation provided."}

## Dataset License
{getattr(ds, "_license") if hasattr(ds, "_license") else "No license provided."}

Created with SeisBench {seisbench.__version__}
""")
```

--------------------------------

### Determine Data Split

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/lendb.html

Helper method to determine the dataset split based on the trace start time.

```python
@staticmethod
    def _get_split_from_time(starttime):
        train_dev_border = "2017-01-16"
```

--------------------------------

### Define abstract download dataset method

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Base method for converting datasets to SeisBench format, intended to be overridden by subclasses.

```python
    def _download_dataset(self, files: list[Path], chunk: str, **kwargs):
        """
        Download and convert the dataset to the standard SeisBench format.

        :param files: List of file paths for the output files. Uses the same order as the ``_files`` list in the class.
        :param chunk: The chunk to be downloaded. Can be ignored if an unchunked data set is created.
        :param kwargs:
        :return: None
        """
        raise NotImplementedError(
            "This dataset does not implement a conversion from source."
        )
```

--------------------------------

### Get Public Dataset Name

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Returns the public name of the dataset. For BenchmarkDatasets, this always matches the class name.

```python
@property
def name(self):
    """
    Name of the dataset. For BenchmarkDatasets, always matches the class name.
    """
    return self._name_internal()
```

--------------------------------

### Initialize MultiWaveformDataset

Source: https://seisbench.readthedocs.io/en/latest/_modules/seisbench/data/base.html

Constructs a MultiWaveformDataset from a list of WaveformDataset objects. Checks for compatibility and copies input datasets.

```python
    def __init__(self, datasets):
        """
        A :py:class:`MultiWaveformDataset` is an ordered collection of :py:class:`WaveformDataset`.
        It exposes mostly the same API as a single :py:class:`WaveformDataset`.

        The constructor checks for compatibility of `dimension_order`, `component_order` and `sampling_rate`.
        The caching strategy of each contained dataset is left unmodified,
        but a warning is issued if different caching schemes are found.

        :param datasets: List of :py:class:`WaveformDataset`.
                         The constructor will create a copy of each dataset using the
                         :py:func:`WaveformDataset.copy` method.
        """
        if not isinstance(datasets, list) or not all(
            isinstance(x, WaveformDataset) for x in datasets
        ):
            raise TypeError(
                "MultiWaveformDataset expects a list of WaveformDataset as input."
            )

        if len(datasets) == 0:
            raise ValueError("MultiWaveformDatasets need to have at least one member.")

        self._datasets = [dataset.copy() for dataset in datasets]
        self._metadata = pd.concat(x.metadata for x in datasets)

        # Identify dataset
        self._metadata["trace_dataset"] = sum(
            ([dataset.name] * len(dataset) for dataset in self.datasets), []
        )
        self._metadata.reset_index(inplace=True, drop=True)

        self._trace_identification_warning_issued = (
            False  # Traced whether warning for trace name was issued already
        )

        self._homogenize_dataformat(datasets)
        self._grouping = None
        self._homogenize_grouping(datasets)
        self._build_trace_name_to_idx_dict()
```

--------------------------------

### SeisBench Utility Modules

Source: https://seisbench.readthedocs.io/en/latest/genindex.html

Overview of utility modules for SeisBench, including helper functions and tools.

```APIDOC
## SeisBench Utility Modules

This section lists the utility modules in SeisBench, providing various helper functions.

### Modules
- seisbench.util.annotations
- seisbench.util.arraytools
- seisbench.util.auxiliary
- seisbench.util.file
- seisbench.util.region
- seisbench.util.torch_helpers
- seisbench.util.trace_ops
```
Tokens:
Raw