Adding Custom Label Channels

A label channel is a per-sample array derived from the harmonized event signal during the Extract phase. The built-in channels (SleepStagesAASM, RespEvents, etc.) are implemented in the LabelChannels class. You can add your own channels — or override existing ones — by subclassing it.

How it works

LabelChannels is a Swappable class. When Extract builds label arrays it calls LabelChannels.getInstance(), so if your project has registered a subclass, that subclass is used instead:

SleePyPhases.LabelChannels          (built-in channels)
        └── yourproject.LabelChannels   (your custom channels)

The dispatcher in LabelChannels.prepare looks for a method whose name matches the label channel name you configure. If no method is found, a clear error is raised.

prepareAll — the entrypoint

prepareAll is the single entrypoint called by Extract for every record. It owns the full events → labels pipeline:

1. Short-circuit — if labelChannels is empty it returns immediately without touching the event list at all. This makes it free to use in self-supervised setups where no labels are needed.
2. Event conversion — calls PSGEventManager.getEventSignalFromList to turn the raw event list into a per-sample signal dict (sleepStage, arousal, apnea, limb).
3. Channel dispatch — calls prepare(name, ...) for each configured channel name, stacks the results into a (signalLength, n_labels) array.

If you need to change how the event list is converted — for example to use a custom event manager, add a pre-processing step, or skip conversion entirely — override prepareAll in your subclass:

class LabelChannels(SPPLabelChannels):

    def prepareAll(self, labelChannelNames, events, signalLength, eventManager, labelFrequency):
        if not labelChannelNames:
            return np.zeros((0, 0))

        # custom pre-processing of events before conversion
        events = [e for e in events if not e.name.startswith("artifact_")]

        return super().prepareAll(labelChannelNames, events, signalLength, eventManager, labelFrequency)

For the common case of just adding or overriding individual channel methods, overriding prepareAll is not necessary — adding a method named after the channel is enough.

Step-by-step

Create the subclass

# myproject/LabelChannels.py
import numpy as np
from SleePyPhases import LabelChannels as SPPLabelChannels
from SleepHarmonizer import PSGEventManager


class LabelChannels(SPPLabelChannels):

    def SleepBin(self, eventSignal: dict, signalLength: int) -> np.ndarray:
        """Binary sleep / wake label derived from the AASM sleep-stage signal."""
        arr = np.zeros(signalLength)
        if "sleepStage" in eventSignal:
            stage = eventSignal["sleepStage"]
            arr[
                (stage == PSGEventManager.INDEX_REM)
                | (stage == PSGEventManager.INDEX_NREM1)
                | (stage == PSGEventManager.INDEX_NREM2)
                | (stage == PSGEventManager.INDEX_NREM3)
            ] = 1
        return arr

Method signature: (self, eventSignal: dict, signalLength: int) -> np.ndarray

Argument	Description
eventSignal	Dict produced by PSGEventManager.getEventSignalFromList — keys: sleepStage, arousal, apnea, limb
signalLength	Number of samples at label frequency

Return a 1-D array of shape (signalLength,). For multi-column outputs return shape (signalLength, n) — the framework will preserve the extra dimension.

Register in your Init phase

# myproject/phases/Init.py
from pyPhases import Phase
from SleePyPhases import LabelChannels as LC
from myproject.LabelChannels import LabelChannels


class Init(Phase):
    def prepareConfig(self):
        LC.setClass(LabelChannels)

Reference in your config

labelChannels:
- SleepStagesAASM   # built-in
- SleepBin          # your custom channel

classification:
    name: sleep
    labelNames: [Sleep, SleepBin]
    classNames:
        - [Wake, R, N1, N2, N3]
        - [Wake, Sleep]

---

Built-in channels

The following channels ship with SleePyPhases.LabelChannels and are always available without any subclassing:

Channel	Source key	Description
SleepStagesAASM	sleepStage	W / R / N1 / N2 / N3 (indices 0–4, unscored = −1)
SleepArousals	arousal	Binary arousal label
SleepArousalRera	arousal	Arousal (1) vs. RERA (2)
SleepApnea	apnea	Four-class respiratory event (obstructive / mixed / central / hypo)
RespEvents	apnea	Like SleepApnea + RERA flow limitation (class 5)
SleepLegMovements	limb	Binary leg movement label
SleepLegMovementsPLM	limb	LM (1) vs. PLM (2)
SleepCombined	both	Four-column array: AASM / Arousals / Apnea / LM combined

---

Overriding a built-in channel

Just define a method with the same name as the built-in:

class LabelChannels(SPPLabelChannels):

    def SleepStagesAASM(self, eventSignal, signalLength):
        """Project-specific: merge N1 into N2."""
        arr = super().SleepStagesAASM(eventSignal, signalLength)
        arr[arr == 2] = 3   # remap N1 → N2
        return arr