Swappable Classes

How SleePyPhases lets you extend processing behaviour without modifying framework code

SleePyPhases defines several swappable classes — base classes that implement built-in processing steps and can be replaced by your own subclass at runtime, on a per-project basis. The mechanism comes from the Swappable mixin in pyPhases:

class Swappable:
    useClass = None

    @classmethod
    def getInstance(cls, *args, **kwargs):
        if cls.useClass is None:
            return cls(*args, **kwargs)
        return cls.useClass(*args, **kwargs)

    @classmethod
    def setClass(cls, subclass):
        cls.useClass = subclass

When the framework needs an instance it calls MyClass.getInstance(...). If your project has called MyClass.setClass(YourSubclass), it gets an instance of YourSubclass instead. Registration might happen in a phase’s prepareConfig method, which runs before any other phase is executed:

# myproject/phases/Init.py
from SleePyPhases import SignalPreprocessing as SP, PreManipulation as PM, DataManipulation as DM

class MySignalPreprocessing(SP):
    # override or add methods here
    pass

class Init(Phase):
    def prepareConfig(self):
        SP.setClass(MySignalPreprocessing) 
        PM.setClass(MyPreManipulation)
        DM.setClass(MyDataManipulation)

Swappable classes in SleePyPhases

`SignalPreprocessing`

Module: SleePyPhases.SignalPreprocessing
Controlled by config key: preprocessing.stepsPerType

Handles per-channel signal preprocessing. Each step is a method on this class that receives (self, signal, recordSignal). Steps are applied per channel type (e.g. eeg, eog, emg) or per named channel (e.g. EEG Fpz-Cz).

from SleePyPhases import SignalPreprocessing as SPPSignalPreprocessing

class SignalPreprocessing(SPPSignalPreprocessing):

    def bandpass(self, signal, recordSignal, low=0.3, high=35.0):
        from scipy.signal import butter, sosfiltfilt
        sos = butter(5, [low, high], btype="band", fs=signal.frequency, output="sos")
        signal.signal = sosfiltfilt(sos, signal.signal)

    def resample(self, signal, recordSignal):
        from scipy.signal import resample as sp_resample
        target = recordSignal.targetFrequency
        n_samples = int(len(signal.signal) / signal.frequency * target)
        signal.signal = sp_resample(signal.signal, n_samples)
        signal.frequency = target

→ Preprocess Signals covers the full step API and config format.

`PreManipulation`

Module: SleePyPhases.PreManipulation
Controlled by config key: preprocessing.manipulationSteps

Operates on the whole RecordSignal (and its event list) after per-channel preprocessing and before the record is written to the cache. Receives (recordSignal, events) and returns the same pair (or raises ChannelsNotPresent to discard the record).

Common uses: removing records that lack required sleep stages, trimming recordings to lights-off windows, merging annotation channels.

from SleePyPhases import PreManipulation as SPPPreManipulation
from pyPhasesRecordloader import ChannelsNotPresent

class PreManipulation(SPPPreManipulation):

    def keep_full_night(self, recordSignal, events):
        """Discard recordings shorter than 6 hours."""
        duration_h = recordSignal.getSignalLength() * recordSignal.targetFrequency / 3600
        if duration_h < 6:
            raise ChannelsNotPresent("recording too short")
        return recordSignal, events

`DataManipulation`

Module: SleePyPhases.DataManipulation
Controlled by config keys: segmentManipulation, batchManipulation, manipulationAfterPredict

Per-segment and per-batch transformation. Steps receive (X, Y, **kwargs) where X has shape (batch, samples, channels) and Y has shape (batch, samples, labels). Steps must mutate X and Y in-place (or re-assign them via return) — the framework picks up whichever is returned.

The same class handles three distinct pipeline stages, configured by three separate YAML keys:

Config key	When called
`segmentManipulation`	Per segment, in the data loader, before batching
`batchManipulation`	Per batch, after collation, before the model
`manipulationAfterPredict`	After inference, before metric computation, where X is model prediction

from SleePyPhases import DataManipulation as SPPDataManipulation
import numpy as np

class DataManipulation(SPPDataManipulation):

    def normalize_ecg(self, X, Y, channel=0):
        """Z-normalize a specific channel."""
        mu = X[:, :, channel].mean(axis=1, keepdims=True)
        sigma = X[:, :, channel].std(axis=1, keepdims=True) + 1e-9
        X[:, :, channel] = (X[:, :, channel] - mu) / sigma

→ Manipulate Data lists all built-in steps and explains return semantics.

`FeatureExtraction`

Module: SleePyPhases.FeatureExtraction
Controlled by config key: preprocessing.featureChannels

Derives additional channels from the raw RecordSignal during the preprocessing step, before data is written to the cache. Each step receives a RecordSignal segment and returns a np.ndarray that is appended as a new channel.

from SleePyPhases import FeatureExtraction as SPPFeatureExtraction
import numpy as np

class FeatureExtraction(SPPFeatureExtraction):

    def log_power(self, segmentSignal):
        """Return log-power of the first EEG channel as a feature channel."""
        eeg = segmentSignal.getSignalByName("EEG").signal
        power = np.convolve(eeg ** 2, np.ones(128) / 128, mode="same")
        return np.log1p(power)

`LabelChannels`

Module: SleePyPhases.LabelChannels
Controlled by config key: labelChannels

Converts the harmonized event-signal dict into per-sample label arrays during the Extract phase. Each method receives (eventSignal, signalLength) and returns a np.ndarray. Built-in channels (SleepStagesAASM, RespEvents, SleepArousals, etc.) are already implemented; subclass to add project-specific ones or override existing behaviour.

from SleePyPhases import LabelChannels as SPPLabelChannels
import numpy as np

class LabelChannels(SPPLabelChannels):

    def SleepBin(self, eventSignal, signalLength):
        """Binary sleep / wake label."""
        from SleepHarmonizer import PSGEventManager
        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

→ Adding Custom Label Channels covers the full API with all built-in channels listed.

`RecordFeatureExtraction`

Module: SleePyPhases.RecordFeatureExtraction

Operates on full recordings to extract tabular features stored in a DataFrame. Used in analysis pipelines, not in the model training path. Steps receive (recordSignal, events, **kwargs) and return a feature dict or pd.Series that is accumulated into the output DataFrame.

from SleePyPhases import RecordFeatureExtraction as SPPRecordFE
import numpy as np

class RecordFeatureExtraction(SPPRecordFE):

    def sleep_efficiency(self, recordSignal, events, channel="SleepStagesAASM"):
        sleep_events = [e for e in events if e.name in ["N1", "N2", "N3", "R"]]
        total_sleep_s = sum(e.duration for e in sleep_events)
        total_s = recordSignal.getDuration()
        return {"sleep_efficiency": total_sleep_s / total_s}

Registering in `Init`

All classes can be registered together in a single phase (e.g. ‘Init’). The wizard automatically generates this boilerplate code, so only add the classes that you want to customize.

# myproject/phases/Init.py
from pyPhases import Phase

from SleePyPhases import (
    SignalPreprocessing as SP,
    PreManipulation as PM,
    DataManipulation as DM,
    FeatureExtraction as FE,
    LabelChannels as LC,
    RecordFeatureExtraction as RFE,
)
from myproject.SignalPreprocessing import SignalPreprocessing
from myproject.PreManipulation import PreManipulation
from myproject.DataManipulation import DataManipulation
from myproject.FeatureExtraction import FeatureExtraction
from myproject.LabelChannels import LabelChannels
from myproject.RecordFeatureExtraction import RecordFeatureExtraction

class Init(Phase):
    def prepareConfig(self):
        SP.setClass(SignalPreprocessing)
        PM.setClass(PreManipulation)
        DM.setClass(DataManipulation)
        FE.setClass(FeatureExtraction)
        LC.setClass(LabelChannels)
        RFE.setClass(RecordFeatureExtraction)

If setClass is never called for a given swappable, the framework uses the built-in base class. You only need to register the classes you extend.

Swappable classes in SleePyPhases

SignalPreprocessing

PreManipulation

DataManipulation

FeatureExtraction

LabelChannels

RecordFeatureExtraction

Registering in Init