diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index d0a77126..d7bed79d 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -26,6 +26,7 @@ jobs: with: python-version: 3.8 - run: python -m pip install -r requirements.txt -r requirements-dev.txt + - run: python -m pytest - run: python -m PyInstaller --onefile app.py - uses: actions/upload-artifact@v2 with: diff --git a/backend/app.py b/backend/app.py index 216b71f3..54272a57 100644 --- a/backend/app.py +++ b/backend/app.py @@ -3,14 +3,14 @@ from waitress import serve from http import HTTPStatus +from backend.request import ClassificationRequest +from backend.response import ClassificationResponse +from backend.spectrogram_generator import SpectrogramGenerator from classification.parser import get_raw_array from classification.exceptions import ClassificationError from classification.config.constants import Sex, ALLOWED_FILE_EXTENSIONS from classification.model import SleepStagesClassifier -from classification.request import ClassificationRequest -from classification.response import ClassificationResponse from classification.features.preprocessing import preprocess -from classification.spectrogram_generator import SpectrogramGenerator app = Flask(__name__) sleep_stage_classifier = SleepStagesClassifier() diff --git a/backend/assets/readme.md b/backend/assets/readme.md index 09e50e65..38c97f4f 100644 --- a/backend/assets/readme.md +++ b/backend/assets/readme.md @@ -18,21 +18,25 @@ "report": { "sleepOnset": 1602211380, // Time at which the subject fell asleep (time of the first non-wake epoch) "sleepOffset": 1602242425, // Time at which the subject woke up (time of the epoch after the last non-wake epoch) - "wakeAfterSleepOffset": 500, // [seconds] (wakeUpTime - sleepOffset) - "totalSleepTime": 31045, // Total amount of time sleeping including nocturnal awakenings (sleepOffset - sleepOnset) - "efficientSleepTime": 27113, // Total amount of seconds passed in non-wake stages - "sleepEffeciency": 0.8733, // Overall sense of how well the patient slept (totalSleepTime/bedTime) - "totalWASO": 3932, // Total amount of time passed in nocturnal awakenings. It is the total time passed in non-wake stage from sleep Onset to sleep offset (totalSleepTime - efficientSleepTime) - "sleepLatency": 1000, // Time to fall asleep [seconds] (sleepOnset - bedTime) "remOnset": 1602214232, // First REM epoch + + "sleepLatency": 1000, // Time to fall asleep [seconds] (sleepOnset - bedTime) "remLatency": 3852, // [seconds] (remOnset- bedTime) + + "sleepEfficiency": 0.8733, // Overall sense of how well the patient slept (totalSleepTime/bedTime) "awakenings": 7, // number of times the subject woke up between sleep onset & offset "stageShifts": 89, // number of times the subject transitionned from one stage to another between sleep onset & offset - "totalWTime": 3932, // [seconds] time passed in this stage between bedTime to wakeUpTime - "totalREMTime": 2370, - "totalN1Time": 3402, - "totalN2Time": 16032, - "totalN3Time": 5309 + + + "wakeAfterSleepOffset": 500, // [seconds] (wakeUpTime - sleepOffset) + "efficientSleepTime": 27113, // Total amount of seconds passed in non-wake stages + "WASO": 3932, // Total amount of time passed in nocturnal awakenings. It is the total time passed in non-wake stage from sleep Onset to sleep offset (totalSleepTime - efficientSleepTime) + "WTime": 3932, // [seconds] time passed in this stage between bedTime to wakeUpTime + "SleepTime": 31045, // Total amount of time sleeping including nocturnal awakenings (sleepOffset - sleepOnset) + "REMTime": 2370, + "N1Time": 3402, + "N2Time": 16032, + "N3Time": 5309 }, "epochs": { "timestamps": [ diff --git a/backend/backend/metric.py b/backend/backend/metric.py new file mode 100644 index 00000000..c9c5615a --- /dev/null +++ b/backend/backend/metric.py @@ -0,0 +1,153 @@ +from collections import Counter +import numpy as np + +from classification.config.constants import SleepStage, EPOCH_DURATION + + +class Metrics(): + def __init__(self, sleep_stages, bedtime): + self.sleep_stages = sleep_stages + self.bedtime = bedtime + self.has_slept = len(np.unique(self.sleep_stages)) != 1 or np.unique(self.sleep_stages)[0] != SleepStage.W.name + + self.is_sleeping_stages = self.sleep_stages != SleepStage.W.name + self.sleep_indexes = np.where(self.is_sleeping_stages)[0] + self.is_last_stage_sleep = self.sleep_stages[-1] != SleepStage.W.name + + self._initialize_sleep_offset() + self._initialize_sleep_latency() + self._initialize_rem_latency() + self._initialize_transition_based_metrics() + + @property + def report(self): + report = { + 'sleepOffset': self._sleep_offset, + 'sleepLatency': self._sleep_latency, + 'remLatency': self._rem_latency, + 'awakenings': self._awakenings, + 'stageShifts': self._stage_shifts, + 'sleepTime': self._sleep_time, + 'WASO': self._wake_after_sleep_onset, + 'sleepEfficiency': self._sleep_efficiency, + 'efficientSleepTime': self._efficient_sleep_time, + 'wakeAfterSleepOffset': self._wake_after_sleep_offset, + 'sleepOnset': self._sleep_onset, + 'remOnset': self._rem_onset, + **self._time_passed_in_stage, + } + + for metric in report: + # json does not recognize NumPy data types + if isinstance(report[metric], np.int64): + report[metric] = int(report[metric]) + + return report + + @property + def _sleep_time(self): + if not self.has_slept: + return 0 + + return self._sleep_offset - self._sleep_onset + + @property + def _wake_after_sleep_onset(self): + if not self.has_slept: + return 0 + + return self._sleep_time - self._efficient_sleep_time + + @property + def _time_passed_in_stage(self): + """Calculates time passed in each stage for all of the sequence""" + nb_epoch_passed_by_stage = Counter(self.sleep_stages) + + def get_time_passed(stage): + return EPOCH_DURATION * nb_epoch_passed_by_stage[stage] if stage in nb_epoch_passed_by_stage else 0 + + return { + f"{stage.upper()}Time": get_time_passed(stage) + for stage in SleepStage.tolist() + } + + @property + def _sleep_efficiency(self): + return len(self.sleep_indexes) / len(self.sleep_stages) + + @property + def _efficient_sleep_time(self): + return len(self.sleep_indexes) * EPOCH_DURATION + + @property + def _wake_after_sleep_offset(self): + if not self.has_slept: + return 0 + + wake_after_sleep_offset_nb_epochs = ( + len(self.sleep_stages) - self.sleep_indexes[-1] - 1 + ) if not self.is_last_stage_sleep else 0 + + return wake_after_sleep_offset_nb_epochs * EPOCH_DURATION + + @property + def _sleep_onset(self): + if not self.has_slept: + return None + + return self._sleep_latency + self.bedtime + + @property + def _rem_onset(self): + rem_latency = self._rem_latency + if rem_latency is None: + return None + + return rem_latency + self.bedtime + + def _initialize_sleep_offset(self): + if not self.has_slept: + sleep_offset = None + else: + sleep_nb_epochs = (self.sleep_indexes[-1] + 1) if len(self.sleep_indexes) else len(self.sleep_stages) + sleep_offset = sleep_nb_epochs * EPOCH_DURATION + self.bedtime + + self._sleep_offset = sleep_offset + + def _initialize_sleep_latency(self): + self._sleep_latency = self._get_latency_of_stage(self.is_sleeping_stages) + + def _initialize_rem_latency(self): + """Time it took to enter REM stage""" + self._rem_latency = self._get_latency_of_stage(self.sleep_stages == SleepStage.REM.name) + + def _initialize_transition_based_metrics(self): + consecutive_stages_occurences = Counter(zip(self.sleep_stages[:-1], self.sleep_stages[1:])) + occurences_by_transition = { + consecutive_stages: consecutive_stages_occurences[consecutive_stages] + for consecutive_stages in consecutive_stages_occurences if consecutive_stages[0] != consecutive_stages[1] + } + transition_occurences = list(occurences_by_transition.values()) + awakenings_occurences = [ + occurences_by_transition[transition_stages] + for transition_stages in occurences_by_transition + if transition_stages[0] != SleepStage.W.name + and transition_stages[1] == SleepStage.W.name + ] + nb_stage_shifts = sum(transition_occurences) + nb_awakenings = sum(awakenings_occurences) + + if self.is_last_stage_sleep and self.has_slept: + nb_stage_shifts += 1 + nb_awakenings += 1 + + self._stage_shifts = nb_stage_shifts + self._awakenings = nb_awakenings + + def _get_latency_of_stage(self, sequence_is_stage): + epochs_of_stage_of_interest = np.where(sequence_is_stage)[0] + + if len(epochs_of_stage_of_interest) == 0: + return None + + return epochs_of_stage_of_interest[0] * EPOCH_DURATION diff --git a/backend/classification/request.py b/backend/backend/request.py similarity index 89% rename from backend/classification/request.py rename to backend/backend/request.py index 38b1e2ee..9e872cc4 100644 --- a/backend/classification/request.py +++ b/backend/backend/request.py @@ -12,18 +12,21 @@ class ClassificationRequest(): - def __init__(self, sex, age, stream_start, bedtime, wakeup, raw_eeg): + def __init__(self, sex, age, stream_start, bedtime, wakeup, raw_eeg, stream_duration=None): self.sex = sex self.age = age self.stream_start = stream_start self.bedtime = bedtime self.wakeup = wakeup - - self.stream_duration = raw_eeg.times[-1] self.raw_eeg = raw_eeg + self.stream_duration = stream_duration if stream_duration else self._get_stream_duration() self._validate() + def _get_stream_duration(self): + PERIOD_DURATION = 1 / self.raw_eeg.info['sfreq'] + return self.raw_eeg.times[-1] + PERIOD_DURATION + @property def in_bed_seconds(self): """timespan, in seconds, from which the subject started the recording and went to bed""" diff --git a/backend/classification/response.py b/backend/backend/response.py similarity index 70% rename from backend/classification/response.py rename to backend/backend/response.py index dce5a2b3..9b5c6455 100644 --- a/backend/classification/response.py +++ b/backend/backend/response.py @@ -1,5 +1,6 @@ import numpy as np +from backend.metric import Metrics from classification.config.constants import EPOCH_DURATION, SleepStage @@ -15,42 +16,46 @@ def __init__(self, request, predictions, spectrogram): self.spectrogram = spectrogram self.predictions = predictions + self.metrics = Metrics(self.sleep_stages, self.bedtime) @property def sleep_stages(self): - ordered_sleep_stage_names = np.array([SleepStage(stage_index).name for stage_index in range(len(SleepStage))]) + ordered_sleep_stage_names = np.array(SleepStage.tolist()) return ordered_sleep_stage_names[self.predictions] @property - def epochs(self): + def response(self): + return { + 'epochs': self._epochs, + 'report': self._report, + 'metadata': self._metadata, + 'subject': self._subject, + 'spectrograms': self.spectrogram, + } + + @property + def _epochs(self): timestamps = np.arange(self.n_epochs * EPOCH_DURATION, step=EPOCH_DURATION) + self.bedtime return {'timestamps': timestamps.tolist(), 'stages': self.sleep_stages.tolist()} @property - def metadata(self): + def _metadata(self): return { "sessionStartTime": self.stream_start, "sessionEndTime": self.stream_duration + self.stream_start, "totalSessionTime": self.stream_duration, "bedTime": self.bedtime, - "wakeUpTime": None, - "totalBedTime": None, + "wakeUpTime": self.wakeup, + "totalBedTime": self.wakeup - self.bedtime, } @property - def subject(self): + def _subject(self): return { 'age': self.age, 'sex': self.sex.name, } @property - def response(self): - return { - 'epochs': self.epochs, - 'report': None, - 'metadata': self.metadata, - 'subject': self.subject, - 'board': None, - 'spectrograms': self.spectrogram, - } + def _report(self): + return self.metrics.report diff --git a/backend/classification/spectrogram_generator.py b/backend/backend/spectrogram_generator.py similarity index 100% rename from backend/classification/spectrogram_generator.py rename to backend/backend/spectrogram_generator.py diff --git a/backend/classification/config/constants.py b/backend/classification/config/constants.py index 9974cc31..2d7bd3f9 100644 --- a/backend/classification/config/constants.py +++ b/backend/classification/config/constants.py @@ -15,6 +15,10 @@ class SleepStage(Enum): N3 = 3 REM = 4 + @staticmethod + def tolist(): + return [e.name for e in SleepStage] + class HiddenMarkovModelProbability(Enum): emission = auto() diff --git a/backend/readme.md b/backend/readme.md index e0e942d2..b398b55c 100644 --- a/backend/readme.md +++ b/backend/readme.md @@ -41,6 +41,14 @@ If you want to run the backend with hot reload enabled (you must have installed hupper -m waitress app:app ``` +## Run the tests + +You can run our unit tests with the following command, after installing the development requirements: + +```bash +pytest +``` + ## Profile application - Run `python profiler.py` diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt index 002c9813..3438bd5d 100644 --- a/backend/requirements-dev.txt +++ b/backend/requirements-dev.txt @@ -1,4 +1,5 @@ hupper==1.10.2 pyinstaller==4.0 +pytest==6.1.2 snakeviz==2.1.0 Werkzeug==1.0.1 diff --git a/backend/tests/setup.py b/backend/tests/setup.py new file mode 100644 index 00000000..3aa1526a --- /dev/null +++ b/backend/tests/setup.py @@ -0,0 +1,28 @@ +from unittest.mock import patch + +from backend.request import ClassificationRequest +from classification.config.constants import Sex + + +def pytest_generate_tests(metafunc): + # called once per each test function + funcarglist = metafunc.cls.params[metafunc.function.__name__] + argnames = sorted(funcarglist[0]) + metafunc.parametrize( + argnames, [[funcargs[name] for name in argnames] for funcargs in funcarglist] + ) + + +def get_mock_request(): + with patch.object(ClassificationRequest, '_validate', lambda *x, **y: None): + mock_request = ClassificationRequest( + sex=Sex.M, + age=22, + stream_start=1582418280, + bedtime=1582423980, + wakeup=1582452240, + raw_eeg=None, + stream_duration=35760, + ) + + return mock_request diff --git a/backend/tests/test_response.py b/backend/tests/test_response.py new file mode 100644 index 00000000..44cfee2e --- /dev/null +++ b/backend/tests/test_response.py @@ -0,0 +1,417 @@ +""" +Not tested as they seemed obvious: +- "SleepTime": 31045, // Total amount of time sleeping including nocturnal awakenings (sleepOffset - sleepOnset) +- "WASO": 3932, // Total amount of time passed in nocturnal awakenings. It is the total time passed in non-wake stage + // from sleep Onset to sleep offset (totalSleepTime - efficientSleepTime) +""" + +import numpy as np + +from tests.setup import pytest_generate_tests # noqa: F401 +from tests.setup import get_mock_request +from backend.response import ClassificationResponse +from classification.config.constants import EPOCH_DURATION, SleepStage + + +def get_report(request, sequence): + value_sequence = np.array([SleepStage[stage].value for stage in sequence]) + response = ClassificationResponse(request, value_sequence, None) + return response.metrics.report + + +class TestReportTimePassedInStage(): + """Tests the time passed in each stage metrics in the response metrics + The evaluated metrics are: + "WTime": 3932, // [seconds] time passed in this stage between bedTime to wakeUpTime + "REMTime": 2370, + "N1Time": 3402, + "N2Time": 16032, + "N3Time": 5309 + """ + params = { + "test_null_time_passed_in_stage": [ + dict( + sequence=['W', 'W', 'W'], + WTime=3 * EPOCH_DURATION, + REMTime=0, + N1Time=0, + N2Time=0, + N3Time=0), + dict( + sequence=['REM', 'REM', 'REM'], + WTime=0, + REMTime=3 * EPOCH_DURATION, + N1Time=0, + N2Time=0, + N3Time=0), + dict( + sequence=['N1', 'N1', 'N1'], + WTime=0, + REMTime=0, + N1Time=3 * EPOCH_DURATION, + N2Time=0, + N3Time=0), + dict( + sequence=['N2', 'N2', 'N2'], + WTime=0, + REMTime=0, + N1Time=0, + N2Time=3 * EPOCH_DURATION, + N3Time=0), + dict( + sequence=['N3', 'N3', 'N3'], + WTime=0, + REMTime=0, + N1Time=0, + N2Time=0, + N3Time=3 * EPOCH_DURATION), + ], "test_partial_time_passed_in_stage": [ + dict( + sequence=['W', 'N1', 'N2', 'W'], + WTime=2 * EPOCH_DURATION, + REMTime=0, + N1Time=EPOCH_DURATION, + N2Time=EPOCH_DURATION, + N3Time=0), + dict( + sequence=['N1', 'W', 'W'], + WTime=2 * EPOCH_DURATION, + REMTime=0, + N1Time=EPOCH_DURATION, + N2Time=0, + N3Time=0), + dict( + sequence=['W', 'N1', 'N2', 'N3', 'REM', 'W'], + WTime=2 * EPOCH_DURATION, + REMTime=EPOCH_DURATION, + N1Time=EPOCH_DURATION, + N2Time=EPOCH_DURATION, + N3Time=EPOCH_DURATION), + dict( + sequence=['N1', 'N2', 'N2', 'REM'], + WTime=0, + REMTime=EPOCH_DURATION, + N1Time=EPOCH_DURATION, + N2Time=2 * EPOCH_DURATION, + N3Time=0), + ] + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_null_time_passed_in_stage(self, sequence, WTime, REMTime, N1Time, N2Time, N3Time): + report = get_report(self.MOCK_REQUEST, sequence) + assert report[sequence[0].upper() + 'Time'] == len(sequence) * EPOCH_DURATION + self.assert_times(sequence, report, WTime, REMTime, N1Time, N2Time, N3Time) + + def test_partial_time_passed_in_stage(self, sequence, WTime, REMTime, N1Time, N2Time, N3Time): + report = get_report(self.MOCK_REQUEST, sequence) + self.assert_times(sequence, report, WTime, REMTime, N1Time, N2Time, N3Time) + + def assert_times(self, sequence, report, WTime, REMTime, N1Time, N2Time, N3Time): + assert ( + report['WTime'] + + report['REMTime'] + + report['N1Time'] + + report['N2Time'] + + report['N3Time'] + ) == len(sequence) * EPOCH_DURATION + + assert report['WTime'] == WTime + assert report['REMTime'] == REMTime + assert report['N1Time'] == N1Time + assert report['N2Time'] == N2Time + assert report['N3Time'] == N3Time + + +class TestReportLatenciesOnset(): + """Tests the event-related latencies and onsets + The evaluated metrics are: + "sleepLatency": 1000, // Time to fall asleep [seconds] (sleepOnset - bedTime) + "remLatency": 3852, // [seconds] (remOnset- bedTime) + + "sleepOnset": 1602211380, // Time at which the subject fell asleep (time of the first non-wake epoch) + "remOnset": 1602214232, // First REM epoch + """ + + params = { + "test_sequence_starts_with_stage": [ + dict(sequence=['REM', 'REM', 'W', 'W'], test_rem=True), + dict(sequence=['REM', 'W', 'N1', 'W'], test_rem=True), + dict(sequence=['REM', 'W', 'N1', 'N2', 'N3', 'REM', 'W'], test_rem=False), + dict(sequence=['N1', 'W', 'N1', 'N2', 'N3', 'REM', 'W'], test_rem=False), + dict(sequence=['N2', 'W', 'N1', 'N2', 'N3', 'REM', 'W'], test_rem=False), + dict(sequence=['N3', 'W', 'N1', 'N2', 'N3', 'REM', 'W'], test_rem=False), + ], "test_sequence_has_no_stage": [ + dict(sequence=['W', 'N1', 'N2', 'N3', 'W'], test_rem=True), + dict(sequence=['W', 'W', 'W', 'W', 'W'], test_rem=False), + dict(sequence=['W'], test_rem=False), + ], "test_sequence_ends_with_stage": [ + dict(sequence=['W', 'W', 'REM'], test_rem=True), + dict(sequence=['W', 'W', 'N1'], test_rem=False), + dict(sequence=['W', 'W', 'N2'], test_rem=False), + dict(sequence=['W', 'W', 'N3'], test_rem=False), + ], "test_sequence_with_stage_at_middle": [ + dict( + sequence=['W', 'N1', 'N2', 'N1', 'REM', 'W'], + test_rem=True, + latency=4 * EPOCH_DURATION, + ), dict( + sequence=['W', 'W', 'N1', 'W', 'N1', 'W'], + test_rem=False, + latency=2 * EPOCH_DURATION, + ), dict( + sequence=['W', 'W', 'N2', 'W', 'N2', 'W'], + test_rem=False, + latency=2 * EPOCH_DURATION, + ), dict( + sequence=['W', 'W', 'N3', 'W', 'N3', 'W'], + test_rem=False, + latency=2 * EPOCH_DURATION, + ), + ], + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_sequence_starts_with_stage(self, sequence, test_rem): + expected_latency = 0 + expected_onset = self.MOCK_REQUEST.bedtime + self.assert_latency_equals_expected(expected_latency, expected_onset, sequence, test_rem) + + def test_sequence_has_no_stage(self, sequence, test_rem): + expected_latency = None + expected_onset = None + self.assert_latency_equals_expected(expected_latency, expected_onset, sequence, test_rem) + + def test_sequence_ends_with_stage(self, sequence, test_rem): + expected_latency = EPOCH_DURATION * (len(sequence) - 1) + expected_onset = expected_latency + self.MOCK_REQUEST.bedtime + self.assert_latency_equals_expected(expected_latency, expected_onset, sequence, test_rem) + + def test_sequence_with_stage_at_middle(self, sequence, test_rem, latency): + expected_onset = latency + self.MOCK_REQUEST.bedtime + self.assert_latency_equals_expected(latency, expected_onset, sequence, test_rem) + + def get_latency_report_key(self, test_rem): + return 'remLatency' if test_rem else 'sleepLatency' + + def get_onset_report_key(self, test_rem): + return 'remOnset' if test_rem else 'sleepOnset' + + def assert_latency_equals_expected(self, expected_latency, expected_onset, sequence, test_rem): + report = get_report(self.MOCK_REQUEST, sequence) + assert report[self.get_latency_report_key(test_rem)] == expected_latency, ( + f"Latency of {'rem' if test_rem else 'sleep'} is not as expected" + ) + assert report[self.get_onset_report_key(test_rem)] == expected_onset, ( + f"Onset of {'rem' if test_rem else 'sleep'} is not as expected" + ) + + +class TestReportSleepOffset(): + """Tests timestamp at which user woke up + "sleepOffset": 1602242425, // Time at which the subject woke up (time of the epoch after the last non-wake epoch) + """ + params = { + 'test_wake_up_end': [dict( + sequence=['W', 'N1', 'N2', 'N3', 'REM', 'N1', 'W'], + )], 'test_wake_up_middle': [dict( + sequence=['W', 'N1', 'N2', 'W', 'W', 'W', 'W'], + awake_index=3, + expected_wake_after_sleep_offset=4 * EPOCH_DURATION, + )], 'test_awakes_and_goes_back_to_sleep_and_wakes': [dict( + sequence=['W', 'N1', 'N2', 'W', 'N1', 'N2', 'W'], + awake_index=6, + expected_wake_after_sleep_offset=EPOCH_DURATION, + )], 'test_awakes_and_goes_back_to_sleep_and_doesnt_awake': [dict( + sequence=['W', 'N1', 'N2', 'W', 'N1', 'N2', 'N2'], + )], 'test_always_awake': [ + dict(sequence=['W', 'W', 'W']), + dict(sequence=['W']), + ], 'test_doesnt_awaken': [ + dict(sequence=['W', 'N1', 'N2']), + dict(sequence=['N1', 'N1', 'N2']), + ], + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_wake_up_end(self, sequence): + expected_sleep_offset = self.MOCK_REQUEST.bedtime + EPOCH_DURATION * (len(sequence) - 1) + expected_wake_after_sleep_offset = EPOCH_DURATION + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def test_wake_up_middle(self, sequence, awake_index, expected_wake_after_sleep_offset): + expected_sleep_offset = self.MOCK_REQUEST.bedtime + EPOCH_DURATION * awake_index + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def test_awakes_and_goes_back_to_sleep_and_wakes(self, sequence, awake_index, expected_wake_after_sleep_offset): + expected_sleep_offset = self.MOCK_REQUEST.bedtime + EPOCH_DURATION * awake_index + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def test_awakes_and_goes_back_to_sleep_and_doesnt_awake(self, sequence): + expected_sleep_offset = self.MOCK_REQUEST.bedtime + EPOCH_DURATION * len(sequence) + expected_wake_after_sleep_offset = 0 + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def test_always_awake(self, sequence): + expected_sleep_offset = None + expected_wake_after_sleep_offset = 0 + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def test_doesnt_awaken(self, sequence): + expected_sleep_offset = self.MOCK_REQUEST.bedtime + EPOCH_DURATION * len(sequence) + expected_wake_after_sleep_offset = 0 + self.assert_sleep_offset_with_wake(sequence, expected_sleep_offset, expected_wake_after_sleep_offset) + + def assert_sleep_offset_with_wake(self, sequence, expected_sleep_offset, expected_wake_after_sleep_offset): + report = get_report(self.MOCK_REQUEST, sequence) + assert report['sleepOffset'] == expected_sleep_offset + assert report['wakeAfterSleepOffset'] == expected_wake_after_sleep_offset + + +class TestReportSleepEfficiency(): + """Tests sleep efficiency related metrics + The evaluated metrics are: + "sleepEfficiency": 0.8733, // Overall sense of how well the patient slept (totalSleepTime/bedTime) + "efficientSleepTime": 27113, // Total amount of seconds passed in non-wake stages + """ + params = { + 'test_sleep_time_null': [ + dict(sequence=['W', 'W', 'W']), + dict(sequence=['W']), + ], 'test_sleep_time_not_null': [ + dict( + sequence=['W', 'W', 'N1', 'W'], + expected_efficiency=(1 / 4), + expected_efficient_sleep_time=EPOCH_DURATION + ), dict( + sequence=['W', 'W', 'N2', 'W'], + expected_efficiency=(1 / 4), + expected_efficient_sleep_time=EPOCH_DURATION + ), dict( + sequence=['W', 'W', 'N3', 'W'], + expected_efficiency=(1 / 4), + expected_efficient_sleep_time=EPOCH_DURATION + ), dict( + sequence=['W', 'W', 'REM', 'W'], + expected_efficiency=(1 / 4), + expected_efficient_sleep_time=EPOCH_DURATION + ), dict( + sequence=['W', 'W', 'N1', 'N2', 'N3', 'REM', 'N1', 'W'], + expected_efficiency=(5 / 8), + expected_efficient_sleep_time=5 * EPOCH_DURATION + ), + ] + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_sleep_time_null(self, sequence): + self.assert_sleep_efficiency(sequence, expected_efficiency=0, expected_efficient_sleep_time=0) + + def test_sleep_time_not_null(self, sequence, expected_efficiency, expected_efficient_sleep_time): + self.assert_sleep_efficiency(sequence, expected_efficiency, expected_efficient_sleep_time) + + def assert_sleep_efficiency(self, sequence, expected_efficiency, expected_efficient_sleep_time): + report = get_report(self.MOCK_REQUEST, sequence) + assert report['sleepEfficiency'] == expected_efficiency + assert report['efficientSleepTime'] == expected_efficient_sleep_time + + +class TestReportAwakenings(): + """Tests number of awakenings per night + "awakenings": 7, // number of times the subject woke up between sleep onset & offset + """ + params = { + 'test_sleep_time_null': [dict(sequence=['W', 'W', 'W']), dict(sequence=['W'])], + 'test_one_awakening': [ + dict(sequence=['W', 'N1', 'W']), + dict(sequence=['W', 'N1', 'N2', 'N3', 'W', 'W']), + ], 'test_doesnt_awaken': [ + dict(sequence=['W', 'N1', 'N2', 'N3', 'REM']), + dict(sequence=['W', 'N1']), + ], 'test_many_awakening': [ + dict(sequence=['W', 'N1', 'W', 'N1', 'W'], nb_awakenings=2), + dict(sequence=['W', 'N1', 'N2', 'W', 'N1', 'W', 'W', 'N1', 'W'], nb_awakenings=3), + ], + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_sleep_time_null(self, sequence): + self.assert_sleep_efficiency(sequence, expected_nb_awakenings=0) + + def test_one_awakening(self, sequence): + self.assert_sleep_efficiency(sequence, expected_nb_awakenings=1) + + def test_doesnt_awaken(self, sequence): + self.assert_sleep_efficiency(sequence, expected_nb_awakenings=1) + + def test_many_awakening(self, sequence, nb_awakenings): + self.assert_sleep_efficiency(sequence, expected_nb_awakenings=nb_awakenings) + + def assert_sleep_efficiency(self, sequence, expected_nb_awakenings): + report = get_report(self.MOCK_REQUEST, sequence) + assert report['awakenings'] == expected_nb_awakenings + + +class TestReportStageShifts(): + """Test number of stage shifts per night + "stageShifts": 89, // number of times the subject transitionned + // from one stage to another between sleep onset & offset + """ + params = { + 'test_sleep_time_null': [ + dict(sequence=['W', 'W', 'W']), + dict(sequence=['W']), + ], 'test_one_sleep_stage': [ + dict(sequence=['W', 'W', 'N1', 'N1', 'N1', 'W']), + dict(sequence=['W', 'N1', 'W']), + ], 'test_sleep_with_awakenings': [ + dict(sequence=['W', 'N1', 'W', 'N1', 'N1', 'W'], sleep_shifts=4), + dict(sequence=['W', 'N1', 'W', 'N2', 'W', 'N3', 'W'], sleep_shifts=6), + ], 'test_does_not_awaken': [ + dict(sequence=['W', 'N1', 'N1'], sleep_shifts=2), + dict(sequence=['W', 'N1', 'N2'], sleep_shifts=3), + dict(sequence=['W', 'N1', 'N3'], sleep_shifts=3), + dict(sequence=['W', 'N1', 'REM'], sleep_shifts=3), + ], 'test_many_sleep_stages': [ + dict(sequence=['W', 'N1', 'N2', 'N2', 'W'], sleep_shifts=3), + dict(sequence=['W', 'N1', 'N1', 'N3', 'N3', 'REM', 'REM', 'N1', 'W'], sleep_shifts=5), + ], + } + + @classmethod + def setup_class(cls): + cls.MOCK_REQUEST = get_mock_request() + + def test_sleep_time_null(self, sequence): + self.assert_sleep_efficiency(sequence, expected_sleep_shifts=0) + + def test_one_sleep_stage(self, sequence): + self.assert_sleep_efficiency(sequence, expected_sleep_shifts=2) + + def test_sleep_with_awakenings(self, sequence, sleep_shifts): + self.assert_sleep_efficiency(sequence, expected_sleep_shifts=sleep_shifts) + + def test_does_not_awaken(self, sequence, sleep_shifts): + self.assert_sleep_efficiency(sequence, expected_sleep_shifts=sleep_shifts) + + def test_many_sleep_stages(self, sequence, sleep_shifts): + self.assert_sleep_efficiency(sequence, expected_sleep_shifts=sleep_shifts) + + def assert_sleep_efficiency(self, sequence, expected_sleep_shifts): + report = get_report(self.MOCK_REQUEST, sequence) + assert report['stageShifts'] == expected_sleep_shifts