New Record Loader Package

A record loader is a Python package that teaches pyPhases how to read signals and metadata from a specific dataset or file format. Use phases create --type recordloader to scaffold one.

Scaffold

phases create --type recordloader -o my-loader

The wizard asks:

Loader name (e.g. MyDataset)
Base format (edf | h5 | mat | wfdb)
File extensions
Record ID pattern (regex)
Default data path

It generates a complete package skeleton with plugin registration, config schema, and a test stub.

Generated structure

my-loader/
  setup.py
  requirements.txt
  .gitignore
  myloader/
    __init__.py
    Plugin.py                      ← registers the loader with pyPhases
    config.yaml                    ← default config values
    recordLoaders/
      __init__.py
      RecordLoaderMyDataset.py     ← your loader logic
  tests/
    test_mydataset.py
  test_run.py ← example code to run you recordloader with the SleepHarmonizer

Implementing the loader

Generated Config:

loader:
  mydataset:
    dataBase: MyDataset
    dataBaseVersion: 1.0.0

    filePath: ./datasets/mydataset
    dataIsFinal: False

    dataset:
      loaderName: RecordLoaderMyDataset
      dataHandler:
        type: allFromFolder
        listFilter: acq
        canReadRemote: True
        basePath: .
        extensions: [.edf, .xml, .erg]
        force: False
        idPattern: .*/(.*).edf
    signal-path: "{recordId}/{recordId}.edf"
    annotation-path: "{recordId}/{recordId}.xml"
    metadata-path: "{recordId}/{recordId}.erg"

    sourceChannels: []

Edit RecordLoaderMyDataset.py. Depending on the base class you chose, implement the required methods. At minimum, you must implement getEventList to return a list of events for a given record ID. You can also override getMetadata and getSignal if your dataset has specific ways to read those. On default the data handler will look for files based on the paths and patterns defined in config.yaml, but you can also override getFilePathAnnotation and getFilePathSignal to customize how file paths are determined. Record IDs are determined by the idPattern regex in the config, which captures a group from the file path. For example, with idPattern: .*/(.*).edf, a file path of /data/mydataset/record123.edf would yield a record ID of record123. Alternatively, you can override getRecordList to return a list of all available record IDs.

There is some additional data required for DICOM-based exports. If you want to generate DICOM files using the Export phase, you must implement getDICOMMetadata to return the required metadata fields.

from pyPhasesRecordloader.recordLoaders.EDFRecordLoader import EDFRecordLoader

class RecordLoaderMyDataset(EDFRecordLoader):

    def getEventList(self, recordId: str, targetFrequency: float = 1) -> List[Event]:
        return super().getEventList(recordId, targetFrequency)
    
    # optional if not implemented in the base class
    def getMetadata(self, recordId: str) -> dict:
        return super().getMetadata(recordId)

    # optional if not implemented in the base class
    def getSignal(self, recordId: str) -> RecordSignal:
        return super().getSignal(recordId)
    
    # optional

    def getFilePathAnnotation(self, recordId: str) -> str:
        # At least one modality DCM must exist
        return super().getFilePathAnnotation(recordId) # default implementation based on annotation-path in dataHandlerConfig
    
    def getFilePathSignal(self, recordId: str) -> str:
        return super().getFilePathSignal(recordId) # default implementation based on signal-path in dataHandlerConfig
    
    def getRecordList(self):
        return super().getRecordList()
    
    def getDICOMMetadata(self, recordId: str) -> dict:
        return {
            "Equipment": {
                "Manufacturer": "",
                "ManufacturerModelName": "",
                "DeviceSerialNumber": "",
                "SoftwareVersions": ""
            }
        }

Plugin registration (`Plugin.py`)

The generated Plugin.py registers the loader so that useLoader: mydataset activates it:

from pyPhases import PluginAdapter


class Plugin(PluginAdapter):
    defaultConfig = "config.yaml"

    def initPlugin(self):
        from pyPhasesRecordloader.Registry import registerRecordLoader
        registerRecordLoader("RecordLoaderBidsDICOM", "pyPhasesRecordloaderBidsDICOM.recordLoaders")
        
        def onConfigChanged(field):
            if field == "mydataset-path":
                path = self.getConfig("mydataset-path")
                self.project.setConfig("loader.mydataset.filePath", path)

        self.project.on("configChanged", onConfigChanged)
        onConfigChanged("mydataset-path")

Config schema (`config.yaml`)

mydataset-path: /data/mydataset

Using the loader in a project

# project.yaml
plugins:
  - pyPhasesRecordloader
  - MyDatasetLoader         # your package name

config:
  useLoader: mydataset
  mydataset-path: /data/mydataset

Using the loader in standalone code:

This will cover all important methods of the record loader, including loading metadata, getting a list of record IDs, and loading signals and events for a specific record. Adjust the paths and record IDs as needed for your dataset. The first ten events are printed, and the first minute of the first signal is plotted using Matplotlib. Make sure to have Matplotlib installed in your environment to run the plotting code.

from phases import DefaultProject
from pyPhasesRecordloader import RecordLoader

project = DefaultProject.create(plugins=[
  "pyPhasesRecordloaderBidsDICOM", 
  "pyPhasesRecordloader", 
  "SleepHarmonizer"
])

project.setConfig("useLoader", "mydataset")
project.setConfig("mydataset-path", "/workspace/data/mydataset")

# execute getRecordList and check if signal and events exist for each record
metadata = project.getData("metadata", dict)

# load all complete recordIds
recordIds = project.getData("allDBRecordIds", list)

# load current record laoder and load the first record
firstGroup = list(recordIds.values())[0]
firstRecord = firstGroup[0]
rl = RecordLoader.get()
signal, events = rl.loadRecord(firstRecord)

# print first 10 events
print(events[:10])

firstSignal = signal.signals[0]
#  plot the first 1 minutes of the first signal
cutoff = min(1*60*firstSignal.frequency, len(firstSignal.signal))
signalName = firstSignal.name
signalArray = firstSignal.signal[:int(cutoff)]

import matplotlib.pyplot as plt
plt.plot(signalArray)
plt.title(signalName)
plt.xlabel("Time (s)")
plt.ylabel("Amplitude")
plt.show()

Base format classes

Format	Base class	Import
EDF	`EDFRecordLoader`	`from pyPhasesRecordloader.recordLoaders.EDFRecordLoader import EDFRecordLoader`
HDF5	`H5RecordLoader`	`from pyPhasesRecordloader.recordLoaders.H5RecordLoader import H5RecordLoader`
MATLAB	`MatRecordLoader`	`from pyPhasesRecordloader.recordLoaders.MatRecordLoader import MatRecordLoader`
WFDB	`WFDBRecordLoader`	`from pyPhasesRecordloader.recordLoaders.WFDBRecordLoader import WFDBRecordLoader`

Install the loader for development

cd my-loader
pip install -e .

Testing

cd my-loader
python -m pytest tests/