ICOschema

Backend-agnostic data model for sensor/hardware recordings and their derived computations.

Install

uv venv --allow-existing
uv sync --all-extras

Usage

Note: In the example below we assume you installed Matplotlib:

uv pip install matplotlib

Other codebases should only import Recording/DatasetBundle from the top-level package:

from icoschema import Recording, DatasetBundle

import matplotlib.pyplot as plt

recording = Recording.from_hdf5("test/test_with_sensors.hdf5")
df = recording.to_dataframe()
plt.plot(recording.timestamps, recording.channel1)
plt.show()
recording.to_hdf5("some_file.hdf5")

bundle = DatasetBundle.from_hdf5("some_file.hdf5")  # recording + any DerivedDataset computations
# Note: The code below does not work, since `derived` is not defined
#       Adding the code for `derived` from the example below will not work,
#       since then `my_wavelet_transform` is not defined.
bundle = bundle.with_computation("wavelet_coefficients/channel1/details", derived)
bundle.to_hdf5("some_file.hdf5")

Sources (STU/STH combinations)

A Recording holds one or more independent Sources, one per STU/STH combination, each with its own clock (start_time), packet counter, ADC settings (so its own sampling frequency) and sensors. Start times of different sources are not synchronized. For a single-source recording (everything read from today's files) the usual shortcuts keep working; with several sources they raise a ValueError and you pick a source:

recording.channel1                          # single source only
source = recording.sources["source1"]       # files without a stored name read as "source1"
source.channel1, source.channel_sampling_frequency(), source.signal_loss_percentage
recording.to_dataframe(source="source1")

Example use cases

Inspect a recording without loading a full DataFrame:

recording = Recording.from_hdf5("some_file.hdf5")
print(recording.signal_loss_percentage)
print(recording.summary())                # JSON-encodable dict: per-channel min/max/mean/std, sensor info

Add a derived computation and persist it alongside the recording:

from icoschema.schema.generated.python.dataset import DerivedDataset

bundle = DatasetBundle.from_hdf5("some_file.hdf5")
coeffs = my_wavelet_transform(bundle.recording.channel1)
derived = DerivedDataset(
    key="wavelet_coefficients/channel1/details",
    values=coeffs.flatten().tolist(),
    shape=list(coeffs.shape),
    dtype=str(coeffs.dtype),
    produced_by="wavelet_transform@1.0.0",
)
bundle = bundle.with_computation(derived.key, derived)
bundle.to_hdf5("processed_file.hdf5")

Place samples on the gap-free grid and get the sampling frequency:

recording = Recording.from_hdf5("some_file.hdf5")
recording.sample_index()                 # grid position of every received row (lost packets leave holes)
recording.filled_mask()                  # True where a packet was lost, over the whole grid
recording.channel_sampling_frequency()   # (Hz per channel, "hardware" | "default")

The total sampling frequency is the ADC rate, shared by all channels: 38.4 MHz / ((adc_prescaler + 1) * (adc_acquisition_time + 13) * adc_oversample_rate), from the adc_* settings recorded in the file, or from the default settings (2, 8, 64 → 9523.81 Hz) when the file doesn't record them. The per-channel frequency is that divided by the number of channels.

Store a preprocessed table (lost packets filled, timestamps on the 1 / f_s grid):

from icoschema import PreprocessedRecording

bundle = DatasetBundle.from_hdf5("some_file.hdf5")
preprocessed = PreprocessedRecording(
    timestamp=...,                     # first raw timestamp + round(j * 1e6 / channel_sampling_frequency)
    channel1=...,                      # raw channel1 on the grid, not converted
    channel_sampling_frequency=fs,     # Hz per channel
    sampling_frequency_source="default",
    fill_method="forward_fill",
    producer="my_tool@1.0",
)
bundle = bundle.with_preprocessed(preprocessed)   # raises if it doesn't fit the recording
bundle.to_hdf5("some_file.hdf5")

The preprocessed channels mean exactly what the raw channels mean (same sensors and compensations). The raw recording stays untouched; while a bundle has a preprocessed table, with_recording raises — call without_preprocessed() first.

With several sources, a bundle holds one preprocessed table per source (each with its own grid and channel_sampling_frequency), and bundle.preprocessed — the shortcut for a single source — raises:

bundle = bundle.with_preprocessed(preprocessed, source="source2")   # or set `PreprocessedRecording.source`
bundle.preprocessed_for("source2")                                  # None if that source has none
bundle = bundle.without_preprocessed("source2")                     # no argument: drop all

A step says where it read from with one name, ComputationStep.source_node, from a single namespace:

A step reading several things lists them in inputs. There is no separate "source" on steps or computations: a step's source follows from what it reads. bundle.source_of(name) traces any of the names above back to its source (None for a step over several sources), and bundle.computations_for_source("source2") selects what descends from one source. Step names must not collide with source names or the reserved preprocessed names; whether a source_node names a recorded step is not checked (it may have failed unrecorded). With several sources, run one step per source (e.g. rms_source1, rms_source2) so that computation keys stay unique.

Record the steps that produced the computations:

from icoschema import ComputationStep

bundle = bundle.with_computation("rms/channel1", derived)          # add the step's data first
bundle = bundle.with_step(ComputationStep(
    node="rms",                        # unique, one HDF5 path segment
    step_index=0,                      # assigned by with_step
    producer="rms@1.0.0",
    params='{"window_size": 200}',
    output_keys=["rms/channel1"],      # authoritative link from the step to its data
    status="ok",
))
bundle.computations_of("rms")          # {"rms/channel1": ...}
bundle.unmanaged_keys()                # computations no step lists
bundle = bundle.without_steps(["rms"]) # drops the step and its output_keys; the rest is renumbered

steps is a list ordered by step_index. with_step raises for a duplicate or malformed node name, or an output_keys entry that isn't in computations. How steps are named or how keys relate to node names is up to the producer.

File format

• 1.0: the recording (primary table + sensors) and optional /computations/.
• 1.1: additionally a preprocessed table. Written only when the bundle has one, so files without it stay readable by older versions, which reject 1.1 instead of silently dropping the table.
• 1.2: additionally a /steps/ group (one attribute-only group per step). Written only when the bundle has steps; read back ordered by step_index. A malformed step, or one whose output_keys are missing, is skipped (or raises with on_error="raise").
• 1.3: several sources (one per STU/STH combination). The root attribute sources lists their names in order, and /sources/<name>/ holds each source's table and sensors table the way a 1.0 file holds them at the root. May also have /steps/ and /computations/. Written only when the recording has more than one source: a single source keeps the 1.0-1.2 layout (its name and STU/STH ids are stored as optional table attributes), so those files stay readable by older versions. A preprocessed table, if a source has one, is stored in that source's group.
• Legacy files (no version, table acceleration) are read as before.
• A file with an empty or missing sensors table (old recordings) is read with an unspecified Sensor() (every field None) per channel and no conversion, as long as its conversion attribute is true; if it isn't, there is no scaling to take from the sensor rows and reading raises. Such a recording is written back without a sensors table. Any other mismatch between channel columns and sensor rows still raises.
• The first acquisition tools' files (table acceleration of physical values in g, only Start_Time and Sensor_Range such as ± 100.0 g₀ as metadata) are read too: no conversion attribute (without sensor rows) means converted, a missing adc_reference_voltage is None, and Sensor_Range becomes each channel's sensor unit and range (g, -100..100). A conversion-less file that does have sensor rows raises.
• All datasets with more than one element are gzip-compressed. The pre_metadata/post_metadata attributes of the primary table are preserved.

Development

Regenerate the Python dataclasses after editing the schema:

uv run gen-python icoschema/schema/linkml/dataset.yaml > icoschema/schema/generated/python/dataset.py

Run the tests:

uv run pytest

Schema

classDiagram
    class Recording {
        +Map~string, Source~ sources
    }
    class Source {
        +string name
        +string stu_id
        +string sth_id
        +HardwareMetadata hardware_metadata
        +RecordingMetadata recording_metadata
        +RecordingData recording_data
    }
    class RecordingMetadata {
        +string start_time
        +string pre_metadata
        +string post_metadata
    }
    class RecordingData {
        +int[] timestamp
        +int[] counter
        +float[] channel1
        +float[] channel2
        +float[] channel3
    }
    class HardwareMetadata {
        +string revision
        +string adc_reference_voltage
        +int adc_prescaler
        +int adc_acquisition_time
        +int adc_oversample_rate
        +ChannelMetadata channel1_metadata
        +ChannelMetadata channel2_metadata
        +ChannelMetadata channel3_metadata
    }
    class ChannelMetadata {
        +Sensor sensor
        +Compensation[] compensations
    }
    class Sensor {
        +string sensor_id
        +string sensor_type
        +string name
        +string unit
        +string dimension
        +float phys_min
        +float phys_max
        +float volt_min
        +float volt_max
    }
    class Compensation {
        <<abstract>>
        +int order
    }
    class Conversion {
        <<abstract>>
    }
    class NoConversion
    class LinearConversion {
        +float gain
        +float offset
    }
    class PolynomialConversion {
        +float[] coefficients
    }
    class FrequencyCompensation {
        <<abstract>>
    }
    class DerivedDataset {
        +string key
        +float[] values
        +int[] shape
        +string dtype
        +string unit
        +string produced_by
        +string produced_at
        +string params
        +string description
    }
    class PreprocessedRecording {
        +string source
        +int[] timestamp
        +float[] channel1
        +float[] channel2
        +float[] channel3
        +float channel_sampling_frequency
        +SamplingFrequencySource sampling_frequency_source
        +string fill_method
        +string producer
    }
    class ComputationStep {
        +string node
        +int step_index
        +string producer
        +string params
        +string resolved_params
        +string source_node
        +string[] inputs
        +ComputationStatus status
        +string error
        +string summary
        +string[] output_keys
        +string started_at
        +float duration_s
        +string environment
    }
    class DatasetBundle {
        +Recording recording
        +PreprocessedRecording[] preprocessed_tables
        +Map~string, DerivedDataset~ computations
        +ComputationStep[] steps
    }

    Compensation <|-- Conversion
    Compensation <|-- FrequencyCompensation
    Conversion <|-- NoConversion
    Conversion <|-- LinearConversion
    Conversion <|-- PolynomialConversion

    Recording *-- "1..*" Source
    Source *-- HardwareMetadata
    Source *-- RecordingMetadata
    Source *-- RecordingData
    HardwareMetadata *-- "1" ChannelMetadata : channel1_metadata
    HardwareMetadata o-- "0..1" ChannelMetadata : channel2_metadata
    HardwareMetadata o-- "0..1" ChannelMetadata : channel3_metadata
    ChannelMetadata *-- Sensor
    ChannelMetadata *-- "1..*" Compensation : compensations
    DatasetBundle *-- Recording
    DatasetBundle o-- "0..*" PreprocessedRecording : preprocessed_tables (one per source)
    DatasetBundle o-- "0..*" DerivedDataset : computations
    DatasetBundle o-- "0..*" ComputationStep : steps

Open questions

• All channels of one Source share one timestamp/counter array set, i.e. one sample rate per source. Mixing sensors with different data rates on different channels of the same STU/STH combination isn't representable — would likely need a per-channel rate/timestamp. Different sources may have different rates.

Development

In the text below we assume that you installed:

• just
• uv

and that you want to release version <VERSION> of the package. Please just replace this version number with the version that you want to release (e.g. 0.1.0).

Release

1. Change the version number and commit your changes:

   just release <VERSION>
   

   Note: GitHub Actions will publish a package based on the tagged commit and upload it to PyPi.

2. Create a new release here

   1. Insert the version number <VERSION> into the tag field
   2. For the release title use “Version <VERSION>”
   3. Paste the release notes for the lastest release into the main text field
   4. Click on “Publish Release”

   Note: Alternatively you can also use the gh command:

   gh release create <VERSION>
   

   to create the release notes.