Skip to content

Sleep

Closed-loop sleep entrainment: circadian rhythm optimization, protocol library, session report generation, multi-objective sleep optimization, and EEG-based stage detection.

Circadian Optimizer

sc_neurocore.sleep.circadian_optimizer

Chronotype

Bases: Enum

Sleep chronotype labels (after Michael Breus' model).

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
31
32
33
34
35
36
37
class Chronotype(Enum):
    """Sleep chronotype labels (after Michael Breus' model)."""

    LION = "lion"
    BEAR = "bear"
    WOLF = "wolf"
    DOLPHIN = "dolphin"

CircadianProfile dataclass

A chronotype's circadian parameters.

Attributes

chronotype : Chronotype The chronotype label. bedtime_hour : float Ideal bedtime in decimal hours (0-24, can exceed 24 for next-day). wake_hour : float Ideal wake time in decimal hours. default_protocol : str Name of the recommended sleep audio protocol. melatonin_peak_hour : float Hour at which endogenous melatonin peaks (typically ~2 h after bedtime onset for most chronotypes). core_body_temp_nadir_hour : float Hour of the core body temperature nadir (~2 h after melatonin peak).

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
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
@dataclass
class CircadianProfile:
    """A chronotype's circadian parameters.

    Attributes
    ----------
    chronotype : Chronotype
        The chronotype label.
    bedtime_hour : float
        Ideal bedtime in decimal hours (0-24, can exceed 24 for next-day).
    wake_hour : float
        Ideal wake time in decimal hours.
    default_protocol : str
        Name of the recommended sleep audio protocol.
    melatonin_peak_hour : float
        Hour at which endogenous melatonin peaks (typically ~2 h after
        bedtime onset for most chronotypes).
    core_body_temp_nadir_hour : float
        Hour of the core body temperature nadir (~2 h after melatonin peak).
    """

    chronotype: Chronotype
    bedtime_hour: float
    wake_hour: float
    default_protocol: str
    melatonin_peak_hour: float
    core_body_temp_nadir_hour: float

CircadianOptimizer

Chronotype-aware circadian rhythm optimizer.

Parameters

chronotype : Chronotype The user's chronotype.

Example::

opt = CircadianOptimizer(Chronotype.BEAR)
profile = opt.get_profile()
print(opt.melatonin_level(23.0))  # near-peak for a Bear
Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
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
class CircadianOptimizer:
    """Chronotype-aware circadian rhythm optimizer.

    Parameters
    ----------
    chronotype : Chronotype
        The user's chronotype.

    Example::

        opt = CircadianOptimizer(Chronotype.BEAR)
        profile = opt.get_profile()
        print(opt.melatonin_level(23.0))  # near-peak for a Bear
    """

    def __init__(self, chronotype: Chronotype) -> None:
        self.chronotype = chronotype
        self._profile = _PROFILES[chronotype]

    # -- public API ---------------------------------------------------------

    def get_profile(self) -> CircadianProfile:
        """Return the full circadian profile for the configured chronotype."""
        return self._profile

    def get_sleep_window(self) -> Tuple[float, float]:
        """Return ``(bedtime_hour, wake_hour)``."""
        return (self._profile.bedtime_hour, self._profile.wake_hour)

    def get_recommended_protocol(self) -> str:
        """Return the default protocol name for this chronotype."""
        return self._profile.default_protocol

    def is_in_sleep_window(self, hour: float) -> bool:
        """Check whether *hour* (0-24) falls inside the sleep window.

        Handles windows that wrap past midnight (e.g. 23.5 -> 6.5).
        """
        bed = self._profile.bedtime_hour
        wake = self._profile.wake_hour

        if bed <= wake:
            return bed <= hour < wake
        else:
            # wraps past midnight
            return hour >= bed or hour < wake

    def melatonin_level(self, hour: float) -> float:
        """Estimate melatonin level at *hour* using a sinusoidal model.

        Returns a value in ``[0, 1]`` where 1 is the peak (at
        ``melatonin_peak_hour``) and 0 is the trough (12 h offset).
        """
        peak = self._profile.melatonin_peak_hour
        # phase so that cos(0) = 1 at the peak hour
        phase = 2.0 * math.pi * (hour - peak) / 24.0
        level = 0.5 * (1.0 + math.cos(phase))
        return float(np.clip(level, 0.0, 1.0))

    def to_dict(self) -> Dict[str, Any]:
        """Serialise the optimizer state to a plain dict."""
        p = self._profile
        return {
            "chronotype": self.chronotype.value,
            "bedtime_hour": p.bedtime_hour,
            "wake_hour": p.wake_hour,
            "default_protocol": p.default_protocol,
            "melatonin_peak_hour": p.melatonin_peak_hour,
            "core_body_temp_nadir_hour": p.core_body_temp_nadir_hour,
        }

get_profile()

Return the full circadian profile for the configured chronotype.

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
140
141
142
def get_profile(self) -> CircadianProfile:
    """Return the full circadian profile for the configured chronotype."""
    return self._profile

get_sleep_window()

Return (bedtime_hour, wake_hour).

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
144
145
146
def get_sleep_window(self) -> Tuple[float, float]:
    """Return ``(bedtime_hour, wake_hour)``."""
    return (self._profile.bedtime_hour, self._profile.wake_hour)

Return the default protocol name for this chronotype.

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
148
149
150
def get_recommended_protocol(self) -> str:
    """Return the default protocol name for this chronotype."""
    return self._profile.default_protocol

is_in_sleep_window(hour)

Check whether hour (0-24) falls inside the sleep window.

Handles windows that wrap past midnight (e.g. 23.5 -> 6.5).

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
def is_in_sleep_window(self, hour: float) -> bool:
    """Check whether *hour* (0-24) falls inside the sleep window.

    Handles windows that wrap past midnight (e.g. 23.5 -> 6.5).
    """
    bed = self._profile.bedtime_hour
    wake = self._profile.wake_hour

    if bed <= wake:
        return bed <= hour < wake
    else:
        # wraps past midnight
        return hour >= bed or hour < wake

melatonin_level(hour)

Estimate melatonin level at hour using a sinusoidal model.

Returns a value in [0, 1] where 1 is the peak (at melatonin_peak_hour) and 0 is the trough (12 h offset).

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
166
167
168
169
170
171
172
173
174
175
176
def melatonin_level(self, hour: float) -> float:
    """Estimate melatonin level at *hour* using a sinusoidal model.

    Returns a value in ``[0, 1]`` where 1 is the peak (at
    ``melatonin_peak_hour``) and 0 is the trough (12 h offset).
    """
    peak = self._profile.melatonin_peak_hour
    # phase so that cos(0) = 1 at the peak hour
    phase = 2.0 * math.pi * (hour - peak) / 24.0
    level = 0.5 * (1.0 + math.cos(phase))
    return float(np.clip(level, 0.0, 1.0))

to_dict()

Serialise the optimizer state to a plain dict.

Source code in src/sc_neurocore/sleep/circadian_optimizer.py 
178
179
180
181
182
183
184
185
186
187
188
def to_dict(self) -> Dict[str, Any]:
    """Serialise the optimizer state to a plain dict."""
    p = self._profile
    return {
        "chronotype": self.chronotype.value,
        "bedtime_hour": p.bedtime_hour,
        "wake_hour": p.wake_hour,
        "default_protocol": p.default_protocol,
        "melatonin_peak_hour": p.melatonin_peak_hour,
        "core_body_temp_nadir_hour": p.core_body_temp_nadir_hour,
    }

Protocol Library

sc_neurocore.sleep.protocol_library

StageAudioParams dataclass

Audio-entrainment parameters for a single sleep stage.

Attributes

binaural_hz : float Binaural beat frequency (Hz). noise_color : str Background noise colour ("pink", "brown", "white", etc.). base_freq_hz : float Carrier / base tone frequency (Hz). volume : float Relative volume in [0, 1]. isochronic_hz : float Isochronic pulse frequency (Hz); 0 disables. spatial_rotation : float Spatial audio rotation speed in degrees per second.

Source code in src/sc_neurocore/sleep/protocol_library.py 
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
@dataclass
class StageAudioParams:
    """Audio-entrainment parameters for a single sleep stage.

    Attributes
    ----------
    binaural_hz : float
        Binaural beat frequency (Hz).
    noise_color : str
        Background noise colour (``"pink"``, ``"brown"``, ``"white"``, etc.).
    base_freq_hz : float
        Carrier / base tone frequency (Hz).
    volume : float
        Relative volume in ``[0, 1]``.
    isochronic_hz : float
        Isochronic pulse frequency (Hz); 0 disables.
    spatial_rotation : float
        Spatial audio rotation speed in degrees per second.
    """

    binaural_hz: float = 2.0
    noise_color: str = "pink"
    base_freq_hz: float = 200.0
    volume: float = 0.5
    isochronic_hz: float = 0.0
    spatial_rotation: float = 0.0

SleepProtocol dataclass

A named sleep-entrainment protocol.

Attributes

name : str Human-readable identifier (must match the registry key). description : str Short description of the protocol's therapeutic goal. stage_audio : Dict[SleepStage, StageAudioParams] Audio parameters keyed by target stage. stage_targets : Dict[SleepStage, float] Target fraction of total sleep time per stage (must sum to 1.0). total_duration_min : float Recommended session length in minutes.

Source code in src/sc_neurocore/sleep/protocol_library.py 
 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
@dataclass
class SleepProtocol:
    """A named sleep-entrainment protocol.

    Attributes
    ----------
    name : str
        Human-readable identifier (must match the registry key).
    description : str
        Short description of the protocol's therapeutic goal.
    stage_audio : Dict[SleepStage, StageAudioParams]
        Audio parameters keyed by target stage.
    stage_targets : Dict[SleepStage, float]
        Target fraction of total sleep time per stage (must sum to 1.0).
    total_duration_min : float
        Recommended session length in minutes.
    """

    name: str = ""
    description: str = ""
    stage_audio: Dict[SleepStage, StageAudioParams] = field(default_factory=dict)
    stage_targets: Dict[SleepStage, float] = field(default_factory=dict)
    total_duration_min: float = 480.0  # 8 hours default

    # -- public API ---------------------------------------------------------

    def get_audio_for_stage(self, stage: SleepStage) -> StageAudioParams:
        """Return audio parameters for *stage*, falling back to WAKE params."""
        return self.stage_audio.get(
            stage, self.stage_audio.get(SleepStage.WAKE, StageAudioParams())
        )

    def get_target_stage(self, progress: float) -> SleepStage:
        """Return the ideal stage for a given session *progress* in [0, 1].

        Progress is mapped linearly through the cumulative stage-target
        fractions in stage order (WAKE, N1, N2, N3, REM).
        """
        progress = max(0.0, min(1.0, progress))
        cumulative = 0.0
        for stage in (SleepStage.WAKE, SleepStage.N1, SleepStage.N2, SleepStage.N3, SleepStage.REM):
            cumulative += self.stage_targets.get(stage, 0.0)
            if progress <= cumulative:
                return stage
        return SleepStage.REM  # fallback at end of night

    def to_dict(self) -> Dict[str, Any]:
        """Serialise the protocol to a plain dict."""
        return {
            "name": self.name,
            "description": self.description,
            "total_duration_min": self.total_duration_min,
            "stage_targets": {s.name: v for s, v in self.stage_targets.items()},
            "stage_audio": {
                s.name: {
                    "binaural_hz": a.binaural_hz,
                    "noise_color": a.noise_color,
                    "base_freq_hz": a.base_freq_hz,
                    "volume": a.volume,
                    "isochronic_hz": a.isochronic_hz,
                    "spatial_rotation": a.spatial_rotation,
                }
                for s, a in self.stage_audio.items()
            },
        }

get_audio_for_stage(stage)

Return audio parameters for stage, falling back to WAKE params.

Source code in src/sc_neurocore/sleep/protocol_library.py 
88
89
90
91
92
def get_audio_for_stage(self, stage: SleepStage) -> StageAudioParams:
    """Return audio parameters for *stage*, falling back to WAKE params."""
    return self.stage_audio.get(
        stage, self.stage_audio.get(SleepStage.WAKE, StageAudioParams())
    )

get_target_stage(progress)

Return the ideal stage for a given session progress in [0, 1].

Progress is mapped linearly through the cumulative stage-target fractions in stage order (WAKE, N1, N2, N3, REM).

Source code in src/sc_neurocore/sleep/protocol_library.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def get_target_stage(self, progress: float) -> SleepStage:
    """Return the ideal stage for a given session *progress* in [0, 1].

    Progress is mapped linearly through the cumulative stage-target
    fractions in stage order (WAKE, N1, N2, N3, REM).
    """
    progress = max(0.0, min(1.0, progress))
    cumulative = 0.0
    for stage in (SleepStage.WAKE, SleepStage.N1, SleepStage.N2, SleepStage.N3, SleepStage.REM):
        cumulative += self.stage_targets.get(stage, 0.0)
        if progress <= cumulative:
            return stage
    return SleepStage.REM  # fallback at end of night

to_dict()

Serialise the protocol to a plain dict.

Source code in src/sc_neurocore/sleep/protocol_library.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def to_dict(self) -> Dict[str, Any]:
    """Serialise the protocol to a plain dict."""
    return {
        "name": self.name,
        "description": self.description,
        "total_duration_min": self.total_duration_min,
        "stage_targets": {s.name: v for s, v in self.stage_targets.items()},
        "stage_audio": {
            s.name: {
                "binaural_hz": a.binaural_hz,
                "noise_color": a.noise_color,
                "base_freq_hz": a.base_freq_hz,
                "volume": a.volume,
                "isochronic_hz": a.isochronic_hz,
                "spatial_rotation": a.spatial_rotation,
            }
            for s, a in self.stage_audio.items()
        },
    }

get_protocol(name)

Look up a protocol by name. Raises KeyError if not found.

Source code in src/sc_neurocore/sleep/protocol_library.py 
495
496
497
def get_protocol(name: str) -> SleepProtocol:
    """Look up a protocol by name.  Raises ``KeyError`` if not found."""
    return PROTOCOL_REGISTRY[name]

list_protocols()

Return a sorted list of all available protocol names.

Source code in src/sc_neurocore/sleep/protocol_library.py 
500
501
502
def list_protocols() -> List[str]:
    """Return a sorted list of all available protocol names."""
    return sorted(PROTOCOL_REGISTRY.keys())

Report Generator

sc_neurocore.sleep.report_generator

SleepReport dataclass

Aggregate report for a completed sleep session.

Attributes

total_duration_min : float Total session length in minutes. sleep_onset_latency_min : float Minutes from session start until the first non-WAKE epoch. sleep_efficiency_pct : float Percentage of total time spent asleep (non-WAKE). quality_score : float Composite quality score in [0, 100]. stage_durations_min : Dict[str, float] Time (minutes) per stage. stage_percentages : Dict[str, float] Percentage of total time per stage. stage_targets : Dict[str, float] Protocol target percentages for comparison. hypnogram : List[int] Stage codes per epoch. wakeups : int Number of WAKE epochs that occurred after initial sleep onset. reinductions : int Number of re-induction sequences triggered. recommendations : List[str] Plain-language suggestions. grade : str Letter grade (A-F).

Source code in src/sc_neurocore/sleep/report_generator.py 
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
@dataclass
class SleepReport:
    """Aggregate report for a completed sleep session.

    Attributes
    ----------
    total_duration_min : float
        Total session length in minutes.
    sleep_onset_latency_min : float
        Minutes from session start until the first non-WAKE epoch.
    sleep_efficiency_pct : float
        Percentage of total time spent asleep (non-WAKE).
    quality_score : float
        Composite quality score in ``[0, 100]``.
    stage_durations_min : Dict[str, float]
        Time (minutes) per stage.
    stage_percentages : Dict[str, float]
        Percentage of total time per stage.
    stage_targets : Dict[str, float]
        Protocol target percentages for comparison.
    hypnogram : List[int]
        Stage codes per epoch.
    wakeups : int
        Number of WAKE epochs that occurred after initial sleep onset.
    reinductions : int
        Number of re-induction sequences triggered.
    recommendations : List[str]
        Plain-language suggestions.
    grade : str
        Letter grade (A-F).
    """

    total_duration_min: float = 0.0
    sleep_onset_latency_min: float = 0.0
    sleep_efficiency_pct: float = 0.0
    quality_score: float = 0.0
    stage_durations_min: Dict[str, float] = field(default_factory=dict)
    stage_percentages: Dict[str, float] = field(default_factory=dict)
    stage_targets: Dict[str, float] = field(default_factory=dict)
    hypnogram: List[int] = field(default_factory=list)
    wakeups: int = 0
    reinductions: int = 0
    recommendations: List[str] = field(default_factory=list)
    grade: str = "F"

SleepReportGenerator

Generates a :class:SleepReport from a completed optimiser session.

Source code in src/sc_neurocore/sleep/report_generator.py 
 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
class SleepReportGenerator:
    """Generates a :class:`SleepReport` from a completed optimiser session."""

    @staticmethod
    def generate(optimizer: SleepOptimizer) -> SleepReport:
        """Analyse *optimizer*'s tick history and return a report."""
        history = optimizer.get_history()
        if not history:
            return SleepReport()

        config = optimizer.config
        interval_min = config.stage_check_interval / (config.sample_rate * 60.0)

        # --- basic metrics ---------------------------------------------------
        total_min = len(history) * interval_min
        hypnogram = optimizer.get_hypnogram()

        # sleep onset latency
        sol_min = 0.0
        for tick in history:
            if tick.current_stage != SleepStage.WAKE:
                break
            sol_min += interval_min

        # stage durations
        durations = optimizer.get_stage_durations()
        dur_named: Dict[str, float] = {s.name: v for s, v in durations.items()}

        # stage percentages
        pct_named: Dict[str, float] = {}
        for s, v in durations.items():
            pct_named[s.name] = (v / total_min * 100.0) if total_min > 0 else 0.0

        # target percentages from protocol
        target_named: Dict[str, float] = {
            s.name: v * 100.0 for s, v in optimizer.protocol.stage_targets.items()
        }

        # sleep efficiency
        wake_min = durations.get(SleepStage.WAKE, 0.0)
        sleep_min = total_min - wake_min
        efficiency = (sleep_min / total_min * 100.0) if total_min > 0 else 0.0

        # wakeups after sleep onset
        sleep_started = False
        wakeups = 0
        in_wake = False
        for tick in history:
            if not sleep_started:
                if tick.current_stage != SleepStage.WAKE:
                    sleep_started = True
                    in_wake = False
                continue
            if tick.current_stage == SleepStage.WAKE:
                if not in_wake:
                    wakeups += 1
                    in_wake = True
            else:
                in_wake = False

        reinductions = optimizer._reinduction_count

        # --- quality score (0-100) -------------------------------------------
        # Component 1: stage match (40%)
        match_count = sum(1 for t in history if t.stage_match)
        stage_match_score = (match_count / len(history)) * 100.0 if history else 0.0

        # Component 2: sleep efficiency (25%)
        efficiency_score = min(100.0, efficiency / 0.85 * 100.0)  # 85% = perfect

        # Component 3: sleep onset latency (20%) -- <15 min is ideal
        if sol_min <= 15.0:
            sol_score = 100.0
        elif sol_min <= 30.0:
            sol_score = 100.0 - (sol_min - 15.0) / 15.0 * 50.0
        else:
            sol_score = max(0.0, 50.0 - (sol_min - 30.0) / 30.0 * 50.0)

        # Component 4: wakeups (15%) -- 0 = perfect, >= 5 = 0
        wakeup_score = max(0.0, 100.0 - wakeups * 20.0)

        quality = (
            stage_match_score * 0.40
            + efficiency_score * 0.25
            + sol_score * 0.20
            + wakeup_score * 0.15
        )
        quality = float(np.clip(quality, 0.0, 100.0))

        # --- grade -----------------------------------------------------------
        if quality >= 90:
            grade = "A"
        elif quality >= 75:
            grade = "B"
        elif quality >= 60:
            grade = "C"
        elif quality >= 40:
            grade = "D"
        else:
            grade = "F"

        # --- recommendations -------------------------------------------------
        recs: List[str] = []

        n3_pct = pct_named.get("N3", 0.0)
        n3_target = target_named.get("N3", 0.0)
        if n3_pct < n3_target * 0.7:
            recs.append(
                f"Deep sleep (N3) was {n3_pct:.1f}% vs target {n3_target:.1f}%. "
                "Consider the deep_sleep_boost protocol or earlier bedtime."
            )

        rem_pct = pct_named.get("REM", 0.0)
        rem_target = target_named.get("REM", 0.0)
        if rem_pct < rem_target * 0.7:
            recs.append(
                f"REM sleep was {rem_pct:.1f}% vs target {rem_target:.1f}%. "
                "Try the rem_enhancement protocol or extend sleep duration."
            )

        if sol_min > 20.0:
            recs.append(
                f"Sleep onset took {sol_min:.1f} min. "
                "The insomnia_relief protocol may help reduce latency."
            )

        if wakeups > 2:
            recs.append(
                f"You had {wakeups} awakenings. "
                "Reduce caffeine/alcohol and ensure a dark, cool environment."
            )

        if not recs:
            recs.append("Excellent session! Keep your current routine.")

        # --- assemble --------------------------------------------------------
        return SleepReport(
            total_duration_min=round(total_min, 2),
            sleep_onset_latency_min=round(sol_min, 2),
            sleep_efficiency_pct=round(efficiency, 2),
            quality_score=round(quality, 2),
            stage_durations_min={k: round(v, 2) for k, v in dur_named.items()},
            stage_percentages={k: round(v, 2) for k, v in pct_named.items()},
            stage_targets={k: round(v, 2) for k, v in target_named.items()},
            hypnogram=hypnogram,
            wakeups=wakeups,
            reinductions=reinductions,
            recommendations=recs,
            grade=grade,
        )

generate(optimizer) staticmethod

Analyse optimizer's tick history and return a report.

Source code in src/sc_neurocore/sleep/report_generator.py 
 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
@staticmethod
def generate(optimizer: SleepOptimizer) -> SleepReport:
    """Analyse *optimizer*'s tick history and return a report."""
    history = optimizer.get_history()
    if not history:
        return SleepReport()

    config = optimizer.config
    interval_min = config.stage_check_interval / (config.sample_rate * 60.0)

    # --- basic metrics ---------------------------------------------------
    total_min = len(history) * interval_min
    hypnogram = optimizer.get_hypnogram()

    # sleep onset latency
    sol_min = 0.0
    for tick in history:
        if tick.current_stage != SleepStage.WAKE:
            break
        sol_min += interval_min

    # stage durations
    durations = optimizer.get_stage_durations()
    dur_named: Dict[str, float] = {s.name: v for s, v in durations.items()}

    # stage percentages
    pct_named: Dict[str, float] = {}
    for s, v in durations.items():
        pct_named[s.name] = (v / total_min * 100.0) if total_min > 0 else 0.0

    # target percentages from protocol
    target_named: Dict[str, float] = {
        s.name: v * 100.0 for s, v in optimizer.protocol.stage_targets.items()
    }

    # sleep efficiency
    wake_min = durations.get(SleepStage.WAKE, 0.0)
    sleep_min = total_min - wake_min
    efficiency = (sleep_min / total_min * 100.0) if total_min > 0 else 0.0

    # wakeups after sleep onset
    sleep_started = False
    wakeups = 0
    in_wake = False
    for tick in history:
        if not sleep_started:
            if tick.current_stage != SleepStage.WAKE:
                sleep_started = True
                in_wake = False
            continue
        if tick.current_stage == SleepStage.WAKE:
            if not in_wake:
                wakeups += 1
                in_wake = True
        else:
            in_wake = False

    reinductions = optimizer._reinduction_count

    # --- quality score (0-100) -------------------------------------------
    # Component 1: stage match (40%)
    match_count = sum(1 for t in history if t.stage_match)
    stage_match_score = (match_count / len(history)) * 100.0 if history else 0.0

    # Component 2: sleep efficiency (25%)
    efficiency_score = min(100.0, efficiency / 0.85 * 100.0)  # 85% = perfect

    # Component 3: sleep onset latency (20%) -- <15 min is ideal
    if sol_min <= 15.0:
        sol_score = 100.0
    elif sol_min <= 30.0:
        sol_score = 100.0 - (sol_min - 15.0) / 15.0 * 50.0
    else:
        sol_score = max(0.0, 50.0 - (sol_min - 30.0) / 30.0 * 50.0)

    # Component 4: wakeups (15%) -- 0 = perfect, >= 5 = 0
    wakeup_score = max(0.0, 100.0 - wakeups * 20.0)

    quality = (
        stage_match_score * 0.40
        + efficiency_score * 0.25
        + sol_score * 0.20
        + wakeup_score * 0.15
    )
    quality = float(np.clip(quality, 0.0, 100.0))

    # --- grade -----------------------------------------------------------
    if quality >= 90:
        grade = "A"
    elif quality >= 75:
        grade = "B"
    elif quality >= 60:
        grade = "C"
    elif quality >= 40:
        grade = "D"
    else:
        grade = "F"

    # --- recommendations -------------------------------------------------
    recs: List[str] = []

    n3_pct = pct_named.get("N3", 0.0)
    n3_target = target_named.get("N3", 0.0)
    if n3_pct < n3_target * 0.7:
        recs.append(
            f"Deep sleep (N3) was {n3_pct:.1f}% vs target {n3_target:.1f}%. "
            "Consider the deep_sleep_boost protocol or earlier bedtime."
        )

    rem_pct = pct_named.get("REM", 0.0)
    rem_target = target_named.get("REM", 0.0)
    if rem_pct < rem_target * 0.7:
        recs.append(
            f"REM sleep was {rem_pct:.1f}% vs target {rem_target:.1f}%. "
            "Try the rem_enhancement protocol or extend sleep duration."
        )

    if sol_min > 20.0:
        recs.append(
            f"Sleep onset took {sol_min:.1f} min. "
            "The insomnia_relief protocol may help reduce latency."
        )

    if wakeups > 2:
        recs.append(
            f"You had {wakeups} awakenings. "
            "Reduce caffeine/alcohol and ensure a dark, cool environment."
        )

    if not recs:
        recs.append("Excellent session! Keep your current routine.")

    # --- assemble --------------------------------------------------------
    return SleepReport(
        total_duration_min=round(total_min, 2),
        sleep_onset_latency_min=round(sol_min, 2),
        sleep_efficiency_pct=round(efficiency, 2),
        quality_score=round(quality, 2),
        stage_durations_min={k: round(v, 2) for k, v in dur_named.items()},
        stage_percentages={k: round(v, 2) for k, v in pct_named.items()},
        stage_targets={k: round(v, 2) for k, v in target_named.items()},
        hypnogram=hypnogram,
        wakeups=wakeups,
        reinductions=reinductions,
        recommendations=recs,
        grade=grade,
    )

Sleep Optimizer

sc_neurocore.sleep.sleep_optimizer

SleepOptimizerConfig dataclass

Tuneable knobs for the optimiser loop.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
32
33
34
35
36
37
38
39
@dataclass
class SleepOptimizerConfig:
    """Tuneable knobs for the optimiser loop."""

    sample_rate: int = 256
    fft_window: int = 512
    stage_check_interval: int = 256
    max_reinduction_attempts: int = 3

SleepTick dataclass

Snapshot produced every stage_check_interval samples.

Attributes

tick : int Monotonic tick counter. elapsed_min : float Wall-clock minutes since session start. current_stage : SleepStage Detected stage at this tick. target_stage : SleepStage Protocol's ideal stage at this progress point. stage_match : bool Whether current == target. audio_params : StageAudioParams Audio parameters being delivered. band_powers : Dict[str, float] Most recent EEG band-power decomposition. reinduction_active : bool Whether a re-induction sequence is currently running.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
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
@dataclass
class SleepTick:
    """Snapshot produced every ``stage_check_interval`` samples.

    Attributes
    ----------
    tick : int
        Monotonic tick counter.
    elapsed_min : float
        Wall-clock minutes since session start.
    current_stage : SleepStage
        Detected stage at this tick.
    target_stage : SleepStage
        Protocol's ideal stage at this progress point.
    stage_match : bool
        Whether current == target.
    audio_params : StageAudioParams
        Audio parameters being delivered.
    band_powers : Dict[str, float]
        Most recent EEG band-power decomposition.
    reinduction_active : bool
        Whether a re-induction sequence is currently running.
    """

    tick: int = 0
    elapsed_min: float = 0.0
    current_stage: SleepStage = SleepStage.WAKE
    target_stage: SleepStage = SleepStage.WAKE
    stage_match: bool = True
    audio_params: StageAudioParams = field(default_factory=StageAudioParams)
    band_powers: Dict[str, float] = field(default_factory=dict)
    reinduction_active: bool = False

SleepOptimizer

Closed-loop sleep optimiser.

Parameters

protocol : SleepProtocol or str The protocol instance (or its registry name) to follow. config : SleepOptimizerConfig, optional Operational parameters.

Example::

opt = SleepOptimizer("insomnia_relief")
opt.start_session()
for sample in eeg_stream:
    opt.add_sample(sample)
    tick = opt.check_and_adapt()
    if tick is not None:
        apply_audio(tick.audio_params)
report = opt.stop_session()
Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
 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
class SleepOptimizer:
    """Closed-loop sleep optimiser.

    Parameters
    ----------
    protocol : SleepProtocol or str
        The protocol instance (or its registry name) to follow.
    config : SleepOptimizerConfig, optional
        Operational parameters.

    Example::

        opt = SleepOptimizer("insomnia_relief")
        opt.start_session()
        for sample in eeg_stream:
            opt.add_sample(sample)
            tick = opt.check_and_adapt()
            if tick is not None:
                apply_audio(tick.audio_params)
        report = opt.stop_session()
    """

    def __init__(
        self,
        protocol: SleepProtocol | str,
        config: Optional[SleepOptimizerConfig] = None,
    ) -> None:
        if isinstance(protocol, str):
            protocol = get_protocol(protocol)
        self.protocol: SleepProtocol = protocol
        self.config = config or SleepOptimizerConfig()

        det_cfg = DetectorConfig(
            sample_rate=self.config.sample_rate,
            fft_window=self.config.fft_window,
        )
        self._detector = SleepStageDetector(det_cfg)

        # session state
        self._active: bool = False
        self._sample_count: int = 0
        self._tick_count: int = 0
        self._history: List[SleepTick] = []
        self._reinduction_count: int = 0
        self._reinduction_active: bool = False
        self._consecutive_wake: int = 0

    # -- session lifecycle --------------------------------------------------

    def start_session(self) -> None:
        """Begin a new optimisation session, resetting all state."""
        self._detector.reset()
        self._active = True
        self._sample_count = 0
        self._tick_count = 0
        self._history = []
        self._reinduction_count = 0
        self._reinduction_active = False
        self._consecutive_wake = 0

    def stop_session(self) -> List[SleepTick]:
        """End the current session and return the full tick history."""
        self._active = False
        return list(self._history)

    # -- sample ingestion ---------------------------------------------------

    def add_sample(self, sample: float) -> None:
        """Feed a single EEG voltage sample."""
        if not self._active:
            return
        self._detector.add_sample(sample)
        self._sample_count += 1

    def add_samples(self, samples: np.ndarray[Any, Any]) -> None:
        """Feed an array of EEG voltage samples."""
        if not self._active:
            return
        self._detector.add_samples(samples)
        self._sample_count += len(np.asarray(samples).ravel())

    # -- adaptation ---------------------------------------------------------

    def check_and_adapt(self) -> Optional[SleepTick]:
        """Run stage detection and protocol adaptation.

        Should be called after every ``stage_check_interval`` samples.
        Returns a :class:`SleepTick` when a check is performed, or
        ``None`` if the interval has not elapsed or the session is
        inactive.
        """
        if not self._active:
            return None
        if self._sample_count < (self._tick_count + 1) * self.config.stage_check_interval:
            return None

        self._tick_count += 1

        stage = self._detector.detect()
        if stage is None:
            stage = SleepStage.WAKE

        total_dur_samples = self.protocol.total_duration_min * 60.0 * self.config.sample_rate
        progress = (
            min(1.0, self._sample_count / total_dur_samples) if total_dur_samples > 0 else 0.0
        )
        target = self.protocol.get_target_stage(progress)

        # reinduction logic: detect unwanted awakenings
        if stage == SleepStage.WAKE and target != SleepStage.WAKE:
            self._consecutive_wake += 1
            if (
                self._consecutive_wake >= 2
                and self._reinduction_count < self.config.max_reinduction_attempts
            ):
                self._reinduction_active = True
                self._reinduction_count += 1
        else:
            self._consecutive_wake = 0
            self._reinduction_active = False

        # select audio: during reinduction use N1 params to gently re-induce
        if self._reinduction_active:
            audio = self.protocol.get_audio_for_stage(SleepStage.N1)
        else:
            audio = self.protocol.get_audio_for_stage(stage)

        elapsed_min = self._sample_count / (self.config.sample_rate * 60.0)
        band_powers = self._detector.get_band_powers() or {}

        tick = SleepTick(
            tick=self._tick_count,
            elapsed_min=elapsed_min,
            current_stage=stage,
            target_stage=target,
            stage_match=(stage == target),
            audio_params=audio,
            band_powers=band_powers,
            reinduction_active=self._reinduction_active,
        )
        self._history.append(tick)
        return tick

    # -- query --------------------------------------------------------------

    def get_history(self) -> List[SleepTick]:
        """Return a copy of all recorded ticks."""
        return list(self._history)

    def get_stage_durations(self) -> Dict[SleepStage, float]:
        """Compute time (minutes) spent in each detected stage."""
        interval_min = self.config.stage_check_interval / (self.config.sample_rate * 60.0)
        durations: Dict[SleepStage, float] = {s: 0.0 for s in SleepStage}
        for tick in self._history:
            durations[tick.current_stage] += interval_min
        return durations

    def get_hypnogram(self) -> List[int]:
        """Return the detected-stage sequence as a list of integer codes."""
        return [int(t.current_stage) for t in self._history]

    def get_state(self) -> Dict[str, Any]:
        """Return a summary dict of the optimiser's current state."""
        last = self._history[-1] if self._history else None
        return {
            "active": self._active,
            "tick_count": self._tick_count,
            "sample_count": self._sample_count,
            "elapsed_min": (
                self._sample_count / (self.config.sample_rate * 60.0) if self._active else 0.0
            ),
            "current_stage": last.current_stage.name if last else None,
            "target_stage": last.target_stage.name if last else None,
            "reinduction_count": self._reinduction_count,
            "reinduction_active": self._reinduction_active,
            "protocol": self.protocol.name,
        }

start_session()

Begin a new optimisation session, resetting all state.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
135
136
137
138
139
140
141
142
143
144
def start_session(self) -> None:
    """Begin a new optimisation session, resetting all state."""
    self._detector.reset()
    self._active = True
    self._sample_count = 0
    self._tick_count = 0
    self._history = []
    self._reinduction_count = 0
    self._reinduction_active = False
    self._consecutive_wake = 0

stop_session()

End the current session and return the full tick history.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
146
147
148
149
def stop_session(self) -> List[SleepTick]:
    """End the current session and return the full tick history."""
    self._active = False
    return list(self._history)

add_sample(sample)

Feed a single EEG voltage sample.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
153
154
155
156
157
158
def add_sample(self, sample: float) -> None:
    """Feed a single EEG voltage sample."""
    if not self._active:
        return
    self._detector.add_sample(sample)
    self._sample_count += 1

add_samples(samples)

Feed an array of EEG voltage samples.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
160
161
162
163
164
165
def add_samples(self, samples: np.ndarray[Any, Any]) -> None:
    """Feed an array of EEG voltage samples."""
    if not self._active:
        return
    self._detector.add_samples(samples)
    self._sample_count += len(np.asarray(samples).ravel())

check_and_adapt()

Run stage detection and protocol adaptation.

Should be called after every stage_check_interval samples. Returns a :class:SleepTick when a check is performed, or None if the interval has not elapsed or the session is inactive.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
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
def check_and_adapt(self) -> Optional[SleepTick]:
    """Run stage detection and protocol adaptation.

    Should be called after every ``stage_check_interval`` samples.
    Returns a :class:`SleepTick` when a check is performed, or
    ``None`` if the interval has not elapsed or the session is
    inactive.
    """
    if not self._active:
        return None
    if self._sample_count < (self._tick_count + 1) * self.config.stage_check_interval:
        return None

    self._tick_count += 1

    stage = self._detector.detect()
    if stage is None:
        stage = SleepStage.WAKE

    total_dur_samples = self.protocol.total_duration_min * 60.0 * self.config.sample_rate
    progress = (
        min(1.0, self._sample_count / total_dur_samples) if total_dur_samples > 0 else 0.0
    )
    target = self.protocol.get_target_stage(progress)

    # reinduction logic: detect unwanted awakenings
    if stage == SleepStage.WAKE and target != SleepStage.WAKE:
        self._consecutive_wake += 1
        if (
            self._consecutive_wake >= 2
            and self._reinduction_count < self.config.max_reinduction_attempts
        ):
            self._reinduction_active = True
            self._reinduction_count += 1
    else:
        self._consecutive_wake = 0
        self._reinduction_active = False

    # select audio: during reinduction use N1 params to gently re-induce
    if self._reinduction_active:
        audio = self.protocol.get_audio_for_stage(SleepStage.N1)
    else:
        audio = self.protocol.get_audio_for_stage(stage)

    elapsed_min = self._sample_count / (self.config.sample_rate * 60.0)
    band_powers = self._detector.get_band_powers() or {}

    tick = SleepTick(
        tick=self._tick_count,
        elapsed_min=elapsed_min,
        current_stage=stage,
        target_stage=target,
        stage_match=(stage == target),
        audio_params=audio,
        band_powers=band_powers,
        reinduction_active=self._reinduction_active,
    )
    self._history.append(tick)
    return tick

get_history()

Return a copy of all recorded ticks.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
231
232
233
def get_history(self) -> List[SleepTick]:
    """Return a copy of all recorded ticks."""
    return list(self._history)

get_stage_durations()

Compute time (minutes) spent in each detected stage.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
235
236
237
238
239
240
241
def get_stage_durations(self) -> Dict[SleepStage, float]:
    """Compute time (minutes) spent in each detected stage."""
    interval_min = self.config.stage_check_interval / (self.config.sample_rate * 60.0)
    durations: Dict[SleepStage, float] = {s: 0.0 for s in SleepStage}
    for tick in self._history:
        durations[tick.current_stage] += interval_min
    return durations

get_hypnogram()

Return the detected-stage sequence as a list of integer codes.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
243
244
245
def get_hypnogram(self) -> List[int]:
    """Return the detected-stage sequence as a list of integer codes."""
    return [int(t.current_stage) for t in self._history]

get_state()

Return a summary dict of the optimiser's current state.

Source code in src/sc_neurocore/sleep/sleep_optimizer.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def get_state(self) -> Dict[str, Any]:
    """Return a summary dict of the optimiser's current state."""
    last = self._history[-1] if self._history else None
    return {
        "active": self._active,
        "tick_count": self._tick_count,
        "sample_count": self._sample_count,
        "elapsed_min": (
            self._sample_count / (self.config.sample_rate * 60.0) if self._active else 0.0
        ),
        "current_stage": last.current_stage.name if last else None,
        "target_stage": last.target_stage.name if last else None,
        "reinduction_count": self._reinduction_count,
        "reinduction_active": self._reinduction_active,
        "protocol": self.protocol.name,
    }

Sleep Stage Detector

sc_neurocore.sleep.sleep_stage_detector

SleepStage

Bases: IntEnum

AASM sleep-stage labels.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
32
33
34
35
36
37
38
39
class SleepStage(IntEnum):
    """AASM sleep-stage labels."""

    WAKE = 0
    N1 = 1
    N2 = 2
    N3 = 3
    REM = 4

DetectorConfig dataclass

Parameters for the sleep-stage detector.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
73
74
75
76
77
78
79
80
@dataclass
class DetectorConfig:
    """Parameters for the sleep-stage detector."""

    sample_rate: int = 256
    fft_window: int = 512
    smoothing_window: int = 5
    min_samples: int = 128

SleepStageDetector

Real-time sleep-stage detector from single-channel EEG.

Usage::

det = SleepStageDetector()
for sample in eeg_stream:
    det.add_sample(sample)
    stage = det.detect()
    if stage is not None:
        print(stage.name)
Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
 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
class SleepStageDetector:
    """Real-time sleep-stage detector from single-channel EEG.

    Usage::

        det = SleepStageDetector()
        for sample in eeg_stream:
            det.add_sample(sample)
            stage = det.detect()
            if stage is not None:
                print(stage.name)
    """

    def __init__(self, config: Optional[DetectorConfig] = None) -> None:
        self.config = config or DetectorConfig()
        self._buffer: deque = deque(maxlen=self.config.fft_window)
        self._stage_history: deque = deque(maxlen=self.config.smoothing_window)
        self._band_powers: Optional[Dict[str, float]] = None

    # -- public API ---------------------------------------------------------

    def add_sample(self, sample: float) -> None:
        """Append a single EEG voltage sample to the internal buffer."""
        self._buffer.append(float(sample))

    def add_samples(self, samples: np.ndarray[Any, Any]) -> None:
        """Append an array of EEG voltage samples."""
        for s in np.asarray(samples).ravel():
            self._buffer.append(float(s))

    def detect(self) -> Optional[SleepStage]:
        """Return the smoothed sleep-stage classification, or ``None`` if
        insufficient data has been collected."""
        if len(self._buffer) < self.config.min_samples:
            return None

        powers = self._compute_band_powers()
        self._band_powers = powers

        power_vec = np.array([powers[b] for b in EEG_BANDS])
        raw_stage = self._classify(power_vec)
        self._stage_history.append(raw_stage)

        # temporal smoothing: majority vote over recent detections
        return self._smooth()

    def get_band_powers(self) -> Optional[Dict[str, float]]:
        """Return the most recently computed band-power dict, or ``None``."""
        return self._band_powers

    def reset(self) -> None:
        """Clear all internal state."""
        self._buffer.clear()
        self._stage_history.clear()
        self._band_powers = None

    # -- internals ----------------------------------------------------------

    def _compute_band_powers(self) -> Dict[str, float]:
        """Compute absolute band powers from the current buffer via FFT."""
        data = np.array(self._buffer, dtype=np.float64)
        # Apply Hann window
        window = np.hanning(len(data))
        data = data * window

        fft_vals = np.fft.rfft(data)
        psd = np.abs(fft_vals) ** 2
        freqs = np.fft.rfftfreq(len(data), d=1.0 / self.config.sample_rate)

        powers: Dict[str, float] = {}
        for band_name, (lo, hi) in EEG_BANDS.items():
            mask = (freqs >= lo) & (freqs < hi)
            powers[band_name] = float(psd[mask].mean()) if mask.any() else 0.0

        return powers

    @staticmethod
    def _classify(power_vec: np.ndarray[Any, Any]) -> SleepStage:
        """Classify by cosine similarity to canonical signatures."""
        norm = np.linalg.norm(power_vec)
        if norm < 1e-12:
            return SleepStage.WAKE

        best_stage = SleepStage.WAKE
        best_sim = -1.0
        for stage, sig in STAGE_SIGNATURES.items():
            sim = float(np.dot(power_vec, sig) / (norm * np.linalg.norm(sig)))
            if sim > best_sim:
                best_sim = sim
                best_stage = stage
        return best_stage

    def _smooth(self) -> SleepStage:
        """Majority-vote smoothing over the recent stage history."""
        counter = Counter(self._stage_history)
        return counter.most_common(1)[0][0]

add_sample(sample)

Append a single EEG voltage sample to the internal buffer.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
109
110
111
def add_sample(self, sample: float) -> None:
    """Append a single EEG voltage sample to the internal buffer."""
    self._buffer.append(float(sample))

add_samples(samples)

Append an array of EEG voltage samples.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
113
114
115
116
def add_samples(self, samples: np.ndarray[Any, Any]) -> None:
    """Append an array of EEG voltage samples."""
    for s in np.asarray(samples).ravel():
        self._buffer.append(float(s))

detect()

Return the smoothed sleep-stage classification, or None if insufficient data has been collected.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def detect(self) -> Optional[SleepStage]:
    """Return the smoothed sleep-stage classification, or ``None`` if
    insufficient data has been collected."""
    if len(self._buffer) < self.config.min_samples:
        return None

    powers = self._compute_band_powers()
    self._band_powers = powers

    power_vec = np.array([powers[b] for b in EEG_BANDS])
    raw_stage = self._classify(power_vec)
    self._stage_history.append(raw_stage)

    # temporal smoothing: majority vote over recent detections
    return self._smooth()

get_band_powers()

Return the most recently computed band-power dict, or None.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
134
135
136
def get_band_powers(self) -> Optional[Dict[str, float]]:
    """Return the most recently computed band-power dict, or ``None``."""
    return self._band_powers

reset()

Clear all internal state.

Source code in src/sc_neurocore/sleep/sleep_stage_detector.py 
138
139
140
141
142
def reset(self) -> None:
    """Clear all internal state."""
    self._buffer.clear()
    self._stage_history.clear()
    self._band_powers = None