pysmo.classes

@define(kw_only=True)
class GeoCsvSeismogram(SeismogramEndtimeMixin):
    r"""Import/export 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.
    Use [`write`][pysmo.classes.GeoCsvSeismogram.write] to serialise the instance back
    to a GeoCSV 2.0 file, or [`pysmo.lib.io.write_geocsv`][] to write one or more
    `Seismogram`-compatible objects in a single call.

    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')
        >>> import pathlib
        >>> seismogram.write("out.geocsv"); recovered = GeoCsvSeismogram.from_text(pathlib.Path("out.geocsv").read_text())
        >>> recovered.sid
        'IU_ANMO_00_LHZ'
        >>>
        ```
    """

    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"))

    def write(self, path: str | PathLike) -> None:
        r"""Write this seismogram to a GeoCSV 2.0 file.

        Serialises the instance as a single GeoCSV 2.0 timeseries dataset.
        To write several seismograms into one multi-dataset file use
        [`pysmo.lib.io.write_geocsv`][] directly.

        Args:
            path: Destination file path. The file is written in UTF-8 text
                mode and any existing content is overwritten.

        Examples:
            ```python
            >>> import pathlib
            >>> 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)
            >>> seismogram.write("out.geocsv"); recovered = GeoCsvSeismogram.from_text(
            ...     pathlib.Path("out.geocsv").read_text()
            ... )
            >>> recovered.sid == seismogram.sid
            True
            >>>
            ```
        """
        write_geocsv(self, path)

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

    [`SAC`][pysmo.classes.SAC] wraps a [`SacIO`][pysmo.lib.io.SacIO] instance
    and adds attributes alongside it that allow using pysmo types. The extra
    attributes are themselves instances of "helper" classes that should not
    be instantiated directly.

    Examples:
        SAC instances are typically created by reading a SAC file:

        ```python
        >>> from pysmo.classes import SAC
        >>> sac = SAC.from_file("example.sac")
        >>> sac.seismogram.delta
        Timedelta('0 days 00:00:00.050000000')
        >>> sac.seismogram.data
        array([-47201., -47361., -47511., ..., -82144., -71072., -59960.],
              shape=(57465,))
        >>>
        ```

        Raw SAC header values are *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]
        headers, 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]),
        the same compatibility issue remains.

        The [`SAC`][pysmo.classes.SAC] class solves this with helper classes
        that map these incompatible attributes to ones compatible with pysmo
        types, accessible under different names:

        ```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 number of header fields
        for metadata, many of them are 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.native.evla = None
        >>>
        ```

    Tip:
        [`SAC`][pysmo.classes.SAC] only exposes a small, curated surface
        directly (file I/O, and the pysmo-typed
        [`station`][pysmo.classes.SAC.station],
        [`event`][pysmo.classes.SAC.event],
        [`seismogram`][pysmo.classes.SAC.seismogram] and
        [`timestamps`][pysmo.classes.SAC.timestamps] helpers) rather than
        the full raw SAC header set. Seismogram data and sampling interval
        are available via [`seismogram`][pysmo.classes.SAC.seismogram].
        Users familiar with the SAC file format who want direct access to a
        header by its native name (e.g. `evla`, `stla`, `kstnm`) can reach
        the underlying [`SacIO`][pysmo.lib.io.SacIO] instance via
        [`SAC.native`][pysmo.classes.SAC.native].
    """

    native: SacIO = field(factory=SacIO, repr=False, on_setattr=setters.frozen)
    """The underlying [`SacIO`][pysmo.lib.io.SacIO] instance.

    This is the escape hatch for direct access to raw SAC headers by their
    native names (e.g. `SAC.native.evla`), for users familiar with the SAC file
    format who need it.

    Fixed for the lifetime of the instance: [`seismogram`][pysmo.classes.SAC.seismogram],
    [`station`][pysmo.classes.SAC.station], [`event`][pysmo.classes.SAC.event]
    and [`timestamps`][pysmo.classes.SAC.timestamps] are bound to this object
    at construction time, so reassigning it would silently orphan them. To
    load different data into an existing instance, use
    [`read`][pysmo.classes.SAC.read]/[`read_buffer`][pysmo.classes.SAC.read_buffer],
    which update this same object in place; otherwise construct a new
    [`SAC`][pysmo.classes.SAC] instance.
    """

    seismogram: SacSeismogram = field(init=False)
    """This SAC object exposed as a [`Seismogram`][pysmo.Seismogram]."""

    station: SacStation = field(init=False)
    """This SAC object exposed as a [`Station`][pysmo.Station]."""

    event: SacEvent = field(init=False)
    """This SAC object exposed as an [`Event`][pysmo.Event]."""

    timestamps: SacTimestamps = field(init=False)
    """Maps SAC time headers such as B, E, O, T0-T9 to
    [`Timestamp`][pandas.Timestamp] objects."""

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

    @classmethod
    def from_file(cls, filename: str | PathLike) -> Self:
        """Create a new SAC instance from a SAC file.

        Args:
            filename: Name of the SAC file to read.

        Returns:
            A new SAC instance.
        """
        return cls(native=SacIO.from_file(filename))

    @classmethod
    def from_buffer(cls, buffer: bytes) -> Self:
        """Create a new SAC instance from a SAC data buffer.

        Args:
            buffer: Buffer containing SAC file content.

        Returns:
            A new SAC instance.
        """
        return cls(native=SacIO.from_buffer(buffer))

    def read(self, filename: str | PathLike) -> None:
        """Read data and headers from a SAC file into an existing SAC instance.

        Args:
            filename: Name of the SAC file to read.
        """
        self.native.read(filename)

    def read_buffer(self, buffer: bytes) -> None:
        """Read data and headers from a SAC byte buffer into an existing SAC instance.

        Args:
            buffer: Buffer containing SAC file content.
        """
        self.native.read_buffer(buffer)

    def write(self, filename: str | PathLike) -> None:
        """Write data and header values to a SAC file.

        Args:
            filename: Name of the SAC file to write to.
        """
        self.native.write(filename)

    @classmethod
    def from_zip(cls, archive: bytes) -> Self:
        """Create a new instance from a zip archive containing exactly one continuous SAC segment.

        Args:
            archive: Raw zip archive bytes containing exactly one SAC file
                (as returned by the FDSN dataselect web service with
                `format=sac.zip`).

        Returns:
            A new SAC instance.

        Raises:
            ValueError: If `archive` is not a valid zip archive, contains no
                members, contains more than one member (e.g. due to a data
                gap, an instrument/metadata epoch change, overlapping
                records, or a wildcarded channel/location code matching
                more than one channel), or a member cannot be parsed as a
                SAC file.

        Tip: See Also
            [`SAC.all_from_zip`][pysmo.classes.SAC.all_from_zip]: Parse
            every segment in the archive without requiring exactly one.
        """
        segments = cls.all_from_zip(archive)
        if len(segments) == 1:
            return segments[0]
        if not segments:
            raise ValueError("Zip archive contains no SAC segments.")

        segment_lines = "\n".join(
            f"  {segment.station.network}.{segment.station.name}."
            f"{segment.station.location}.{segment.station.channel}  "
            f"{segment.seismogram.begin_time} -- {segment.seismogram.end_time}"
            for segment in segments
        )
        raise ValueError(
            f"Zip archive contains {len(segments)} segments; "
            f"SAC.from_zip() requires exactly one continuous segment. "
            f"Use SAC.all_from_zip() to get all segments instead. "
            f"Segments found:\n{segment_lines}"
        )

    @classmethod
    def fetch(
        cls, *, station: Station, starttime: pd.Timestamp, endtime: pd.Timestamp
    ) -> Self:
        """Fetch and parse a SAC 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.

        To fetch once and interpret later (e.g. offline, or without
        repeating the network request), use
        [`pysmo.tools.web.fetch_sac`][] and
        [`from_zip`][pysmo.classes.SAC.from_zip] /
        [`all_from_zip`][pysmo.classes.SAC.all_from_zip] directly instead.

        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 SAC instance.

        Raises:
            ValueError: If no waveform data is returned for the given
                window, if more than one continuous segment is returned
                (e.g. due to a data gap, an instrument/metadata epoch
                change, overlapping records, or a wildcarded channel/
                location code matching more than one channel), or if a
                returned segment cannot be parsed as a SAC file.
            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 SAC
            >>> station = MiniStation(
            ...     name="ANMO", network="IU", location="00", channel="LHZ",
            ...     latitude=34.945981, longitude=-106.457133,
            ... )
            >>> sac = SAC.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)
        archive = fetch_sac(station=station, starttime=starttime, endtime=endtime)

        # dataselect returns an empty (zero-length) body, not a
        # zero-member zip archive, when a request is well-formed but
        # matches no data (HTTP 204, the FDSN default `nodata` handling) —
        # confirmed live.
        if not archive:
            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_zip(archive)

    @classmethod
    def all_from_zip(cls, archive: bytes) -> list[Self]:
        """Create one instance per SAC file in a zip archive.

        Unlike [`from_zip`][pysmo.classes.SAC.from_zip], this does not
        require exactly one segment — a response covering a data gap, an
        instrument/metadata epoch change, or a wildcarded channel/location
        code returns several, which callers can inspect or merge
        themselves.

        Args:
            archive: Raw zip archive bytes, as returned by the FDSN
                dataselect web service with `format=sac.zip`.

        Returns:
            One SAC instance per member of the archive, in archive order.
            Empty if the archive has no members.

        Raises:
            ValueError: If `archive` is not a valid zip archive, or a
                member cannot be parsed as a SAC file.
        """
        try:
            with ZipFile(BytesIO(archive)) as archive_zip:
                names = archive_zip.namelist()
                segments = []
                for name in names:
                    try:
                        segments.append(cls.from_buffer(archive_zip.read(name)))
                    except Exception as error:
                        raise ValueError(
                            f"Could not parse segment {name!r} in zip archive: {error}"
                        ) from error
        except BadZipFile as error:
            raise ValueError(f"Not a valid zip archive: {error}") from error
        return segments