pysmo.classes

@define(kw_only=True)
class GeoCsvSeismogram(SeismogramEndtimeMixin):
    """Import class for seismograms in the GeoCSV timeseries format.

    Reads a waveform from the timeseries flavour of
    [GeoCSV](https://ds.iris.edu/files/documents/GeoCSV.pdf) and exposes
    it as a [`Seismogram`][pysmo.Seismogram]-compatible object.

    This class is intended as a data-ingestion step. Once loaded, use
    [`clone_to_mini`][pysmo.functions.clone_to_mini] to convert the
    waveform to a [`MiniSeismogram`][pysmo.MiniSeismogram], which can then
    be passed to [`copy_from_mini`][pysmo.functions.copy_from_mini] to
    populate another object such as a [`SAC`][pysmo.classes.SAC] instance.
    Instances can be modified in memory but there is no write-back to
    GeoCSV.

    Examples:
        ```python
        >>> from pysmo import Seismogram
        >>> from pysmo.classes import GeoCsvSeismogram
        >>> text = '''\\
        ... # dataset: GeoCSV 2.0
        ... # delimiter: ,
        ... # field_unit: UTC, Counts
        ... # field_type: datetime, INTEGER
        ... # SID: IU_ANMO_00_LHZ
        ... # sample_count: 3
        ... # sample_rate_hz: 1.0
        ... # start_time: 2010-02-27T06:30:00Z
        ... Time, Sample
        ... 2010-02-27T06:30:00Z, -47297
        ... 2010-02-27T06:30:01Z, -47298
        ... 2010-02-27T06:30:02Z, -47299'''
        >>> seismogram = GeoCsvSeismogram.from_text(text)
        >>> isinstance(seismogram, Seismogram)
        True
        >>> seismogram.sid
        'IU_ANMO_00_LHZ'
        >>> seismogram.data
        array([-47297., -47298., -47299.])
        >>> seismogram.end_time
        Timestamp('2010-02-27 06:30:02+0000', tz='UTC')
        >>>
        ```
    """

    begin_time: UtcTimestamp = field(
        converter=convert_to_utc_timestamp,
        on_setattr=setters.convert,
    )
    """Seismogram begin time."""

    delta: PositiveTimedelta = field(
        converter=convert_to_timedelta,
        validator=[
            validators.instance_of(pd.Timedelta),
            validators.gt(pd.Timedelta(0)),
        ],
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Seismogram sampling interval."""

    data: np.ndarray = field(
        converter=convert_to_ndarray,
        validator=validators.instance_of(np.ndarray),
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Seismogram data."""

    sid: str = field(
        validator=validators.instance_of(str),
        on_setattr=setters.validate,
    )
    """FDSN source identifier of the parsed GeoCSV data (e.g. `IU_ANMO_00_LHZ`).

    This is parse-time metadata: it describes the GeoCSV data the instance
    was created from and is not updated when other attributes change.
    """

    @property
    def sample_count(self) -> int:
        """Number of samples, always equal to `len(data)`."""
        return len(self.data)

    @classmethod
    def from_text(cls, text: str) -> Self:
        """Create a new instance from a GeoCSV text body.

        The text may contain several timeseries datasets (the EarthScope
        dataselect service returns one dataset per contiguous segment);
        they are merged into a single continuous waveform.

        Args:
            text: GeoCSV text containing one or more timeseries datasets.

        Returns:
            A new GeoCsvSeismogram instance.

        Raises:
            ValueError: If the text contains no GeoCSV datasets, a dataset
                is not a valid timeseries, or the datasets cannot be merged
                into a continuous waveform (data gaps, differing channels
                or sample rates).
        """
        datasets = parse_geocsv(text)
        if not datasets:
            raise ValueError("No GeoCSV datasets found in text.")
        segment = merge_geocsv_timeseries(
            [extract_geocsv_timeseries(dataset) for dataset in datasets]
        )
        return cls(
            begin_time=segment.start_time,
            delta=pd.Timedelta(seconds=1.0 / segment.sample_rate_hz),
            data=segment.data,
            sid=segment.sid,
        )

    @classmethod
    def fetch(
        cls, *, station: Station, starttime: pd.Timestamp, endtime: pd.Timestamp
    ) -> Self:
        """Fetch and parse a seismogram from the EarthScope FDSN dataselect
        web service, for an absolute time window.

        For a window relative to a predicted phase arrival instead, compute
        the window yourself (e.g. with [`pysmo.tools.web.fetch_travel_times`][],
        which shows exactly this in its own Examples) and pass the
        resulting *starttime*/*endtime* here.

        Args:
            station: Any object satisfying the [`Station`][pysmo.Station]
                protocol. Provides the network, station code, location, and
                channel for the request.
            starttime: Start of the requested time window (UTC).
            endtime: End of the requested time window (UTC).

        Returns:
            A new GeoCsvSeismogram instance.

        Raises:
            ValueError: If no waveform data is returned for the given
                window, or the returned segments cannot be merged into a
                continuous trace (data gaps, differing channels or sample
                rates).
            urllib3.exceptions.ResponseError: If the dataselect web service
                returns an HTTP error.

        Examples:
            ```python
            >>> import pandas as pd
            >>> from pysmo import MiniStation
            >>> from pysmo.classes import GeoCsvSeismogram
            >>> station = MiniStation(
            ...     name="ANMO", network="IU", location="00", channel="LHZ",
            ...     latitude=34.945981, longitude=-106.457133,
            ... )
            >>> seismogram = GeoCsvSeismogram.fetch(
            ...     station=station,
            ...     starttime=pd.Timestamp("2010-02-27T06:44:00Z"),
            ...     endtime=pd.Timestamp("2010-02-27T06:54:00Z"),
            ... )  # doctest: +SKIP
            >>>
            ```
        """
        starttime = convert_to_utc_timestamp(starttime)
        endtime = convert_to_utc_timestamp(endtime)
        waveform_bytes = fetch_geocsvseismogram(
            station=station, starttime=starttime, endtime=endtime
        )
        if not waveform_bytes.strip():
            raise ValueError(
                f"No waveform data returned for "
                f"{station.network}.{station.name}.{station.location}."
                f"{station.channel} between {starttime} and {endtime}."
            )
        return cls.from_text(waveform_bytes.decode("utf-8"))

@define(kw_only=True)
class SAC(SacIO):
    """Access and modify data stored in SAC files.

    The [`SAC`][pysmo.classes.SAC] class inherits all attributes and methods
    of the [`SacIO`][pysmo.lib.io.SacIO] class, and extends it with attributes
    that allow using pysmo types. The extra attributes are themselves instances
    of "helper" classes that shouldn't be instantiated directly.

    Examples:
        SAC instances are typically created by reading a SAC file. Users
        familiar with the SAC file format can access header and data using
        the names they are used to:

        ```python
        >>> from pysmo.classes import SAC
        >>> sac = SAC.from_file("example.sac")
        >>> sac.delta
        0.019999999552965164
        >>> sac.data
        array([2302., 2313., 2345., ..., 2836., 2772., 2723.], shape=(180000,))
        >>> sac.evla
        -31.465999603271484
        >>>
        ```

        Presenting the data in the above way is *not* compatible with pysmo
        types. For example, event coordinates are stored in the
        [`evla`][pysmo.lib.io.SacIO.evla] and [`evlo`][pysmo.lib.io.SacIO.evlo]
        attributes, which do not match the pysmo [`Location`][pysmo.Location]
        type. Renaming or aliasing `evla` to `latitude` and `evlo` to
        `longitude` would solve the problem for the event coordinates, but
        since the SAC format also specifies station coordinates
        ([`stla`][pysmo.lib.io.SacIO.stla], [`stlo`][pysmo.lib.io.SacIO.stlo])
        we still run into compatibility issues.

        In order to map these incompatible attributes to ones that can be
        used with pysmo types, we use helper classes as a way to access the
        attributes under different names that *are* compatible with pysmo
        types:

        ```python
        >>> # Import the Seismogram type to check if the nested class is compatible:
        >>> from pysmo import Seismogram
        >>>
        >>> # First verify that a SAC instance is not a pysmo Seismogram:
        >>> isinstance(sac, Seismogram)
        False
        >>> # The sac.seismogram object is, however:
        >>> isinstance(sac.seismogram, Seismogram)
        True
        >>>
        ```

        Because the SAC file format defines a large amount of header fields for
        metadata, it needs to allow for many of these to be optional. Since the
        helper classes are more specific (and intended to be used with pysmo
        types), their attributes typically may *not* be [`None`][]:

        ```python
        >>> # No error: a SAC file doesn't have to contain event information:
        >>> sac.evla = None
        >>>
        ```

    Tip:
        The [`SAC`][pysmo.classes.SAC] class directly inherits from the
        [`SacIO`][pysmo.lib.io.SacIO] class. This gives access to all
        SAC headers, ability to load from a file, download data, and so on.
        Using [`SAC`][pysmo.classes.SAC] is therefore almost always
        preferred over using [`SacIO`][pysmo.lib.io.SacIO].
    """

    seismogram: SacSeismogram = field(init=False)
    """Access data stored in the SAC object compatible with the [`Seismogram`][pysmo.Seismogram] type."""

    station: SacStation = field(init=False)
    """Access data stored in the SAC object compatible with the [`Station`][pysmo.Station] type."""

    event: SacEvent = field(init=False)
    """Access data stored in the SAC object compatible with the [`Event`][pysmo.Event] type."""

    timestamps: SacTimestamps = field(init=False)
    """Maps a SAC times such as B, E, O, T0-T9 to Timestamp objects."""

    def __attrs_post_init__(self) -> None:
        self.seismogram = SacSeismogram(parent=self)
        self.station = SacStation(parent=self)
        self.event = SacEvent(parent=self)
        self.timestamps = SacTimestamps(parent=self)

@define(kw_only=True)
class SacEvent(_SacNested):
    """Helper class for SAC event attributes.

    The `SacEvent` class is used to map SAC attributes in a way that
    matches pysmo types. An instance of this class is created for each
    new (parent) [`SAC`][pysmo.classes.SAC] instance to enable pysmo
    types compatibility.

    Examples:
        Checking if a SacEvent matches the pysmo
        [`Event`][pysmo.Event] type:

        ```python
        >>> from pysmo.classes import SAC
        >>> from pysmo import Event
        >>> sac = SAC.from_file("example.sac")
        >>> isinstance(sac.event, Event)
        True
        >>>
        ```

    Note:
        Not all SAC files contain event information.
    """

    @property
    def latitude(self) -> int | float:
        """Event Latitude."""

        if self._parent.evla is None:
            raise TypeError("SAC object event latitude 'evla' is None.")
        return self._parent.evla

    @latitude.setter
    def latitude(self, value: int | float) -> None:
        setattr(self._parent, "evla", value)

    @property
    def longitude(self) -> int | float:
        """Event Longitude."""

        if self._parent.evlo is None:
            raise TypeError("SAC object event longitude 'evlo' is None.")
        return self._parent.evlo

    @longitude.setter
    def longitude(self, value: int | float) -> None:
        setattr(self._parent, "evlo", value)

    @property
    def depth(self) -> int | float:
        """Event depth in metres (positive downward from the surface)."""

        if self._parent.evdp is None:
            raise TypeError("Sac object event depth 'evdp' is None.")
        return self._parent.evdp * 1000

    @depth.setter
    def depth(self, value: int | float) -> None:
        setattr(self._parent, "evdp", value / 1000)

    @property
    def time(self) -> UtcTimestamp:
        """Event origin time (UTC).

        Important:
            This property uses the [`SacIO.o`][pysmo.lib.io.SacIO.o] time
            header. If [`SacIO.iztype`][pysmo.lib.io.SacIO.iztype] is set to
            `"o"`, then this is also the "Reference time equivalance" and
            [`SacIO.o`][pysmo.lib.io.SacIO.o] cannot be changed (it is always
            0). Changing the [`time`][pysmo.classes.SacEvent.time] directly
            is not possible if this is the case.
        """

        event_time = self._get_timestamp_from_sac(SAC_OPTIONAL_TIME_HEADERS.o)
        if event_time is None:
            raise TypeError("SAC object event time 'o' is None.")
        return event_time

    @time.setter
    def time(self, value: pd.Timestamp) -> None:
        self._set_sac_from_timestamp(SAC_OPTIONAL_TIME_HEADERS.o, value)

@define(kw_only=True, slots=True)
class SacPZ:
    """Import class for SAC PZ (pole-zero) files.

    Reads an analog instrument response from a
    [SAC PZ](https://ds.iris.edu/files/sac-manual/commands/transfer.html)
    file (as produced by e.g. the EarthScope SACPZ web service) and exposes
    it as a [`Response`][pysmo.Response]-compatible object. `SacPZ` only ever
    satisfies [`Response`][pysmo.Response], never
    [`StagedResponse`][pysmo.StagedResponse] — the SAC PZ format has no
    digital-stage fields to parse.

    Examples:
        ```python
        >>> from pysmo import Response
        >>> from pysmo.classes import SacPZ
        >>> text = '''\\
        ... * NETWORK   (KNETWK): IU
        ... * STATION    (KSTNM): ANMO
        ... * LOCATION   (KHOLE): 00
        ... * CHANNEL   (KCMPNM): BHZ
        ... * START             : 2018-07-09T20:45:00
        ... * END               :
        ... * INPUT UNIT        : M
        ... ZEROS 2
        ... \\t+0.000000e+00\\t+0.000000e+00
        ... \\t+0.000000e+00\\t+0.000000e+00
        ... POLES 1
        ... \\t-1.000000e-02\\t+0.000000e+00
        ... CONSTANT 1.0e+09
        ... '''
        >>> response = SacPZ.from_text(text)
        >>> isinstance(response, Response)
        True
        >>> response.network, response.station
        ('IU', 'ANMO')
        >>>
        ```
    """

    poles: list[complex] = field()
    """Response poles.

    See [`Response.poles`][pysmo.Response.poles] for more details.
    """

    zeros: list[complex] = field()
    """Response zeros.

    See [`Response.zeros`][pysmo.Response.zeros] for more details.
    """

    overall_sensitivity: NonZeroNumber = field(
        converter=float, validator=validate_nonzero
    )
    """Total system sensitivity (the SAC PZ file's `CONSTANT`).

    See [`Response.overall_sensitivity`][pysmo.Response.overall_sensitivity]
    for more details.
    """

    reference_sensitivity: NonZeroNumber | None = field(
        default=None,
        converter=_convert_optional_float,
        validator=validators.optional(validate_nonzero),
    )
    """Total system sensitivity at the reference frequency, `A0` excluded
    (the SAC PZ file's `SENSITIVITY` header, if present).

    See
    [`Response.reference_sensitivity`][pysmo.Response.reference_sensitivity]
    for more details.
    """

    input_units: str = field(validator=validators.instance_of(str))
    """Physical units produced by removing this response via full spectral
    deconvolution — not necessarily via the sensitivity-only path, see
    [`remove_response`][pysmo.tools.signal.remove_response] for why.

    See [`Response.input_units`][pysmo.Response.input_units] for more details.
    """

    network: str = field(validator=validators.instance_of(str))
    """Network code parsed from the SAC PZ file's comment header."""

    station: str = field(validator=validators.instance_of(str))
    """Station code parsed from the SAC PZ file's comment header."""

    location: str = field(validator=validators.instance_of(str))
    """Location code parsed from the SAC PZ file's comment header."""

    channel: str = field(validator=validators.instance_of(str))
    """Channel code parsed from the SAC PZ file's comment header."""

    start_date: pd.Timestamp = field()
    """Start of the epoch this response applies to."""

    end_date: pd.Timestamp | None = field(default=None)
    """End of the epoch this response applies to, or `None` if still open."""

    @classmethod
    def from_text(cls, text: str) -> Self:
        """Create a new instance from a single-record SAC PZ text body.

        Args:
            text: SAC PZ text containing exactly one record (the common
                sidecar-file case, e.g. one `.pz`/`SACPZ.NET.STA.LOC.CHA`
                file matched to one channel epoch by filename convention).

        Returns:
            A new SacPZ instance.

        Raises:
            ValueError: If the text contains zero or more than one SAC PZ
                record.

        Tip: See Also
            [`SacPZ.all_from_text`][pysmo.classes.SacPZ.all_from_text]: Parse
            a bulk/concatenated multi-record text body.

        Examples:
            Reading a SAC PZ file already saved to disk — the common case for
            archived/legacy data, e.g. extracted from an old SEED volume with
            `rdseed -p`, rather than fetched live from EarthScope:

            ```python
            >>> from pathlib import Path
            >>> from pysmo import Response
            >>> from pysmo.classes import SacPZ
            >>> text = Path("SACPZ.IU.ANMO.00.BHZ").read_text()
            >>> response = SacPZ.from_text(text)
            >>> isinstance(response, Response)
            True
            >>> response.network, response.station
            ('IU', 'ANMO')
            >>>
            ```
        """
        records = parse_sacpz(text)
        if len(records) != 1:
            raise ValueError(
                f"Expected exactly one SAC PZ record in text, found {len(records)}."
            )
        return cls._from_raw(records[0])

    @classmethod
    def fetch(cls, *, station: Station, time: pd.Timestamp | None = None) -> Self:
        """Fetch and parse an instrument response from the EarthScope SACPZ
        web service, selecting one epoch.

        Unlike [`StationXML.fetch`][pysmo.classes.StationXML.fetch], epoch
        selection happens server-side: the SACPZ web service's own `time`
        parameter is passed through, so exactly one record is returned
        (the epoch active at *time* if given, otherwise the one currently
        open) without needing to fetch the full response history first.

        Args:
            station: Any object satisfying the [`Station`][pysmo.Station]
                protocol. Provides the network, station code, location, and
                channel for the request.
            time: Timestamp used to select the response epoch. If `None`,
                the currently-open epoch is selected.

        Returns:
            A new SacPZ instance for the response epoch active at *time*
            (or currently open, if `time` is `None`).

        Raises:
            ValueError: If the web service's response does not contain
                exactly one SAC PZ record.
            urllib3.exceptions.ResponseError: If the web service returns an
                HTTP error.

        Tip:
            When fetching live from EarthScope rather than reading an
            existing SAC PZ file, prefer
            [`StationXML.fetch`][pysmo.classes.StationXML.fetch]: the
            StationXML response also captures digital FIR/IIR stages, so it
            always satisfies [`StagedResponse`][pysmo.StagedResponse], unlike
            `SacPZ`.

        Examples:
            ```python
            >>> from pysmo import MiniStation
            >>> from pysmo.classes import SacPZ
            >>> station = MiniStation(
            ...     name="ANMO", network="IU", location="00", channel="BHZ",
            ...     latitude=34.945981, longitude=-106.457133,
            ... )
            >>> response = SacPZ.fetch(station=station)  # doctest: +SKIP
            >>>
            ```
        """
        text = fetch_sacpz(station=station, time=time)
        return cls.from_text(text)

    @classmethod
    def all_from_text(cls, text: str) -> list[Self]:
        """Create one instance per record in a bulk/concatenated SAC PZ text body.

        Unlike [`from_text`][pysmo.classes.SacPZ.from_text], this does not
        require (or merge to) a single record — a SACPZ retrieval that is
        not pinned to a single channel epoch returns multiple concatenated
        records, each with its own `network`/`station`/`location`/`channel`/
        `start_date`/`end_date` provenance, which callers can filter
        themselves.

        Args:
            text: SAC PZ text containing one or more records.

        Returns:
            One SacPZ instance per record found, in order of appearance.
        """
        return [cls._from_raw(record) for record in parse_sacpz(text)]

    @classmethod
    def _from_raw(cls, record: _RawSacPzResponse) -> Self:
        return cls(
            poles=record.poles,
            zeros=record.zeros,
            overall_sensitivity=record.overall_sensitivity,
            reference_sensitivity=record.reference_sensitivity,
            input_units=record.input_units,
            network=record.network,
            station=record.station,
            location=record.location,
            channel=record.channel,
            start_date=record.start_date,
            end_date=record.end_date,
        )