Spaces:
Running
Running
File size: 1,999 Bytes
3bb804c |
1 2 3 4 5 6 7 8 9 10 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 |
from typing import Any, Generic, Optional, TypeVar, Union
import numpy as np
from numpy.typing import NDArray
_DigitalDtype = TypeVar("_DigitalDtype", bound=Union[np.int16, np.int32])
class LazyLoader(Generic[_DigitalDtype]):
"""
Class to load data for a single signal from a buffer of EDF data records (array or memmap).
Parameters
----------
buffer : numpy.ndarray or numpy.memmap
Buffer of EDF data records.
start_sample : int
Offset of the signal in the data records (in samples).
end_sample : int
End of the signal in the data records (in samples).
"""
def __init__(
self,
buffer: Union[NDArray[_DigitalDtype], np.memmap[Any, np.dtype[_DigitalDtype]]],
start_sample: int,
end_sample: int,
) -> None:
self.buffer = buffer
self.start_sample = start_sample
self.end_sample = end_sample
def load(
self, start_record: Optional[int] = None, end_record: Optional[int] = None
) -> NDArray[_DigitalDtype]:
"""
Load signal data from the buffer.
Parameters
----------
start_record : int, optional
The first EDF data record to load samples from. If None, load from the beginning of the buffer.
end_record : int, optional
The last EDF data record to load samples from. If None, load until the end of the buffer.
Returns
-------
numpy.ndarray
Signal data (digital).
"""
if start_record is None:
start_record = 0
if end_record is None:
end_record = self.buffer.shape[0]
if (
end_record < start_record
or start_record < 0
or end_record > self.buffer.shape[0]
):
raise ValueError("Invalid slice: Slice exceeds EDF duration")
return self.buffer[
start_record:end_record, self.start_sample : self.end_sample
].flatten()
|