sgnligo.sources.framecachesrc

Read GW frame files from a frame cache file.

FrameReader dataclass

Bases: TSSource

              flowchart TD
              sgnligo.sources.framecachesrc.FrameReader[FrameReader]

              

              click sgnligo.sources.framecachesrc.FrameReader href "" "sgnligo.sources.framecachesrc.FrameReader"
            

Read GW frame files from a frame cache file

Parameters:

Name	Type	Description	Default
channel_names	Optional[list[str]]	list[str], a list of channel names of the data, e.g., ["L1:GWOSC-16KHZ_R1_STRAIN", "L1:GWOSC-16KHZ_R1_DQMASK"]. Source pads will be automatically generated for each channel, with channel name as pad name.	None
framecache	Optional[str]	str, cache file to read data from	None
instrument	Optional[str]	str, optional, only read gwf files from this instrument	None

Source code in sgnligo/sources/framecachesrc.py

@dataclass
class FrameReader(TSSource):
    """Read GW frame files from a frame cache file

    Args:
        channel_names:
            list[str], a list of channel names of the data, e.g.,
            ["L1:GWOSC-16KHZ_R1_STRAIN", "L1:GWOSC-16KHZ_R1_DQMASK"]. Source pads will
            be automatically generated for each channel, with channel name as pad name.
        framecache:
            str, cache file to read data from
        instrument:
            str, optional, only read gwf files from this instrument
    """

    channel_names: Optional[list[str]] = None
    framecache: Optional[str] = None
    instrument: Optional[str] = None

    def __post_init__(self):
        if len(self.source_pad_names) > 0:
            if self.source_pad_names != tuple(self.channel_names):
                raise ValueError("Expected source pad names to match channel names")
        else:
            print(f"Generating source pads from channel names {self.channel_names}...")
            self.source_pad_names = tuple(self.channel_names)

        # Sanity check
        assert isinstance(self.channel_names, list)
        assert isinstance(self.framecache, str)

        super().__post_init__()
        self.cnt = {p: 0 for p in self.source_pads}

        self.last_epoch = self.start

        if self.instrument is not None:
            for c in self.channel_names:
                assert (
                    self.instrument in c
                ), "instrument provided does not match channel name"

        # init analysis segment
        self.analysis_seg = segments.segment(
            LIGOTimeGPS(self.start), LIGOTimeGPS(self.end)
        )

        # load the cache file
        print(f"Loading {self.framecache}...")
        cache = list(map(CacheEntry, open(self.framecache)))

        if self.instrument is not None:
            # only keep files with the correct instrument
            # sometimes there are frame files with multiple instruments, in that case
            # don't filter by instrument
            cache = [
                c
                for c in cache
                for o in c.observatory
                if o in self.ifo_strings(self.instrument)
            ]

        # only keep files that intersect the analysis segment
        self.cache = []
        for c in cache:
            try:
                self.analysis_seg & c.segment
            except ValueError:
                continue
            else:
                self.cache.append(c)

        # make sure it is sorted by gps time
        self.cache.sort(key=lambda x: x.segment[0])

        # Check if there are missing segments
        segment_remaining = self.analysis_seg
        missing_segments = []
        for c in self.cache:
            if segment_remaining in c.segment:
                # the cache contains all the rest of the proposed segment
                segment_remaining = segments.segment(0, 0)
            elif segment_remaining[0] < c.segment[0]:
                # there is a discontinuity
                missing_segments.append(
                    segments.segment(segment_remaining[0], c.segment[0])
                )
                if c.segment[1] <= segment_remaining[1]:
                    segment_remaining = segments.segment(
                        c.segment[1], segment_remaining[1]
                    )
                else:
                    segment_remaining = segments.segment(0, 0)
            else:
                segment_remaining -= c.segment

        if segment_remaining:
            missing_segments.append(segment_remaining)

        if missing_segments:
            logger.getChild(self.name).warning(
                "%s has missing segment %s, padding with gaps",
                self.name,
                missing_segments,
            )

        self.A = {c: Audioadapter() for c in self.channel_names}

        # load first segment of data to read sample rate
        self.rates = {}
        self.load_gwf_data(self.cache[0])

        # FIXME: support multiple pads with different sample rates
        # FIXME: do we want multiple channels in one buffer?
        for pad in self.source_pads:
            self.set_pad_buffer_params(
                sample_shape=(), rate=self.rates[self.rsrcs[pad]], pad=pad
            )

        # now that we have loaded data from this frame,
        # remove it from the cache
        self.cache.pop(0)

    @staticmethod
    def ifo_strings(ifo: str) -> tuple[str, str]:
        """Make a tuple of possible ifo strings, with and without the "1" at the end.
        I dont know if the given self.instrument will be in the form of e.g., "H" or
        "H1", just make a tuple of both options for string comparison

        Args:
            ifo:
                str, the ifo name, e.g., "H" or "H1"

        Returns:
            tuple[str, str], a tuple of the ifo name with and without the "1" at the end
        """
        if ifo[-1] == "1":
            return (ifo[0], ifo)
        else:
            return (ifo, ifo + "1")

    @staticmethod
    @contextmanager
    def _localize(frame_file: CacheEntry):
        """Yield a local filesystem path for ``frame_file``.

        Local entries are yielded as-is. Remote entries are streamed to a
        NamedTemporaryFile and retried on network-level errors.
        """
        if frame_file.scheme in (None, "", "file"):
            yield frame_file.path
            return

        # CacheEntry.url drops to the opaque "scheme:/path" form when host is
        # empty (typical for osdf:/... cache lines), which requests won't
        # match against its "osdf://" adapter mount. Rebuild explicitly so the
        # URL always starts with "scheme://".
        url = f"{frame_file.scheme}://{frame_file.host}{frame_file.path}"
        with tempfile.NamedTemporaryFile(suffix=".gwf") as tf:
            for attempt in range(_LOCALIZE_MAX_ATTEMPTS):
                try:
                    tf.seek(0)
                    tf.truncate()
                    with requests_pelican.get(
                        url, stream=True, timeout=_LOCALIZE_TIMEOUT
                    ) as resp:
                        resp.raise_for_status()
                        expected = resp.headers.get("Content-Length")
                        expected = int(expected) if expected else None
                        got = 0
                        for chunk in resp.iter_content(_LOCALIZE_CHUNK_SIZE):
                            if chunk:
                                tf.write(chunk)
                                got += len(chunk)
                        if expected is not None and got < expected:
                            raise ChunkedEncodingError(
                                f"short read: {got}/{expected} bytes"
                            )
                    tf.flush()
                    break
                except (ChunkedEncodingError, RequestsConnectionError):
                    if attempt == _LOCALIZE_MAX_ATTEMPTS - 1:
                        raise
                    backoff = 2**attempt
                    _localize_logger.warning(
                        "fetch of %s failed on attempt %d/%d; retrying in %ds",
                        url,
                        attempt + 1,
                        _LOCALIZE_MAX_ATTEMPTS,
                        backoff,
                        exc_info=True,
                    )
                    time.sleep(backoff)
            yield tf.name

    def load_gwf_data(self, frame_file: CacheEntry):
        """Load timeseries data from a gwf frame file.

        Args:
            frame_file:
                CacheEntry, the gwf frame file to read timeseries data from

        Returns:
            dict[str, np.ndarray], a dictionary with channel names as keys and
            numpy arrays of timeseries data as values
        """

        # get first cache entry
        segment = frame_file.segment

        intersection = self.analysis_seg & segment
        start = intersection[0]
        end = intersection[1]

        # lalframe (via gwpy) segfaults nondeterministically when reading
        # straight from pelican-backed URLs; stage remote frames to a local
        # temp file first so lalframe only ever sees plain file paths.
        with self._localize(frame_file) as local_path:
            data_dict = TimeSeriesDict.read(
                local_path, self.channel_names, start=start, end=end
            )

        if len(self.rates) == 0:
            for channel, data in data_dict.items():
                self.rates[channel] = int(data.sample_rate.value)

        for channel, data in data_dict.items():
            if self.last_epoch < start:
                print(
                    f"Unepected epoch: {start}, expected: {self.last_epoch}, sending "
                    "gap buffer"
                )
                self.A[channel].push(
                    SeriesBuffer(
                        offset=Offset.fromsec(float(self.last_epoch)),
                        sample_rate=self.rates[channel],
                        data=None,
                        shape=(int((start - self.last_epoch) * self.rates[channel]),),
                    )
                )
            elif self.last_epoch > start:
                raise ValueError(
                    f"Unepected epoch: {start}, expected: {self.last_epoch}, sending "
                    "gap buffer"
                )
            self.A[channel].push(
                SeriesBuffer(
                    offset=Offset.fromsec(float(start)),
                    sample_rate=self.rates[channel],
                    data=np.array(data),
                )
            )

        self.last_epoch = end

    def internal(self) -> None:
        """Check if we need to read the next gw frame file in the cache. All channels
        are read at once. If the cache is exhausted but the analysis window has not
        yet been drained, pad the remainder with a gap buffer so downstream sees a
        clean EOS instead of starving on an under-filled audioadapter.
        """
        super().internal()

        # load next frame of data from disk when we have less than
        # one buffer length of data left
        read_new = False
        for channel, adapter in self.A.items():
            if adapter.size < self.num_samples(self.rates[channel]):
                read_new = True
                break

        if not read_new:
            return

        if self.cache:
            # Read multiple channels at once
            self.load_gwf_data(self.cache[0])

            # now that we have loaded data from this frame,
            # remove it from the cache
            self.cache.pop(0)
            return

        # Cache exhausted; pad any remaining window with a gap buffer.
        assert self.last_epoch is not None and self.end is not None
        if self.last_epoch < self.end:
            gap_start = float(self.last_epoch)
            gap_end = float(self.end)
            for channel in self.A:
                rate = self.rates[channel]
                nsamples = int(round((gap_end - gap_start) * rate))
                if nsamples <= 0:
                    continue
                self.A[channel].push(
                    SeriesBuffer(
                        offset=Offset.fromsec(gap_start),
                        sample_rate=rate,
                        data=None,
                        shape=(nsamples,),
                    )
                )
            self.last_epoch = LIGOTimeGPS(self.end)

    def new(self, pad: SourcePad) -> TSFrame:
        """New frames are created on "pad" with an instance specific count and a name
        derived from the channel name. "EOS" is set once we have procssed all data in
        the cache within the analysis segment.

        Args:
            pad:
                SourcePad, the pad for which to produce a new TSFrame

        Returns:
            TSFrame, the TSFrame that carries a list of SeriesBuffers
        """

        self.cnt[pad] += 1

        channel = self.rsrcs[pad]

        metadata = {"cnt": self.cnt[pad], "name": "'%s'" % pad.name}

        frame = self.prepare_frame(pad, metadata=metadata)

        if self.A[channel].end_offset < frame.end_offset:
            raise RuntimeError(
                f"{self.name}: audioadapter for {channel} underfilled "
                f"({self.A[channel].end_offset} < {frame.end_offset}); "
                "internal() must load or pad before new()."
            )

        bufs = self.A[channel].get_sliced_buffers((frame.offset, frame.end_offset))
        frame.set_buffers(bufs)
        self.A[channel].flush_samples_by_end_offset(frame.end_offset)

        return frame

ifo_strings(ifo) staticmethod

Make a tuple of possible ifo strings, with and without the "1" at the end. I dont know if the given self.instrument will be in the form of e.g., "H" or "H1", just make a tuple of both options for string comparison

Parameters:

Name	Type	Description	Default
ifo	str	str, the ifo name, e.g., "H" or "H1"	required

Returns:

Type	Description
tuple[str, str]	tuple[str, str], a tuple of the ifo name with and without the "1" at the end

Source code in sgnligo/sources/framecachesrc.py

@staticmethod
def ifo_strings(ifo: str) -> tuple[str, str]:
    """Make a tuple of possible ifo strings, with and without the "1" at the end.
    I dont know if the given self.instrument will be in the form of e.g., "H" or
    "H1", just make a tuple of both options for string comparison

    Args:
        ifo:
            str, the ifo name, e.g., "H" or "H1"

    Returns:
        tuple[str, str], a tuple of the ifo name with and without the "1" at the end
    """
    if ifo[-1] == "1":
        return (ifo[0], ifo)
    else:
        return (ifo, ifo + "1")

internal()

Check if we need to read the next gw frame file in the cache. All channels are read at once. If the cache is exhausted but the analysis window has not yet been drained, pad the remainder with a gap buffer so downstream sees a clean EOS instead of starving on an under-filled audioadapter.

Source code in sgnligo/sources/framecachesrc.py

def internal(self) -> None:
    """Check if we need to read the next gw frame file in the cache. All channels
    are read at once. If the cache is exhausted but the analysis window has not
    yet been drained, pad the remainder with a gap buffer so downstream sees a
    clean EOS instead of starving on an under-filled audioadapter.
    """
    super().internal()

    # load next frame of data from disk when we have less than
    # one buffer length of data left
    read_new = False
    for channel, adapter in self.A.items():
        if adapter.size < self.num_samples(self.rates[channel]):
            read_new = True
            break

    if not read_new:
        return

    if self.cache:
        # Read multiple channels at once
        self.load_gwf_data(self.cache[0])

        # now that we have loaded data from this frame,
        # remove it from the cache
        self.cache.pop(0)
        return

    # Cache exhausted; pad any remaining window with a gap buffer.
    assert self.last_epoch is not None and self.end is not None
    if self.last_epoch < self.end:
        gap_start = float(self.last_epoch)
        gap_end = float(self.end)
        for channel in self.A:
            rate = self.rates[channel]
            nsamples = int(round((gap_end - gap_start) * rate))
            if nsamples <= 0:
                continue
            self.A[channel].push(
                SeriesBuffer(
                    offset=Offset.fromsec(gap_start),
                    sample_rate=rate,
                    data=None,
                    shape=(nsamples,),
                )
            )
        self.last_epoch = LIGOTimeGPS(self.end)

load_gwf_data(frame_file)

Load timeseries data from a gwf frame file.

Parameters:

Name	Type	Description	Default
frame_file	CacheEntry	CacheEntry, the gwf frame file to read timeseries data from	required

Returns:

Type	Description
	dict[str, np.ndarray], a dictionary with channel names as keys and
	numpy arrays of timeseries data as values

Source code in sgnligo/sources/framecachesrc.py

def load_gwf_data(self, frame_file: CacheEntry):
    """Load timeseries data from a gwf frame file.

    Args:
        frame_file:
            CacheEntry, the gwf frame file to read timeseries data from

    Returns:
        dict[str, np.ndarray], a dictionary with channel names as keys and
        numpy arrays of timeseries data as values
    """

    # get first cache entry
    segment = frame_file.segment

    intersection = self.analysis_seg & segment
    start = intersection[0]
    end = intersection[1]

    # lalframe (via gwpy) segfaults nondeterministically when reading
    # straight from pelican-backed URLs; stage remote frames to a local
    # temp file first so lalframe only ever sees plain file paths.
    with self._localize(frame_file) as local_path:
        data_dict = TimeSeriesDict.read(
            local_path, self.channel_names, start=start, end=end
        )

    if len(self.rates) == 0:
        for channel, data in data_dict.items():
            self.rates[channel] = int(data.sample_rate.value)

    for channel, data in data_dict.items():
        if self.last_epoch < start:
            print(
                f"Unepected epoch: {start}, expected: {self.last_epoch}, sending "
                "gap buffer"
            )
            self.A[channel].push(
                SeriesBuffer(
                    offset=Offset.fromsec(float(self.last_epoch)),
                    sample_rate=self.rates[channel],
                    data=None,
                    shape=(int((start - self.last_epoch) * self.rates[channel]),),
                )
            )
        elif self.last_epoch > start:
            raise ValueError(
                f"Unepected epoch: {start}, expected: {self.last_epoch}, sending "
                "gap buffer"
            )
        self.A[channel].push(
            SeriesBuffer(
                offset=Offset.fromsec(float(start)),
                sample_rate=self.rates[channel],
                data=np.array(data),
            )
        )

    self.last_epoch = end

new(pad)

New frames are created on "pad" with an instance specific count and a name derived from the channel name. "EOS" is set once we have procssed all data in the cache within the analysis segment.

Parameters:

Name	Type	Description	Default
pad	SourcePad	SourcePad, the pad for which to produce a new TSFrame	required

Returns:

Type	Description
TSFrame	TSFrame, the TSFrame that carries a list of SeriesBuffers

Source code in sgnligo/sources/framecachesrc.py

def new(self, pad: SourcePad) -> TSFrame:
    """New frames are created on "pad" with an instance specific count and a name
    derived from the channel name. "EOS" is set once we have procssed all data in
    the cache within the analysis segment.

    Args:
        pad:
            SourcePad, the pad for which to produce a new TSFrame

    Returns:
        TSFrame, the TSFrame that carries a list of SeriesBuffers
    """

    self.cnt[pad] += 1

    channel = self.rsrcs[pad]

    metadata = {"cnt": self.cnt[pad], "name": "'%s'" % pad.name}

    frame = self.prepare_frame(pad, metadata=metadata)

    if self.A[channel].end_offset < frame.end_offset:
        raise RuntimeError(
            f"{self.name}: audioadapter for {channel} underfilled "
            f"({self.A[channel].end_offset} < {frame.end_offset}); "
            "internal() must load or pad before new()."
        )

    bufs = self.A[channel].get_sliced_buffers((frame.offset, frame.end_offset))
    frame.set_buffers(bufs)
    self.A[channel].flush_samples_by_end_offset(frame.end_offset)

    return frame