import numpy as np
import csv
from tqdm import tqdm
from math import ceil
from ada.data_containers._base import _Epoched
from ada.data_containers.scored import ScoredShort
[docs]
class Mesa(ScoredShort, _Epoched): # type: ignore[misc]
"""A class for storing data from the MESA dataset. Do not convert to generic!! These files are specific, are scored and epoched, but also do not contain absolute timestamp at all, and therefore are not compatible with Generic type."""
__slots__ = ('_epoching_method_metadata', '_scoring_method_metadata', '_original_format_metadata')
def __init__(self, data: np.ndarray, metadata: dict, fs: float, epoching_method_metadata: dict, scoring_method_metadata: dict):
self._data = data
self._metadata = metadata
self._fs = fs
self._channel_names = ['score',
'timestamp',
'offwrist',
'activity',
'marker',
'whitelight',
'redlight',
'greenlight',
'bluelight',
'interval',
'dayofweek',
'daybymidnight',
'daybynoon']
self._epoching_method_metadata = epoching_method_metadata
self._scoring_method_metadata = scoring_method_metadata
self._original_format_metadata = {'data': 0,
'metadata': 1,
'original_format_metadata': 2,
'scoring_method_metadata': 3,
'epoching_method_metadata': 4,
'fs': self._fs,
'channel_names': self._channel_names}
@property
def to_score(self) -> np.ndarray:
return self._data[3]
@property
def first_sample_timestamp(self) -> float:
"""Timestamp of first sample (relative)"""
return self.timestamp[0]
[docs]
@staticmethod
def load_file(path: str) -> "Mesa":
"""Loading file saved in the MESA format (csv).
Args:
path (str): Path to the .csv file.
Returns:
Mesa: Object containing data.
"""
def nan_converter(x):
try:
return float(x)
except ValueError:
return np.nan
def interval_converter(x):
if x == 'ACTIVE':
return 1
elif x == 'REST-S':
return 0
elif x == 'REST':
return 0.5
return np.nan
print('Loading data from file: {}'.format(path))
numeric_data = np.loadtxt( # type: ignore[call-overload]
path, skiprows=1,
usecols=(0, 3, 4, 5, 6, 7, 8, 9, 10, 12, 13, 14),
delimiter=',', converters=nan_converter).T
text_data = np.loadtxt( # type: ignore[call-overload]
path, skiprows=1, usecols=(1, 2, 11),
delimiter=',', dtype=str).T
start_idx = np.where(text_data[2] != 'EXCLUDED')[0][0]
end_idx = text_data.shape[1] - np.where(np.flip(text_data[2]) != 'EXCLUDED')[0][0]
numeric_data = numeric_data[:, start_idx:end_idx]
text_data = text_data[:, start_idx:end_idx]
out_data = np.empty((numeric_data.shape[0] + 1, numeric_data.shape[1])) # diregard mesaid here, which lives in column 0
out_data[0] = numeric_data[8]
out_data[1] = np.arange(0, numeric_data.shape[1] * 30, 30)
out_data[2:9] = numeric_data[1:8]
out_data[9] = [interval_converter(x) for x in text_data[2]]
out_data[10:] = numeric_data[9:]
metadata = {
'mesa_id': int(numeric_data[0, 0]),
'sampling_frequency': 1 / 30,
'start_hour': text_data[1, 0],
'first_line': int(text_data[0, 0])
}
print('Done loading from {}'.format(path))
return Mesa(out_data, metadata, 1 / 30, {'epoching method': 'MESA', 'to_score channel': 3}, {'algorithm': 'MESA'})
[docs]
def export(self, path: str):
"""Exports object data to the MESA-formatted csv.
Args:
path (str): Path to the .csv file.
"""
def convert_hour(start_time, timestamp):
timestamp += start_time
hour = int(timestamp // 3600)
minute = int((timestamp % 3600) / 60)
second = int((timestamp % 3600) % 60)
return f"{hour:02}:{minute:02}:{second:02}"
def convert_interval(x):
if x == 1:
return 'ACTIVE'
elif x == 0.5:
return 'REST'
elif x == 0:
return 'REST-S'
return 'EXCLUDED'
def convert_nan(x):
return [e if ~np.isnan(e) else '' for e in x]
start_hour = self.metadata['start_hour'].split(':')
start_time = int(start_hour[0]) * 3600 + int(start_hour[1]) * 60 + int(start_hour[2])
top_row = ['mesaid',
'line',
'linetime',
'offwrist',
'activity',
'marker',
'whitelight',
'redlight',
'greenlight',
'bluelight',
'wake',
'interval',
'dayofweek',
'daybymidnight',
'daybynoon']
with open(path, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(top_row)
for i in tqdm(range(self._data.shape[1]), position=0, leave=True, desc='Saving to file: {}'.format(path), dynamic_ncols=True):
row = [self._metadata['mesa_id'], self._metadata['first_line'] + i, convert_hour(start_time, self.timestamp[i])]
writer.writerow(row + convert_nan(self._data[2:9, i]) + convert_nan([self.score[i]]) + [convert_interval(self._data[9, i])] + convert_nan(self._data[10:, i]))
[docs]
def cut_by_samples(self, start_sample: int, end_sample: int | None) -> "Mesa":
"""Create new object with the data cut by given indexes.
Args:
start_sample (int): First sample of output data.
end_sample (int | None): Sample after the last sample of output data. If None, last sample of output data will be last sample of input data.
Returns:
Mesa: Object containing the cutted data.
"""
if end_sample is None:
end_sample = self._data.shape[1] - 1
metadata = self._metadata.copy()
metadata['first_line'] = metadata['first_line'] + start_sample
return Mesa(self._data[:, start_sample:end_sample], metadata, self._fs, self._epoching_method_metadata, self._scoring_method_metadata) # type: ignore[arg-type]
[docs]
def cut_by_timestamp(self, start_ts: float, end_ts: float | None) -> "Mesa":
"""Create new object with the data cut by given timestamps.
Args:
start_ts (float): Relative timestamp of output data beginning.
end_ts (float | None): Relative timestamp of output data end. If None, last sample of output data will be last sample of input data.
Returns:
Mesa: Object containing the cutted data.
"""
if end_ts is not None:
return self.cut_by_samples(ceil(start_ts / 30), ceil(end_ts / 30))
return self.cut_by_samples(ceil(start_ts / 30), None)