Mondaic

salvus.flow.collections.event_data

salvus.flow.collections.event_data salvus flow collections event_data

The EventData object which is an Event object with waveform data and/or other types of attached data.

Functions

plot_event_data_gathers()

def plot_event_data_gathers(
    event_data_objects: list[EventData],
    receiver_field: str,
    component: str,
    event_data_labels: str | list[str] | None = None,
    sort_by: typing.Callable | None = None,
    filter_by: typing.Callable[[_Base], bool] | None = None,
    select_by: (
        typing.Callable[
            [list[tuple[_Base, obspy.Stream]], EventData],
            list[tuple[_Base, obspy.Stream]],
        ]
        | None
    ) = None,
    interleave: bool = False,
    colors: list[typing.Any] | None = None,
    alpha: float | None = None,
    plot_using: PlotProperties | mpl.axes.Axes = PlotProperties(
        width=11.5, height=6.33, dpi=100
    ),
    fast_unsafe: bool = False,
) -> mpl.axes.Axes: ...

Plot a custom gather using a list of event_data objects.

Parameters
  • event_data_objects list[EventData] — A list of the event data objects to use.
  • receiver_field str — The receiver field to plot.
  • component str — The component to plot.
  • event_data_labels str | list[str] | None — Data name or list of data names to plot. This variable will simply be used for the labelling of the data in the event_data_objects list. If not specified, names will be generated from the event_data_objects list.
  • sort_by typing.Callable | None — Callback function called with the receiver objects to determine the sorting. Works like the key callback for the standard Python sorted() method.
  • filter_by typing.Callable[[_Base], bool] | None — Callback function called with the receiver objects in internal Salvus format to determine the filtering. Works like the callback for the standard Python filter() method. For complex filter operations requiring all receivers, defer to the select_by method.
  • select_by typing.Callable[[list[tuple[_Base, obspy.Stream]], EventData], list[tuple[_Base, obspy.Stream]]] | None — Callback function called with a list of tuples in the receiver objects in both internal Salvus and ObsPy Stream format, as well as the EventData objects.
  • interleave bool — Whether or not to alternatingly plot data. If false, traces from all data will be plotted on top of each other.
  • colors list[typing.Any] | None — The colors for the individual traces. Must be exactly as many colors as data items. If not given, it will attempt to choose suitable colors.
  • alpha float | None — The opacity of all data, as used in Matplotlib. useful when setting interleave to false.
  • plot_using PlotProperties | mpl.axes.Axes — The configuration of the axis to create or reuse. Will default to the standard size as defined in the PlotProperties dataclass.
  • fast_unsafe bool — Collect data on the largest instead of smallest time step. Will potentially alias data if largest time step of the event data objects is larger than the highest relevant frequencies across the data, but does speed up plotting in cases where time axes are restrictive, e.g. simulations with relatively many time steps, which would normally lead to massive interpolation expense.
Returns mpl.axes.Axes

Classes

EventData

class EventData(salvus.flow.collections.event.Event):
    def __init__(
        self,
        event: Event,
        data: pathlib.Path | dict | xr.Dataset | xr.DataArray,
        data_hash: str | None = None,
        path_to_meta_json: pathlib.Path | None = None,
        path_to_job_info_json: pathlib.Path | None = None,
    ): ...

An event object that also contains waveform data.

Parameters
  • event Event — The event object.
  • data pathlib.Path | dict | xr.Dataset | xr.DataArray — Path to the actual waveform data or a dictionary describing the data.
  • data_hash str | None — Uniquely identifying hash of that pieces of data. Might not always be available.
  • path_to_meta_json pathlib.Path | None — Optionally specify the meta.json path of the run. Necessary if you later want to compute adjoint sources.
  • path_to_job_info_json pathlib.Path | None — Optionally specify the job_info.json path of the run.
Attributes
data _Backends

Access the actual data via a proxy that abstracts different file formats.

file_format str

Returns the file format, either "hdf5", "xarray", or "asdf".

is_synthetic bool

Is the data from a Salvus simulation?

job_info_json_contents dict | None

Contents of the job_info.json file.

meta_json_contents dict | None

Contents of the meta.json file.

path_to_meta_json pathlib.Path | None

Path to the meta.json file for the simulation that created the data.

Can be None for observed data.

receiver_channels list[ReceiverChannel]

The receiver channels for this event.

receiver_name_list list[str]

Return a list of receiver names for this event.

receivers list[simple_config.receiver._Base]

The receivers for this event.

sources list[simple_config.source._Base]

The sources for this event.

Methods
from_output_folder()
def from_output_folder(output_folder: pathlib.Path) -> EventData: ...

Construct an EventData object from a folder containing Salvus outputs.

Parameters
  • output_folder pathlib.Path — Folder where the output files are stored.
Returns EventData
from_output_hdf5_file()
def from_output_hdf5_file(filename: pathlib.Path) -> EventData: ...

Construct an EventData object from an HDF5 file containing Salvus outputs.

Parameters
  • filename pathlib.Path — File where the output data are stored.
Returns EventData
compute_data_hash()
def compute_data_hash(self) -> str: ...

Compute a unique hash for the event data.

Returns str
delete_associated_job_and_data()
def delete_associated_job_and_data(self) -> None: ...

Delete the associated Salvus job, if it still exists, and any potentially existing remote outputs.

Returns None
download_extra_outputs()
def download_extra_outputs(
    self,
    data_types_to_download: (
        typing.Collection[typing.Literal["volume", "surface"]]
        | typing.Literal["volume", "surface"]
    ) = frozenset({"volume", "surface"}),
    localpath: pathlib.Path | str | None = None,
) -> None: ...

Download extra outputs from the simulation if they don’t locally exist yet.

Parameters
  • data_types_to_download typing.Collection[typing.Literal['volume', 'surface']] | typing.Literal['volume', 'surface'] — The data types to download. Defaults to downloading everything.
  • localpath pathlib.Path | str | None — By default, downloads will be stored next to the meta.json of the EventData object. If this is set to a folder (that must exist) it will instead download to that folder and only put symlinks in the meta.json file’s folder.
Returns None
expected_components_from_receiver()
def expected_components_from_receiver(
    receiver: simple_config.receiver._Base | dict, receiver_field: str
) -> list[str]: ...

Get the expected components from a receiver object and field name.

Parameters
  • receiver simple_config.receiver._Base | dict — The receiver object or dictionary representation.
  • receiver_field str — The receiver field.
Returns list[str]
get_associated_salvus_job()
def get_associated_salvus_job(self) -> SalvusJob: ...

Get the associated Salvus job, if it still exists.

Returns SalvusJob
get_data_cube()
def get_data_cube(
    self,
    receiver_field: str,
    component: str,
    sort_by: typing.Callable[[_Base], int] | None = None,
    filter_by: typing.Callable[[_Base], bool] | None = None,
    select_by: (
        typing.Callable[
            [list[tuple[_Base, obspy.core.stream.Stream]], EventData],
            list[tuple[_Base, obspy.core.stream.Stream]],
        ]
        | None
    ) = None,
    ignore_deprecation: bool = False,
) -> tuple[np.ndarray, np.ndarray]: ...

Deprecated method, please use get_waveform_data_xarray for similar functionality. Get a data cube for a given field and component. Useful for plotting shotgathers and wiggle plots.

Returns a 2-D array with dimensions of (number of receivers, npts). Sorting, filtering, and composed operations are available through sort_by(), filter_by() and select_by(). In absence of these, receivers will be sorted as in the event object, unless sort_by or select_by are altering the sort order.

Parameters
  • receiver_field str — Receiver field to retrieve.
  • component str — The component to retrieve. Note that for legacy reasons, this is always converted to an uppercase string, to mirror ObsPy behaviour, which was case-insensitive.
  • sort_by typing.Callable[[_Base], int] | None — Callback function called with the receiver objects to determine the sorting. Works like the key callback for the standard Python sorted() method.
  • filter_by typing.Callable[[_Base], bool] | None — Callback function called with the receiver objects in internal Salvus format to determine the filtering. Works like the callback for the standard Python filter() method. For complex filter operations requiring all receivers, defer to the select_by method.
  • select_by typing.Callable[[list[tuple[_Base, obspy.core.stream.Stream]], EventData], list[tuple[_Base, obspy.core.stream.Stream]]] | None — Callback function called with a list of tuples in the receiver objects in both internal Salvus and ObsPy Stream format, as well as the EventData object.
  • ignore_deprecation bool — Hide the deprecation warning.
Returns tuple[np.ndarray, np.ndarray]
get_receiver()
def get_receiver(self, receiver_name: str) -> simple_config.receiver._Base: ...

Returns a receiver object by name.

The receiver name should be NET.STA.LOC - if LOC is omitted it will be assumed to be an empty string.

Parameters
  • receiver_name str — The name of the receiver.
Returns simple_config.receiver._Base
get_receiver_channel()
def get_receiver_channel(
    self, receiver_channel_name: str
) -> ReceiverChannel: ...

Returns a receiver channel object by its identifier.

Parameters
  • receiver_channel_name str — The name of the receiver channel.
Returns ReceiverChannel
get_receiver_data()
def get_receiver_data(
    self,
    receiver_name: str,
    receiver_field: str,
    header: typing.Any | None = None,
    _ignore_temporal_weights: bool = False,
    _use_cache: bool = False,
) -> obspy.Stream: ...

Please use .get_waveform_data() instead.

Parameters
  • receiver_name str — The name of the receiver
  • receiver_field str — The field to get.
  • header typing.Any | None — Manually set the tr.stats.header to that value. Sometimes useful to have access to extra meta data in the processing and data selection functions.
  • _ignore_temporal_weights bool — Ignore temporal weights, even if they are set. Useful mainly for plotting and debugging purposes. Don’t set this if you are not aware of the consequences.
  • _use_cache bool — Utilize a cache for this function. Useful for certain use cases with repeated per-station access.
Returns obspy.Stream
get_remote_extra_output_filenames()
def get_remote_extra_output_filenames(
    self,
) -> dict[tuple[str, ...], pathlib.Path]: ...

Get a dictionary of the remote paths of extra output files.

These files might no longer exist and might already have been transferred to the local site or deleted. These are the paths of the extra output files of the solver.

Returns dict[tuple[str, ...], pathlib.Path]
get_time_axis_from_meta_json()
def get_time_axis_from_meta_json(
    self,
    output_type: typing.Literal[
        "simulation", "point_data", "surface_data", "volume_data"
    ] = "simulation",
) -> TimeAxis: ...

Return the start time, end time, sampling rate and number of samples by parsing the meta.json file.

Naturally only works for synthetic data.

Parameters
  • output_type typing.Literal['simulation', 'point_data', 'surface_data', 'volume_data'] — Optionally define for which output type the time axis is constructed. The output_type “simulation” will always output the time axis as used by SalvusCompute. The other variations will adhere to the sampling rates of the output data for the relevant type.
Returns TimeAxis
get_wavefield_output()
def get_wavefield_output(
    self,
    output_type: typing.Literal["volume", "surface"],
    field: str,
    time_steps: slice = slice(None, None, None),
    download_if_not_exists: bool = False,
    delete_remote_data_after_download: bool = False,
) -> salvus.toolbox.helpers.wavefield_output.WavefieldOutput: ...

Get a surface or volume wavefield output object for the given simulation.

This will read the full volume or surface output to memory so please use this with caution.

Parameters
  • output_type typing.Literal['volume', 'surface'] — The output type to get, "surface" or "volume".
  • field str — The field to get.
  • time_steps slice — The time step indices to read expressed as a Python slice object.
  • download_if_not_exists bool — The data might not yet exist locally and only on a remote Salvus site. If that is the case, try to automatically download it. It will be stored in the same folder as the meta.json file for the simulation.
  • delete_remote_data_after_download bool — Delete the remote data and associated Salvus job after the download has succeeded. If set, this will delete all remote files for the current event, even if only a subset has been downloaded.
Returns salvus.toolbox.helpers.wavefield_output.WavefieldOutput
get_waveform_data()
def get_waveform_data(
    self,
    receiver_name: str,
    receiver_field: str,
    header: typing.Any | None = None,
    enforce_regular_time_grid: bool = True,
    time_axis: TimeAxis | None = None,
    _ignore_temporal_weights: bool = False,
    _use_cache: bool = False,
) -> obspy.Stream: ...

Get the waveform data for a given receiver and field in a consistent manner. The receiver name can also correspond to a receiver channel.

This is the central waveform retrieval function in SalvusProject and thus will return fully usable waveforms - this means fully filtered and selected/windowed waveforms, if so requested.

Might contain multiple traces per component, but only for data with a data selection configuration.

The order of the traces is always in the order of the expected output components.

It uses ObsPy’s Stream/Trace classes as they are just very suitable containers for waveform data.

Parameters
  • receiver_name str — The name of the receiver or receiver channel
  • receiver_field str — The field to get.
  • header typing.Any | None — Manually set the tr.stats.header to that value. Sometimes useful to have access to extra meta data in the processing and data selection functions.
  • enforce_regular_time_grid bool — Whether to check if a synthetic dataset that is subsampled during the simulation has an appropriate time axis. By default, SalvusCompute will always output the last time step of a simulation, even when the subsampling chosen wouldn’t match that point. Setting this parameter to True will check the meta.json specifications and ensure that the final point is removed again if it doesn’t match the regular grid of the time axis.
  • time_axis TimeAxis | None — Optionally specify a time axis for the data. Can be useful when dealing with heterogeneous data sources, such as seismological data, where the time axis may not be uniform across different stations.
  • _ignore_temporal_weights bool — Ignore temporal weights, even if they are set. Useful mainly for plotting and debugging purposes. Don’t set this if you are not aware of the consequences.
  • _use_cache bool — Utilize a cache for this function. Useful for certain use cases with repeated per-station access.
Returns obspy.Stream
get_waveform_data_xarray()
def get_waveform_data_xarray(
    self,
    receiver_field: str,
    enforce_regular_time_grid: bool = True,
    include_receiver_channel_inputs: bool = False,
    receiver_ids: list[str] | None = None,
    time_axis: TimeAxis | None = None,
    _ignore_temporal_weights: bool = False,
    _header_template: xr.DataArray | None = None,
    _use_cache: bool = False,
    _extra_output: ExtraOutput | None = None,
    _extra_output_force_is_observed: bool | None = None,
) -> xr.DataArray: ...

Get the waveform data for a given field for all receivers or receiver channels contained in the Event.

If the Event contains receiver channels. Only the data of the receiver channels will be returned and not of the individual receivers that form the channel.

Parameters
  • receiver_field str — The field to get.
  • enforce_regular_time_grid bool — Whether to check if a synthetic dataset that is subsampled during the simulation has an appropriate time axis. By default, SalvusCompute will always output the last time step of a simulation, even when the subsampling chosen wouldn’t match that point. Setting this parameter to True will check the meta.json specifications and ensure that the final point is removed again if it doesn’t match the regular grid of the time axis.
  • include_receiver_channel_inputs bool — Include the input receivers for receiver channels in the returned data array. By default, only the final receiver channel data will be included.
  • receiver_ids list[str] | None — Optionally specify a list of receiver IDs to filter the data by. If this is set, only the data for the specified receivers will be returned. If this is None, all receivers will be returned.
  • time_axis TimeAxis | None — An optional time axis to resample the data to. This will overwrite any potentially set temporal interpolation settings of the event data object.
  • _ignore_temporal_weights bool — Ignore temporal weights, even if they are set. Useful mainly for plotting and debugging purposes. Don’t set this if you are not aware of the consequences.
  • _header_template xr.DataArray | None — A data set to use as a template for the header.
  • _use_cache bool — Utilize a cache for this function. Useful for certain use cases with repeated per-station access.
  • _extra_output ExtraOutput | None — Optionally write intermediate data to disk.
  • _extra_output_force_is_observed bool | None — Force the waveform data to be considered observed data for the purpose of the extra output. Salvus considers all data that has been computed with it as synthetic but for the synthetic inversions the “observed” data should be written as observed extra output.
Returns xr.DataArray
has_receiver()
def has_receiver(self, receiver_name: str) -> bool: ...

Figure out if the given receiver is available.

Parameters
  • receiver_name str — Name of the receiver.
Returns bool
has_receiver_nsl()
def has_receiver_nsl(
    self, network_code: str, station_code: str, location_code: str
) -> bool: ...

Figure out if the given receiver is available.

Parameters
  • network_code str — The network code.
  • station_code str — The station code.
  • location_code str — The location code.
Returns bool
plot()
def plot(
    self,
    plot_types: typing.Iterable[str] = ("wiggles",),
    receiver_field: str,
    component: str,
    sort_by: typing.Callable | None = None,
) -> None: ...

Plot the waveforms.

Parameters
  • plot_types typing.Iterable[str] — How to plot the data.
  • receiver_field str — The receiver field to plot.
  • component str — The component to plot.
  • sort_by typing.Callable | None — Callback function called with the receiver objects to determine the sorting. Works like the key callback for the standard Python sorted() method.
Returns None
register_processing_function()
def register_processing_function(
    self,
    function: typing.Callable | None,
    chain_to_existing_function: bool = False,
) -> None: ...

Register a function to process data on the fly.

The function must take the following arguments if it is supposed to operate on structured data:

  • da: An xarray.DataArray object with the data.
  • sources: The source objects of this event

If the function should operate on obspy.Stream objects per receiver, the function must take the following arguments:

  • st: An obspy.Stream object with the data.
  • receiver: The receiver object for the receiver.
  • sources: All sources for the event.

In the case of ASDF data it can also optionally take:

  • inv: An obspy.Inventory object for the station in case it is stored in an ASDF file. Might be None.

The function must return the processed xarray.DataArray or obspy.Stream. Both must contain exactly the same number of traces, components and time steps as the input data.

Parameters
  • function typing.Callable | None — Function to process data on the fly. Or None to unset it.
  • chain_to_existing_function bool — If True the new function will be chained to a possibly already pre-existing processing function. If this is False and there is an already pre-existing function it will raise.
Returns None
register_temporal_weights_function()
def register_temporal_weights_function(
    self, function: typing.Callable | None, extra_kwargs: dict | None = None
) -> None: ...

Register a function to weight data on the fly.

The function must take two arguments if operating on structured xarray.DataArray objects:

  • da: An xarray.DataArray object with the data.
  • sources: All sources for the event.

If the function should operate on obspy.Stream objects per receiver, the function must take the following arguments:

  • st: An obspy.Stream object with the data.
  • receiver: The receiver object for the receiver.
  • sources: All sources for the event.

It furthermore must return a dictionary with a list of dictionaries for each component, e.g.

def compute_temporal_weights(
    da: xr.DataArray,
    sources: list[simple_config.source._Base],
) -> dict[str, list[dict]]:
    ...
    # Omit components which you want to ignore for the adjoint source
    # computation.
    weights = {
        "X": [
            {"values": np.array([...]), "misfit_weight": 0.5},
            {"values": np.array([...]), "misfit_weight": 0.5},
        ],
        "Y": [
            # If no "misfit_weight" is given, it defaults to 1.0.
            {"values": np.array([...])},
        ]}
    return weights
Parameters
  • function typing.Callable | None — Function to compute windows on the fly. Or None to unset it.
  • extra_kwargs dict | None — Extra keyword arguments to pass to the function.
Returns None
set_temporal_interpolation()
def set_temporal_interpolation(
    self,
    start_time_in_seconds: float,
    sampling_rate_in_hertz: float,
    npts: int,
    lanczos_window_width: int = 12,
    taper_in_samples: int = 5,
    reference_time: float | None = None,
) -> None: ...

Set the desired temporal interpolation. Once this is set, all returned waveform data will be resampled in time using a sinc-based Lanczos interpolator.

The interpolation will taper and pad with zeros if necessary. It will raise if it would result in more than 50 % of padded new samples.

Parameters
  • start_time_in_seconds float — The new start time in seconds. Can be interpreted as a UNIX timestamp if so desired.
  • sampling_rate_in_hertz float — The new sampling rate in Hertz.
  • npts int — The new number of samples.
  • lanczos_window_width int — The width in samples on either side to use when interpolating onto a new sample.
  • taper_in_samples int — In case padding with zeros is necessary each traces will be tapered to zero using this many samples at each end it is being padded at. Necessary to avoid introducing unresolved frequencies.
  • reference_time float | None — Reference time as a UTC timestamp in seconds. Needs to be set for seismological data.
Returns None
shallow_copy()
def shallow_copy(self) -> EventData: ...

Create a shallow copy of the object - things like data and sources and receivers are just references. Useful to have multiple versions of an event data object.

Returns EventData
unset_temporal_interpolation()
def unset_temporal_interpolation(self) -> None: ...

Unsets any temporal interpolation and data will from now on be returned with the original sampling.

Returns None

PlotProperties

class PlotProperties(builtins.object):
    def __init__(
        self, width: float = 11.5, height: float = 6.33, dpi: float = 100
    ) -> None: ...

Properties to use of an internally generated plot. This dataclass configures a Matplotlib axes object used in the internal plotting routine. It allows you to customize the dimensions and DPI of the resulting plot.

Parameters
  • width float — The width of the figure in inches. This parameter determines the horizontal size of the plot.
  • height float — The height of the figure in inches. This parameter determines the vertical size of the plot.
  • dpi float — Dots per inch (DPI) for the figure. DPI is a measure of the resolution or quality of the plot. Higher DPI values result in smoother and more detailed output.

Used in tutorials