import numpy as np
from dataclasses import dataclass, field
import pprint
from ada.data_containers._base import _Epoched
from ada.data_containers.generic import GenericData
[docs]
@dataclass(slots=True, eq=False)
class EpochedActivityIndex(_Epoched):
"""A class for storing and handling data epoched by the ActivityIndex method."""
_original_format_metadata: dict = field(init=False)
def __post_init__(self):
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'epoching_method_metadata': 3,
'fs': self._fs,
'channel_names': self._channel_names}
def __repr__(self):
str1 = pprint.pformat(self._metadata, indent=4)
str2 = pprint.pformat(self._epoching_method_metadata, indent=4)
return 'Raw metadata:\n{}\n\nEpoching metadata:\n{}'.format(str1, str2)
@property
def to_score(self) -> np.ndarray:
return self._data[0]
@property
def timestamp(self) -> np.ndarray:
return self._data[1]
@property
def first_sample_timestamp(self) -> float:
"""Unix timestamp of first sample."""
device = GenericData._get_device(self)
return device._convert_timestamp(self.timestamp[0], self._metadata, True)
@property
def id(self) -> str:
return GenericData._get_id(self)
[docs]
def export(self, path: str):
"""Exports object data to the generic format. All metadata are preserved in the file.
Args:
path (str): Path to the .ada file.
"""
to_export = [self._data, self._metadata, self._original_format_metadata, self._epoching_method_metadata]
GenericData._generic_export(path, to_export)
[docs]
@staticmethod
def load_file(path: str) -> "EpochedActivityIndex":
"""Loading file saved in the generic format provided by this package.
Args:
path (str): Path to the .ada file.
Returns:
EpochedActivityIndex: Object containing data.
"""
data, metadata, fs, channel_names, epoching_method_metadata, _ = GenericData._generic_load(path)
if epoching_method_metadata is None:
raise ValueError("Provided file is not epoched!")
elif epoching_method_metadata['epoching method'] != 'Activity Index':
raise ValueError("Provided file is not epoched by chosen method. Use container designed for {}".format(epoching_method_metadata['epoching method']))
return EpochedActivityIndex(data, metadata, fs, channel_names, epoching_method_metadata)
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "EpochedActivityIndex":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Unix timestamp of output data beginning.
end_ts (float | None): Unix timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
EpochedActivityIndex: Object containing the cutted data.
"""
data_cut = GenericData._generic_cut(self, start_ts, end_ts)
return EpochedActivityIndex(data_cut, self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int) -> "EpochedActivityIndex":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int): Sample after the last sample of output data.
Returns:
EpochedActivityIndex: Object containing the cutted data.
"""
return EpochedActivityIndex(self._data[:, start_sample:end_sample], self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
@dataclass(slots=True, eq=False)
class EpochedCK(_Epoched):
"""A class for storing and handling data epoched by the Cole-Kripke-like method method."""
_original_format_metadata: dict = field(init=False)
def __post_init__(self):
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'epoching_method_metadata': 3,
'fs': self._fs,
'channel_names': self._channel_names}
def __repr__(self):
str1 = pprint.pformat(self._metadata, indent=4)
str2 = pprint.pformat(self._epoching_method_metadata, indent=4)
return 'Raw metadata:\n{}\n\nEpoching metadata:\n{}'.format(str1, str2)
@property
def to_score(self) -> np.ndarray:
return self._data[self._epoching_method_metadata['to_score channel']]
@property
def timestamp(self) -> np.ndarray:
return self._data[self._channel_names.index('timestamp')]
@property
def first_sample_timestamp(self) -> float:
"""Unix timestamp of first sample."""
device = GenericData._get_device(self)
return device._convert_timestamp(self.timestamp[0], self._metadata, True)
@property
def id(self) -> str:
return GenericData._get_id(self)
[docs]
def export(self, path: str):
"""Exports object data to the generic format. All metadata are preserved in the file.
Args:
path (str): Path to the .ada file.
"""
to_export = [self._data, self._metadata, self._original_format_metadata, self._epoching_method_metadata]
GenericData._generic_export(path, to_export)
[docs]
@staticmethod
def load_file(path: str) -> "EpochedCK":
"""Loading file saved in the generic format provided by this package.
Args:
path (str): Path to the .ada file.
Returns:
EpochedCK: Object containing data.
"""
data, metadata, fs, channel_names, epoching_method_metadata, _ = GenericData._generic_load(path)
if epoching_method_metadata is None:
raise ValueError("Provided file is not epoched!")
elif epoching_method_metadata['epoching method'] != 'Cole-Kripke style':
raise ValueError("Provided file is not epoched by chosen method. Use container designed for {}".format(epoching_method_metadata['epoching method']))
return EpochedCK(data, metadata, fs, channel_names, epoching_method_metadata)
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "EpochedCK":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Unix timestamp of output data beginning.
end_ts (float | None): Unix timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
EpochedCK: Object containing the cutted data.
"""
data_cut = GenericData._generic_cut(self, start_ts, end_ts)
return EpochedCK(data_cut, self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int) -> "EpochedCK":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int): Sample after the last sample of output data.
Returns:
EpochedCK: Object containing the cutted data.
"""
return EpochedCK(self._data[:, start_sample:end_sample], self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
@dataclass(slots=True, eq=False)
class Resampled(_Epoched):
"""A class for storing and handling data epoched by the standard downsampling."""
_original_format_metadata: dict = field(init=False)
def __post_init__(self):
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'epoching_method_metadata': 3,
'fs': self._fs,
'channel_names': self._channel_names}
def __repr__(self):
str1 = pprint.pformat(self._metadata, indent=4)
str2 = pprint.pformat(self._epoching_method_metadata, indent=4)
return 'Raw metadata:\n{}\n\nEpoching metadata:\n{}'.format(str1, str2)
@property
def to_score(self) -> np.ndarray:
return self._data[self._epoching_method_metadata['to_score channel']]
@property
def timestamp(self) -> np.ndarray:
return self._data[self._channel_names.index('timestamp')]
@property
def first_sample_timestamp(self) -> float:
"""Unix timestamp of first sample."""
device = GenericData._get_device(self)
return device._convert_timestamp(self.timestamp[0], self._metadata, True)
@property
def id(self) -> str:
return GenericData._get_id(self)
[docs]
def export(self, path: str):
"""Exports object data to the generic format. All metadata are preserved in the file.
Args:
path (str): Path to the .ada file.
"""
to_export = [self._data, self._metadata, self._original_format_metadata, self._epoching_method_metadata]
GenericData._generic_export(path, to_export)
[docs]
@staticmethod
def load_file(path: str) -> "Resampled":
"""Loading file saved in the generic format provided by this package.
Args:
path (str): Path to the .ada file.
Returns:
Resampled: Object containing data.
"""
data, metadata, fs, channel_names, epoching_method_metadata, _ = GenericData._generic_load(path)
if epoching_method_metadata is None:
raise ValueError("Provided file is not epoched!")
elif epoching_method_metadata['epoching method'] != 'standard resampling':
raise ValueError("Provided file is not epoched by chosen method. Use container designed for {}".format(epoching_method_metadata['epoching method']))
return Resampled(data, metadata, fs, channel_names, epoching_method_metadata)
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "Resampled":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Unix timestamp of output data beginning.
end_ts (float | None): Unix timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
Resampled: Object containing the cutted data.
"""
data_cut = GenericData._generic_cut(self, start_ts, end_ts)
return Resampled(data_cut, self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int) -> "Resampled":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int): Sample after the last sample of output data.
Returns:
Resampled: Object containing the cutted data.
"""
return Resampled(self._data[:, start_sample:end_sample], self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
@dataclass(slots=True, eq=False)
class EpochedMIMS(_Epoched):
"""A class for storing and handling data epoched by the MIMS method."""
_original_format_metadata: dict = field(init=False)
def __post_init__(self):
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'epoching_method_metadata': 3,
'fs': self._fs,
'channel_names': self._channel_names}
def __repr__(self):
str1 = pprint.pformat(self._metadata, indent=4)
str2 = pprint.pformat(self._epoching_method_metadata, indent=4)
return 'Raw metadata:\n{}\n\nEpoching metadata:\n{}'.format(str1, str2)
@property
def to_score(self) -> np.ndarray:
return self._data[self._epoching_method_metadata['to_score channel']]
@property
def timestamp(self) -> np.ndarray:
return self._data[self._channel_names.index('timestamp')]
@property
def first_sample_timestamp(self) -> float:
"""Unix timestamp of first sample."""
device = GenericData._get_device(self)
return device._convert_timestamp(self.timestamp[0], self._metadata, True)
@property
def id(self) -> str:
return GenericData._get_id(self)
[docs]
def export(self, path: str):
"""Exports object data to the generic format. All metadata are preserved in the file.
Args:
path (str): Path to the .ada file.
"""
to_export = [self._data, self._metadata, self._original_format_metadata, self._epoching_method_metadata]
GenericData._generic_export(path, to_export)
[docs]
@staticmethod
def load_file(path: str) -> "EpochedMIMS":
"""Loading file saved in the generic format provided by this package.
Args:
path (str): Path to the .ada file.
Returns:
Resampled: Object containing data.
"""
data, metadata, fs, channel_names, epoching_method_metadata, _ = GenericData._generic_load(path)
if epoching_method_metadata is None:
raise ValueError("Provided file is not epoched!")
elif epoching_method_metadata['epoching method'] != 'standard resampling':
raise ValueError("Provided file is not epoched by chosen method. Use container designed for {}".format(epoching_method_metadata['epoching method']))
return EpochedMIMS(data, metadata, fs, channel_names, epoching_method_metadata)
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "EpochedMIMS":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Unix timestamp of output data beginning.
end_ts (float | None): Unix timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
Resampled: Object containing the cutted data.
"""
data_cut = GenericData._generic_cut(self, start_ts, end_ts)
return EpochedMIMS(data_cut, self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int) -> "EpochedMIMS":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int): Sample after the last sample of output data.
Returns:
Resampled: Object containing the cutted data.
"""
return EpochedMIMS(self._data[:, start_sample:end_sample], self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
@dataclass(slots=True, eq=False)
class GenericMVM(_Epoched):
"""A class for storing and handling data epoched by the MVM method provided the that was not GeneActivRaw."""
_original_format_metadata: dict = field(init=False)
def __post_init__(self):
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'epoching_method_metadata': 3,
'fs': self._fs,
'channel_names': self._channel_names}
def __repr__(self):
str1 = pprint.pformat(self._metadata, indent=4)
str2 = pprint.pformat(self._epoching_method_metadata, indent=4)
return 'Raw metadata:\n{}\n\nEpoching metadata:\n{}'.format(str1, str2)
@property
def to_score(self) -> np.ndarray:
return self._data[self._epoching_method_metadata['to_score channel']]
@property
def timestamp(self) -> np.ndarray:
return self._data[self._channel_names.index('timestamp')]
@property
def first_sample_timestamp(self) -> float:
"""Unix timestamp of first sample."""
device = GenericData._get_device(self)
return device._convert_timestamp(self.timestamp[0], self._metadata, True)
@property
def id(self) -> str:
return GenericData._get_id(self)
[docs]
def export(self, path: str):
"""Exports object data to the generic format. All metadata are preserved in the file.
Args:
path (str): Path to the .ada file.
"""
to_export = [self._data, self._metadata, self._original_format_metadata, self._epoching_method_metadata]
GenericData._generic_export(path, to_export)
[docs]
@staticmethod
def load_file(path: str) -> "GenericMVM":
"""Loading file saved in the generic format provided by this package.
Args:
path (str): Path to the .ada file.
Returns:
GenericMVM: Object containing data.
"""
data, metadata, fs, channel_names, epoching_method_metadata, _ = GenericData._generic_load(path)
if epoching_method_metadata is None:
raise ValueError("Provided file is not epoched!")
elif epoching_method_metadata['epoching method'] != 'standard resampling':
raise ValueError("Provided file is not epoched by chosen method. Use container designed for {}".format(epoching_method_metadata['epoching method']))
return GenericMVM(data, metadata, fs, channel_names, epoching_method_metadata)
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "GenericMVM":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Unix timestamp of output data beginning.
end_ts (float | None): Unix timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
GenericMVM: Object containing the cutted data.
"""
data_cut = GenericData._generic_cut(self, start_ts, end_ts)
return GenericMVM(data_cut, self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int) -> "GenericMVM":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int): Sample after the last sample of output data.
Returns:
GenericMVM: Object containing the cutted data.
"""
return GenericMVM(self._data[:, start_sample:end_sample], self._metadata, self._fs, self._channel_names, self._epoching_method_metadata)