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.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.sourceid
        '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.sourceid
        '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: npt.NDArray[np.floating] = field(
        converter=convert_to_ndarray,
        validator=validators.instance_of(np.ndarray),
        on_setattr=setters.pipe(setters.convert, setters.validate),
        eq=cmp_using(eq=np.array_equal),
    )
    """Seismogram data."""

    sourceid: str = field(
        validator=validators.instance_of(str),
        on_setattr=setters.validate,
    )
    """FDSN Source Identifier as carried in the GeoCSV `SID` header.

    Stored verbatim as parsed, e.g. `IU_ANMO_00_LHZ` — no `FDSN:` URN
    prefix, and the channel is not split into band/source/subsource. This
    differs from [`MSeed.sourceid`][pysmo.classes.MSeed.sourceid], which
    keeps the full URN form. 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,
            # integer ns keeps sub-microsecond precision the float form loses
            delta=pd.Timedelta(round(1_000_000_000 / segment.sample_rate_hz), "ns"),
            data=segment.data,
            sourceid=segment.sourceid,
        )

    @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.traveltime.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:
            <!-- skip: start if(not run_real_web_requests) -->
            ```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"),
            ... )
            >>>
            ```
            <!-- skip: end -->
        """
        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(
                "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[str]) -> 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.sourceid == seismogram.sourceid
            True
            >>>
            ```
        """
        write_geocsv(self, path)

@define(kw_only=True)
class MSeed(SeismogramEndtimeMixin):
    """Import/export class for one contiguous miniSEED trace segment.

    Wraps EarthScope's `pymseed` and exposes a single regularly-sampled
    segment as a [`Seismogram`][pysmo.Seismogram]-compatible object. The
    `pymseed` trace-list hierarchy is flattened at read time: each
    contiguous segment becomes one `MSeed`.

    miniSEED carries no station coordinates and no event data — only
    channel identity, timing and samples. `MSeed` exposes the network,
    station, location and channel codes as read-only properties derived
    from `sourceid` (an `MSeed` is a [`StationCode`][pysmo.StationCode] at
    runtime), but not [`Station`][pysmo.Station]. `sourceid` is the single
    authoritative identity value; to relabel, set it directly. For a
    `Station`, build a [`MiniStation`][pysmo.MiniStation] (or fetch a
    [`StationXML`][pysmo.classes.StationXML]) separately and combine.

    Use [`from_file`][pysmo.classes.MSeed.from_file] /
    [`from_bytes`][pysmo.classes.MSeed.from_bytes] to read a single
    segment, their `all_*` counterparts to read every segment, and
    [`fetch`][pysmo.classes.MSeed.fetch] to read directly from the
    EarthScope dataselect web service. Use
    [`write`][pysmo.classes.MSeed.write] to serialise back to a miniSEED
    file, or [`pysmo.lib.io.write_mseed`][] to write several segments in a
    single call.

    Examples:
        ```python
        >>> from pysmo.classes import MSeed
        >>> seismogram = MSeed.from_file("example.mseed")
        >>> seismogram.sourceid
        'FDSN:IU_ANMO_00_B_H_Z'
        >>> seismogram.network, seismogram.name, seismogram.location, seismogram.channel
        ('IU', 'ANMO', '00', 'BHZ')
        >>>
        ```
    """

    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: npt.NDArray[np.floating] = field(
        converter=_convert_mseed_data,
        validator=validators.instance_of(np.ndarray),
        on_setattr=setters.pipe(setters.convert, setters.validate),
        eq=cmp_using(eq=np.array_equal),
    )
    """Seismogram data, always `float64`."""

    sourceid: str = field(
        validator=validators.instance_of(str),
        on_setattr=setters.validate,
    )
    """FDSN Source Identifier this segment was read from.

    The full URN form as carried in miniSEED and returned by `pymseed`,
    e.g. `FDSN:IU_ANMO_00_B_H_Z` — the `FDSN:` prefix is kept and the
    channel is split into band/source/subsource. This differs from
    [`GeoCsvSeismogram.sourceid`][pysmo.classes.GeoCsvSeismogram.sourceid],
    which keeps the shorter GeoCSV `SID` header form. This is parse-time
    metadata and is not updated when other attributes change.
    """

    publication_version: int = field(converter=int)
    """miniSEED publication (quality) version this segment was read from."""

    @property
    def network(self) -> str:
        """Network code, derived from `sourceid` (read-only)."""
        return pymseed.sourceid2nslc(self.sourceid)[0]

    @property
    def name(self) -> str:
        """Station code, derived from `sourceid` (read-only)."""
        return pymseed.sourceid2nslc(self.sourceid)[1]

    @property
    def location(self) -> str:
        """Location code, derived from `sourceid` (read-only)."""
        return pymseed.sourceid2nslc(self.sourceid)[2]

    @property
    def channel(self) -> str:
        """Channel code, derived from `sourceid` (read-only)."""
        return pymseed.sourceid2nslc(self.sourceid)[3]

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

    @classmethod
    def _from_segment(
        cls, sourceid: str, publication_version: int, segment: MS3TraceSeg
    ) -> Self:
        return cls(
            begin_time=pd.Timestamp(segment.starttime, unit="ns", tz="UTC"),
            delta=pd.Timedelta(seconds=1.0 / segment.samprate),
            data=segment.np_datasamples,
            sourceid=sourceid,
            publication_version=publication_version,
        )

    @classmethod
    def _all_from_tracelist(cls, tracelist: MS3TraceList) -> list[Self]:
        return [cls._from_segment(*entry) for entry in _segments(tracelist)]

    @classmethod
    def _one_from_tracelist(cls, tracelist: MS3TraceList, source: str) -> Self:
        segments = cls._all_from_tracelist(tracelist)
        if len(segments) == 1:
            return segments[0]
        if not segments:
            raise ValueError(f"No miniSEED data found in {source}.")
        segment_lines = "\n".join(
            f"  {s.network}.{s.name}.{s.location}.{s.channel}  "
            + f"{s.begin_time} -- {s.end_time}"
            for s in segments
        )
        raise ValueError(
            f"{source} holds {len(segments)} contiguous segments; "
            + "MSeed.from_bytes()/from_file() requires exactly one. Use "
            + "MSeed.all_from_bytes()/all_from_file() instead. Segments found:\n"
            + f"{segment_lines}"
        )

    @classmethod
    def from_file(cls, filename: str | PathLike[str]) -> Self:
        """Create a new instance from a miniSEED file holding exactly one contiguous segment.

        Args:
            filename: Path to the miniSEED file to read.

        Returns:
            A new MSeed instance.

        Raises:
            ValueError: If the file holds zero, or more than one, contiguous
                segment (a data gap, or more than one channel).
            pymseed.MiniSEEDError: If the file cannot be read as miniSEED.
        """
        with MS3TraceList.from_file(filename, unpack_data=True) as tracelist:
            return cls._one_from_tracelist(tracelist, f"file {filename!r}")

    @classmethod
    def all_from_file(cls, filename: str | PathLike[str]) -> list[Self]:
        """Create one instance per contiguous segment in a miniSEED file.

        Args:
            filename: Path to the miniSEED file to read.

        Returns:
            One MSeed instance per contiguous segment, grouped by source
            identifier and ordered by time within each. Empty if the file
            holds no data.

        Raises:
            pymseed.MiniSEEDError: If the file cannot be read as miniSEED.
        """
        with MS3TraceList.from_file(filename, unpack_data=True) as tracelist:
            return cls._all_from_tracelist(tracelist)

    @classmethod
    def from_bytes(cls, data: bytes) -> Self:
        """Create a new instance from miniSEED bytes holding exactly one contiguous segment.

        Args:
            data: Raw miniSEED bytes.

        Returns:
            A new MSeed instance.

        Raises:
            ValueError: If the data holds zero, or more than one, contiguous
                segment (a data gap, or more than one channel).
            pymseed.MiniSEEDError: If the data cannot be read as miniSEED.
        """
        with MS3TraceList.from_buffer(data, unpack_data=True) as tracelist:
            return cls._one_from_tracelist(tracelist, "the given bytes")

    @classmethod
    def all_from_bytes(cls, data: bytes) -> list[Self]:
        """Create one instance per contiguous segment in miniSEED bytes.

        Args:
            data: Raw miniSEED bytes.

        Returns:
            One MSeed instance per contiguous segment, grouped by source
            identifier and ordered by time within each. Empty if the data
            holds no segments.

        Raises:
            pymseed.MiniSEEDError: If the data cannot be read as miniSEED.
        """
        with MS3TraceList.from_buffer(data, unpack_data=True) as tracelist:
            return cls._all_from_tracelist(tracelist)

    @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.traveltime.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_mseed`][] and
        [`from_bytes`][pysmo.classes.MSeed.from_bytes] /
        [`all_from_bytes`][pysmo.classes.MSeed.all_from_bytes] 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 MSeed instance.

        Raises:
            ValueError: If no waveform data is returned for the given
                window, or more than one contiguous segment is returned
                (a data gap, or a wildcarded channel/location code matching
                more than one channel).
            urllib3.exceptions.ResponseError: If the dataselect web service
                returns an HTTP error.

        Examples:
            <!-- skip: start if(not run_real_web_requests) -->
            ```python
            >>> import pandas as pd
            >>> from pysmo import MiniStation
            >>> from pysmo.classes import MSeed
            >>> station = MiniStation(
            ...     name="ANMO", network="IU", location="00", channel="LHZ",
            ...     latitude=34.945981, longitude=-106.457133,
            ... )
            >>> seismogram = MSeed.fetch(
            ...     station=station,
            ...     starttime=pd.Timestamp("2010-02-27T06:44:00Z"),
            ...     endtime=pd.Timestamp("2010-02-27T06:54:00Z"),
            ... )
            >>>
            ```
            <!-- skip: end -->
        """
        starttime = convert_to_utc_timestamp(starttime)
        endtime = convert_to_utc_timestamp(endtime)
        waveform_bytes = fetch_mseed(
            station=station, starttime=starttime, endtime=endtime
        )
        if not waveform_bytes:
            raise ValueError(
                "No waveform data returned for "
                + f"{station.network}.{station.name}.{station.location}."
                + f"{station.channel} between {starttime} and {endtime}."
            )
        return cls.from_bytes(waveform_bytes)

    def write(self, path: str | PathLike[str]) -> None:
        """Write this seismogram to a miniSEED file.

        Samples are written as `float64` (uncompressed) and the
        publication version is set to 1 — `sourceid` and timing are
        preserved, `publication_version` is not. For STEIM integer
        compression, or to write several seismograms into one file, use
        [`pysmo.lib.io.write_mseed`][] directly.

        Args:
            path: Destination file path. Any existing content is
                overwritten.
        """
        identity = MiniStationCode(
            network=self.network,
            name=self.name,
            location=self.location,
            channel=self.channel,
        )
        write_mseed([(identity, self)], path)