salvus.flow.collections.event_data

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

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

Is the data from a Salvus simulation?

Contents of the job_info.json file.

Contents of the meta.json file.

The receiver channels for this event.

Return a list of receiver names for this event.

The receivers for this event.

The sources for this event.

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

Construct an EventData object from a folder containing Salvus outputs.

def _get_data_cube(
    self,
    receiver_field: str,
    component: typing.Union[str, int],
    sort_by: typing.Optional[typing.Callable[[_Base], int]] = None,
    filter_by: typing.Optional[typing.Callable[[_Base], bool]] = None,
    select_by: typing.Optional[
        typing.Callable[
            [typing.List[typing.Tuple[_Base, obspy.Stream]], EventData],
            typing.List[typing.Tuple[_Base, obspy.Stream]],
        ]
    ] = None,
) -> typing.Tuple[np.ndarray, np.ndarray]:
    ...

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.

def compute_data_hash(self) -> str:
    ...

Compute a unique hash for the event data.

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

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

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

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

def expected_components_from_receiver(
    receiver: Union[salvus.flow.simple_config.receiver._Base, Dict],
    receiver_field: str,
) -> List[str]:
    ...

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

def get_associated_salvus_job(self) -> "SalvusJob":
    ...

Get the associated Salvus job, if it still exists.

def get_data_cube(
    self,
    receiver_field: str,
    component: typing.Union[str, int],
    sort_by: typing.Optional[typing.Callable[[_Base], int]] = None,
    filter_by: typing.Optional[typing.Callable[[_Base], bool]] = None,
    select_by: typing.Optional[
        typing.Callable[
            [typing.List[typing.Tuple[_Base, obspy.Stream]], EventData],
            typing.List[typing.Tuple[_Base, obspy.Stream]],
        ]
    ] = None,
) -> typing.Tuple[np.ndarray, np.ndarray]:
    ...

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.

def get_receiver(
    self, receiver_name: str
) -> salvus.flow.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.

def get_receiver_data(
    self,
    receiver_name: str,
    receiver_field: str,
    header: typing.Optional[typing.Any] = None,
    _ignore_temporal_weights: bool = False,
) -> obspy.Stream:
    ...

Get the waveform data for a given receiver and field in a consistent manner.

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.

def get_remote_extra_output_filenames(
    self,
) -> typing.Dict[typing.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.

def get_time_axis_from_meta_json(self) -> 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.

def get_wavefield_output(
    self,
    output_type: typing_extensions.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.

def get_waveform_data(
    self,
    receiver_name: str,
    receiver_field: str,
    header: typing.Optional[typing.Any] = None,
    _ignore_temporal_weights: 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.

def get_waveform_data_for_receiver_channel(
    self,
    identifier: str,
    receiver_field: str,
    header: typing.Optional[typing.Any] = None,
    _ignore_temporal_weights: bool = False,
) -> obspy.Stream:
    ...

Get waveform data for a specific receiver channel.

Automatically sums up the signals of the individual channels and optionally time-shifts receivers before summation.

def has_receiver(self, receiver_name: str) -> bool:
    ...

Figure out if the given receiver is available.

def has_receiver_nsl(
    self, network_code: str, station_code: str, location_code: str
) -> bool:
    ...

Figure out if the given receiver is available.

def plot(
    self,
    plot_types: typing.Iterable[str] = ("wiggles",),
    receiver_field: str,
    component: str,
    sort_by: typing.Optional[typing.Callable] = None,
) -> None:
    ...

Plot the waveforms.

def register_processing_function(
    self,
    function: typing.Optional[typing.Callable],
    chain_to_existing_function: bool = False,
) -> None:
    ...

Register a function to process data on the fly.

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 obspy.Stream object which must contain exactly the same number of traces.

def register_temporal_weights_function(
    self, function: typing.Optional[typing.Callable]
) -> None:
    ...

Register a function to weight data on the fly.

The function must take three 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(
    st: obspy.Stream,
    receiver: simple_config.receiver._Base,
    sources: typing.List[simple_config.source._Base],
) -> typing.Dict[str, typing.List[typing.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

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,
) -> 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.

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.

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

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