Skip to content

This is an abstration to use multiple measurement files as a continuous data stream that can be indexed seamlessly.

Tip

This is especially useful if the measurements with the Analyzer4D software are consecutive measurements that are only restarted to keep file sizes manageable.

qass.tools.analyzer.consecutive_stream

Classes:

Name Description
ConsecutiveStream

This class is a collection of data streams that are or were recorded seamlessly

ConsecutiveStream

This class is a collection of data streams that are or were recorded seamlessly without any gap between streams. Data streams are collected in a FIFO buffer and the class provides an interface that allows to index this collection as if it were a single data stream file.

Parameters:

Name Type Description Default
max_streams int

The maximum number of streams that is held in the FIFO buffer (minimum = 1)

 required

Examples:

>>> from pathlib import Path

>>> from qass.tools.analyzer.buffer_parser import Buffer
>>> from qass.tools.analyzer.consecutive_stream import ConsecutiveStream

>>> file_one = Path("file1") # has 100 specs
>>> file_two = Path("file2") # has 150 specs
>>> con_stream = ConsecutiveStream(max_streams=2)
>>> con_stream.add_stream(Buffer(file_one))
>>> con_stream.add_stream(Buffer(file_two))
>>> con_stream.spec_count # yields 250
>>> con_stream.min_sample_idx # yields 0
>>> con_stream.max_sample_idx # yields 250
>>> con_stream.get_data(50, 75)
yields the range 50 - 75 from the first file
>>> con_stream.get_data(150, 200)
yields the range 50 - 100 from the second file
>>> con_stream.get_data(75, 125)
yields the range 75 - 100 from the first file
and the range 0 - 25 from the second file

Raises:

Type Description
InvalidDataStream

If multiple aggregations or channels are found in the input streams
Note

If max_streams < len(streams) the first len(streams) - max_streams data streams will be discarded.

Methods:

Name Description
__init__
__repr__
__str__
add_stream

Add a new data stream to the FIFO buffer and return the start index of
add_streams

Add a collection of stream objects at once
get_data

Fetch a range of spectra from the data
get_stream

Returns the stream at stream_idx from the FIFO buffer.
get_stream_start_idx

Returns the global start idx of a given stream_idx
time_to_spec
to_local_idx

Convert an global index into the ConsecutiveStream into a local idx

Attributes:

Name Type Description
avg_frq int
avg_time int
channel int
compression_frq int
compression_time int
frq_bands int
frq_per_band int
max_sample_idx int

The maximum sample idx that is available in the FIFO buffer
min_sample_idx int

The minimum sample idx that is still available in the FIFO buffer
ref_energy float
spec_count int

Returns the maximum global sample idx available in the FIFO buffer
spec_duration int
Source code in src/qass/tools/analyzer/consecutive_stream.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
class ConsecutiveStream:
    """
    This class is a collection of data streams that are or were recorded seamlessly
    without any gap between streams.
    Data streams are collected in a FIFO buffer and the class provides an interface
    that allows to index this collection as if it were a single data stream file.

    Parameters
    ----------
    max_streams : int
        The maximum number of streams that is held in the FIFO buffer (minimum = 1)

    Examples
    -------
    >>> from pathlib import Path

    >>> from qass.tools.analyzer.buffer_parser import Buffer
    >>> from qass.tools.analyzer.consecutive_stream import ConsecutiveStream

    >>> file_one = Path("file1") # has 100 specs
    >>> file_two = Path("file2") # has 150 specs
    >>> con_stream = ConsecutiveStream(max_streams=2)
    >>> con_stream.add_stream(Buffer(file_one))
    >>> con_stream.add_stream(Buffer(file_two))
    >>> con_stream.spec_count # yields 250
    >>> con_stream.min_sample_idx # yields 0
    >>> con_stream.max_sample_idx # yields 250
    >>> con_stream.get_data(50, 75)
    yields the range 50 - 75 from the first file
    >>> con_stream.get_data(150, 200)
    yields the range 50 - 100 from the second file
    >>> con_stream.get_data(75, 125)
    yields the range 75 - 100 from the first file
    and the range 0 - 25 from the second file

    Raises
    ------
    InvalidDataStream
         If multiple aggregations or channels are found in the input streams

    Note
    ----
    If `max_streams` < `len(streams)` the first `len(streams) - max_streams` data streams will be discarded.
    """

    def __init__(self, max_streams: int):
        if max_streams < 1:
            raise ValueError("`max_streams` needs to be greater than zero")
        self._max_streams = max_streams
        self._stream_buffer = deque(maxlen=max_streams)
        # stream indices contain global the start index of each stream
        # stream indices are sorted and monotonically increasing
        self._stream_indices = deque(maxlen=max_streams)

    def to_local_idx(self, idx: int) -> tuple[int, int]:
        """
        Convert an global index into the `ConsecutiveStream` into a local idx
        of a specific stream.
        The function returns the local index and the related data stream file
        index that can be used to retrieve the actual stream instance using
        `ConsecutiveStream.get_stream`.

        Raises
        ------
        LookupError
            If the local data stream is not in the FIFO buffer anymore
        IndexError
            If the local idx is too large for the newest stream

        Parameters
        ----------
        idx : int
            The global index into the `ConsecutiveStream`

        Returns
        -------
        tuple
            A tuple of two indices. The first index is a local index into a
            concrete data stream. The second index is the index of the corresponding
            data stream in the FIFO buffer.
        """
        # INFO: This is binary search and runs in O(log n) with n streams
        # We could still optimize this to amortized O(1) if we cache consecutive accesses
        corresponding_stream_idx = np.searchsorted(self._stream_indices, idx)
        # searchsorted gives the sorted position which is before the related stream
        # in the case of equality and behind the stream if the index is larger
        # in the case it is larger we need to subtract 1 from the index to reference
        # the correct stream
        if (
            corresponding_stream_idx == len(self._stream_indices)
            or idx < self._stream_indices[corresponding_stream_idx]
        ):
            corresponding_stream_idx -= 1

        if corresponding_stream_idx == -1:
            raise LookupError(
                f"Stream for global idx: {idx} has already been evicted from the FIFO buffer"
            )
        stream = self._stream_buffer[corresponding_stream_idx]
        local_idx = idx - self._stream_indices[corresponding_stream_idx]
        if local_idx > stream.spec_count:
            raise IndexError(
                f"Local index {local_idx} is too high for last seen stream "
                f"with {stream.spec_count} samples"
            )
        return local_idx, int(corresponding_stream_idx)

    def add_stream(self, stream: Buffer) -> int:
        """
        Add a new data stream to the FIFO buffer and return the start index of
        the new stream

        Raises
        ------
        InvalidDataStream
            If the provided data stream does not match with the streams that are
            already in the FIFO buffer

        Parameters
        ----------
        stream : Buffer
            The data stream with the same aggregation as the data streams already
            in the FIFO buffer

        Returns
        -------
        int
            The global starting index in the `ConsecutiveStream` of the new stream
        """
        with stream:
            pass
        if len(self._stream_buffer) == 0:
            # initialization
            self._stream_buffer.append(stream)
            self._stream_indices.append(0)
            return self._stream_indices[-1]
        try:
            assert stream.spec_duration == self._stream_buffer[-1].spec_duration
            assert stream.channel == self._stream_buffer[-1].channel
            assert stream.frq_bands == self._stream_buffer[-1].frq_bands
            assert stream.compression_time == self._stream_buffer[-1].compression_time
            assert stream.compression_frq == self._stream_buffer[-1].compression_frq
            assert stream.avg_time == self._stream_buffer[-1].avg_time
            assert stream.avg_frq == self._stream_buffer[-1].avg_frq
        except AssertionError:
            raise InvalidDataStream(
                "The provided data stream does not have the correct aggregation"
            )
        self._stream_indices.append(
            self._stream_indices[-1] + self._stream_buffer[-1].spec_count
        )
        self._stream_buffer.append(stream)
        return self._stream_indices[-1]

    def add_streams(self, streams: List[Buffer]) -> List[int]:
        """Add a collection of stream objects at once

        Note
        ----
        If `len(streams) > ConsecutiveStream.max_streams` only the last
        `max_streams` stream objects will be available.

        Parameters
        ----------
        streams : List[Buffer]
            A list of stream objects that are added to the FIFO buffer

        Returns
        -------
        List[int]
            The global start indices of the added streams.
        """
        start_indices = []
        for stream in streams:
            start_indices.append(self.add_stream(stream))
        return start_indices

    def get_data(
        self, from_: int, to: int, conversion: Union[DataConversion, None] = None
    ) -> npt.NDArray:
        """
        Fetch a range of spectra from the data

        Raises
        ------
        ValueError
            If `from_` >= `to`

        Parameters
        ----------
        from_ : int
            Start spectrum for the range (inclusive)
        to : int
            End spectrum for the range (exclusive)
        conversion : DataConversion, optional
            The used conversion method forwarded to the stream object.

        Returns
        -------
        npt.NDArray
            A numpy array containing the spectra (in case the data stream is a spectrogram)
            or the data points (in case the data stream is a signal stream)
        """
        if from_ >= to:
            raise ValueError(f"`from_` must be lower than `to` but got {(from_, to)}")
        local_from, from_stream_idx = self.to_local_idx(from_)
        local_to, to_stream_idx = self.to_local_idx(to)

        from_stream = self._stream_buffer[from_stream_idx]
        to_stream = self._stream_buffer[to_stream_idx]
        if from_stream_idx == to_stream_idx:
            with from_stream:
                return from_stream.get_data(local_from, local_to, conversion=conversion)
        between_data = []
        for stream_idx in range(from_stream_idx + 1, to_stream_idx):
            stream = self._stream_buffer[stream_idx]
            with stream:
                between_data.append(
                    stream.get_data(0, stream.spec_count, conversion=conversion)
                )

        with from_stream, to_stream:
            left = from_stream.get_data(
                local_from, from_stream.spec_count, conversion=conversion
            )
            right = to_stream.get_data(0, local_to, conversion=conversion)
        return np.concatenate((left, *between_data, right))

    def time_to_spec(self, time: int) -> int:
        return int(time / self._stream_buffer[0].spec_duration)

    def get_stream(self, stream_idx: int) -> Buffer:
        """
        Returns the stream at `stream_idx` from the FIFO buffer.
        The `stream_idx` can be fetched with a call to `self.to_local_idx`

        Parameters
        ----------
        stream_idx : int
            Index into the FIFO buffer
        """
        return self._stream_buffer[stream_idx]

    def get_stream_start_idx(self, stream_idx: int) -> int:
        """
        Returns the global start idx of a given `stream_idx`
        in the FIFO buffer

        Parameters
        ----------
        stream_idx : int
            Index into the FIFO buffer
        """
        return self._stream_indices[stream_idx]

    def __str__(self) -> str:
        return self.__repr__()

    def __repr__(self) -> str:
        return (
            f"ConsecutiveStream(streams={len(self._stream_buffer)}, "
            f"max_streams={self._max_streams})"
        )

    @property
    def frq_bands(self) -> int:
        return self._stream_buffer[0].frq_bands

    @property
    def spec_duration(self) -> int:
        return self._stream_buffer[0].spec_duration

    @property
    def channel(self) -> int:
        return self._stream_buffer[0].channel

    @property
    def compression_time(self) -> int:
        return self._stream_buffer[0].compression_time

    @property
    def compression_frq(self) -> int:
        return self._stream_buffer[0].compression_frq

    @property
    def avg_time(self) -> int:
        return self._stream_buffer[0].avg_time

    @property
    def avg_frq(self) -> int:
        return self._stream_buffer[0].avg_frq

    @property
    def frq_per_band(self) -> int:
        return self._stream_buffer[0].frq_per_band

    @property
    def ref_energy(self) -> float:
        return self._stream_buffer[0].ref_energy

    @property
    def spec_count(self) -> int:
        """Returns the maximum global sample idx available in the FIFO buffer

        Warning
        -------
        This does not mean that all samples up to this sample are still available
        because some of the data streams might have already been evicted from the
        FIFO buffer.
        """
        return self._stream_indices[-1] + self._stream_buffer[-1].spec_count

    @property
    def min_sample_idx(self) -> int:
        """The minimum sample idx that is still available in the FIFO buffer"""
        return self._stream_indices[0]

    @property
    def max_sample_idx(self) -> int:
        """The maximum sample idx that is available in the FIFO buffer"""
        return self._stream_indices[-1] + self._stream_buffer[-1].spec_count

avg_frq property 

avg_frq: int

avg_time property 

avg_time: int

channel property 

channel: int

compression_frq property 

compression_frq: int

compression_time property 

compression_time: int

frq_bands property 

frq_bands: int

frq_per_band property 

frq_per_band: int

max_sample_idx property 

max_sample_idx: int

The maximum sample idx that is available in the FIFO buffer

min_sample_idx property 

min_sample_idx: int

The minimum sample idx that is still available in the FIFO buffer

ref_energy property 

ref_energy: float

spec_count property 

spec_count: int

Returns the maximum global sample idx available in the FIFO buffer

Warning

This does not mean that all samples up to this sample are still available because some of the data streams might have already been evicted from the FIFO buffer.

spec_duration property 

spec_duration: int

__init__ 

__init__(max_streams: int)
Source code in src/qass/tools/analyzer/consecutive_stream.py 
56
57
58
59
60
61
62
63
def __init__(self, max_streams: int):
    if max_streams < 1:
        raise ValueError("`max_streams` needs to be greater than zero")
    self._max_streams = max_streams
    self._stream_buffer = deque(maxlen=max_streams)
    # stream indices contain global the start index of each stream
    # stream indices are sorted and monotonically increasing
    self._stream_indices = deque(maxlen=max_streams)

__repr__ 

__repr__() -> str
Source code in src/qass/tools/analyzer/consecutive_stream.py 
269
270
271
272
273
def __repr__(self) -> str:
    return (
        f"ConsecutiveStream(streams={len(self._stream_buffer)}, "
        f"max_streams={self._max_streams})"
    )

__str__ 

__str__() -> str
Source code in src/qass/tools/analyzer/consecutive_stream.py 
266
267
def __str__(self) -> str:
    return self.__repr__()

add_stream 

add_stream(stream: Buffer) -> int

Add a new data stream to the FIFO buffer and return the start index of the new stream

Raises:

Type Description
InvalidDataStream

If the provided data stream does not match with the streams that are already in the FIFO buffer

Parameters:

Name Type Description Default
stream Buffer

The data stream with the same aggregation as the data streams already in the FIFO buffer

 required

Returns:

Type Description
int

The global starting index in the ConsecutiveStream of the new stream
Source code in src/qass/tools/analyzer/consecutive_stream.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def add_stream(self, stream: Buffer) -> int:
    """
    Add a new data stream to the FIFO buffer and return the start index of
    the new stream

    Raises
    ------
    InvalidDataStream
        If the provided data stream does not match with the streams that are
        already in the FIFO buffer

    Parameters
    ----------
    stream : Buffer
        The data stream with the same aggregation as the data streams already
        in the FIFO buffer

    Returns
    -------
    int
        The global starting index in the `ConsecutiveStream` of the new stream
    """
    with stream:
        pass
    if len(self._stream_buffer) == 0:
        # initialization
        self._stream_buffer.append(stream)
        self._stream_indices.append(0)
        return self._stream_indices[-1]
    try:
        assert stream.spec_duration == self._stream_buffer[-1].spec_duration
        assert stream.channel == self._stream_buffer[-1].channel
        assert stream.frq_bands == self._stream_buffer[-1].frq_bands
        assert stream.compression_time == self._stream_buffer[-1].compression_time
        assert stream.compression_frq == self._stream_buffer[-1].compression_frq
        assert stream.avg_time == self._stream_buffer[-1].avg_time
        assert stream.avg_frq == self._stream_buffer[-1].avg_frq
    except AssertionError:
        raise InvalidDataStream(
            "The provided data stream does not have the correct aggregation"
        )
    self._stream_indices.append(
        self._stream_indices[-1] + self._stream_buffer[-1].spec_count
    )
    self._stream_buffer.append(stream)
    return self._stream_indices[-1]

add_streams 

add_streams(streams: List[Buffer]) -> List[int]

Add a collection of stream objects at once

Note

If len(streams) > ConsecutiveStream.max_streams only the last max_streams stream objects will be available.

Parameters:

Name Type Description Default
streams List[Buffer]

A list of stream objects that are added to the FIFO buffer

 required

Returns:

Type Description
List[int]

The global start indices of the added streams.
Source code in src/qass/tools/analyzer/consecutive_stream.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def add_streams(self, streams: List[Buffer]) -> List[int]:
    """Add a collection of stream objects at once

    Note
    ----
    If `len(streams) > ConsecutiveStream.max_streams` only the last
    `max_streams` stream objects will be available.

    Parameters
    ----------
    streams : List[Buffer]
        A list of stream objects that are added to the FIFO buffer

    Returns
    -------
    List[int]
        The global start indices of the added streams.
    """
    start_indices = []
    for stream in streams:
        start_indices.append(self.add_stream(stream))
    return start_indices

get_data 

get_data(from_: int, to: int, conversion: Union[DataConversion, None] = None) -> NDArray

Fetch a range of spectra from the data

Raises:

Type Description
ValueError

If from_ >= to

Parameters:

Name Type Description Default
from_ int

Start spectrum for the range (inclusive)

 required
to int

End spectrum for the range (exclusive)

 required
conversion DataConversion

The used conversion method forwarded to the stream object.

 None

Returns:

Type Description
NDArray

A numpy array containing the spectra (in case the data stream is a spectrogram) or the data points (in case the data stream is a signal stream)
Source code in src/qass/tools/analyzer/consecutive_stream.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def get_data(
    self, from_: int, to: int, conversion: Union[DataConversion, None] = None
) -> npt.NDArray:
    """
    Fetch a range of spectra from the data

    Raises
    ------
    ValueError
        If `from_` >= `to`

    Parameters
    ----------
    from_ : int
        Start spectrum for the range (inclusive)
    to : int
        End spectrum for the range (exclusive)
    conversion : DataConversion, optional
        The used conversion method forwarded to the stream object.

    Returns
    -------
    npt.NDArray
        A numpy array containing the spectra (in case the data stream is a spectrogram)
        or the data points (in case the data stream is a signal stream)
    """
    if from_ >= to:
        raise ValueError(f"`from_` must be lower than `to` but got {(from_, to)}")
    local_from, from_stream_idx = self.to_local_idx(from_)
    local_to, to_stream_idx = self.to_local_idx(to)

    from_stream = self._stream_buffer[from_stream_idx]
    to_stream = self._stream_buffer[to_stream_idx]
    if from_stream_idx == to_stream_idx:
        with from_stream:
            return from_stream.get_data(local_from, local_to, conversion=conversion)
    between_data = []
    for stream_idx in range(from_stream_idx + 1, to_stream_idx):
        stream = self._stream_buffer[stream_idx]
        with stream:
            between_data.append(
                stream.get_data(0, stream.spec_count, conversion=conversion)
            )

    with from_stream, to_stream:
        left = from_stream.get_data(
            local_from, from_stream.spec_count, conversion=conversion
        )
        right = to_stream.get_data(0, local_to, conversion=conversion)
    return np.concatenate((left, *between_data, right))

get_stream 

get_stream(stream_idx: int) -> Buffer

Returns the stream at stream_idx from the FIFO buffer. The stream_idx can be fetched with a call to self.to_local_idx

Parameters:

Name Type Description Default
stream_idx int

Index into the FIFO buffer

 required
Source code in src/qass/tools/analyzer/consecutive_stream.py 
242
243
244
245
246
247
248
249
250
251
252
def get_stream(self, stream_idx: int) -> Buffer:
    """
    Returns the stream at `stream_idx` from the FIFO buffer.
    The `stream_idx` can be fetched with a call to `self.to_local_idx`

    Parameters
    ----------
    stream_idx : int
        Index into the FIFO buffer
    """
    return self._stream_buffer[stream_idx]

get_stream_start_idx 

get_stream_start_idx(stream_idx: int) -> int

Returns the global start idx of a given stream_idx in the FIFO buffer

Parameters:

Name Type Description Default
stream_idx int

Index into the FIFO buffer

 required
Source code in src/qass/tools/analyzer/consecutive_stream.py 
254
255
256
257
258
259
260
261
262
263
264
def get_stream_start_idx(self, stream_idx: int) -> int:
    """
    Returns the global start idx of a given `stream_idx`
    in the FIFO buffer

    Parameters
    ----------
    stream_idx : int
        Index into the FIFO buffer
    """
    return self._stream_indices[stream_idx]

time_to_spec 

time_to_spec(time: int) -> int
Source code in src/qass/tools/analyzer/consecutive_stream.py 
239
240
def time_to_spec(self, time: int) -> int:
    return int(time / self._stream_buffer[0].spec_duration)

to_local_idx 

to_local_idx(idx: int) -> tuple[int, int]

Convert an global index into the ConsecutiveStream into a local idx of a specific stream. The function returns the local index and the related data stream file index that can be used to retrieve the actual stream instance using ConsecutiveStream.get_stream.

Raises:

Type Description
LookupError

If the local data stream is not in the FIFO buffer anymore
IndexError

If the local idx is too large for the newest stream

Parameters:

Name Type Description Default
idx int

The global index into the ConsecutiveStream

 required

Returns:

Type Description
tuple

A tuple of two indices. The first index is a local index into a concrete data stream. The second index is the index of the corresponding data stream in the FIFO buffer.
Source code in src/qass/tools/analyzer/consecutive_stream.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def to_local_idx(self, idx: int) -> tuple[int, int]:
    """
    Convert an global index into the `ConsecutiveStream` into a local idx
    of a specific stream.
    The function returns the local index and the related data stream file
    index that can be used to retrieve the actual stream instance using
    `ConsecutiveStream.get_stream`.

    Raises
    ------
    LookupError
        If the local data stream is not in the FIFO buffer anymore
    IndexError
        If the local idx is too large for the newest stream

    Parameters
    ----------
    idx : int
        The global index into the `ConsecutiveStream`

    Returns
    -------
    tuple
        A tuple of two indices. The first index is a local index into a
        concrete data stream. The second index is the index of the corresponding
        data stream in the FIFO buffer.
    """
    # INFO: This is binary search and runs in O(log n) with n streams
    # We could still optimize this to amortized O(1) if we cache consecutive accesses
    corresponding_stream_idx = np.searchsorted(self._stream_indices, idx)
    # searchsorted gives the sorted position which is before the related stream
    # in the case of equality and behind the stream if the index is larger
    # in the case it is larger we need to subtract 1 from the index to reference
    # the correct stream
    if (
        corresponding_stream_idx == len(self._stream_indices)
        or idx < self._stream_indices[corresponding_stream_idx]
    ):
        corresponding_stream_idx -= 1

    if corresponding_stream_idx == -1:
        raise LookupError(
            f"Stream for global idx: {idx} has already been evicted from the FIFO buffer"
        )
    stream = self._stream_buffer[corresponding_stream_idx]
    local_idx = idx - self._stream_indices[corresponding_stream_idx]
    if local_idx > stream.spec_count:
        raise IndexError(
            f"Local index {local_idx} is too high for last seen stream "
            f"with {stream.spec_count} samples"
        )
    return local_idx, int(corresponding_stream_idx)