lunapi.segsrv

Segment-service API.

Exports:

  • segsrv for efficient windowed signal/annotation access

  • stgcol() and stgn() for sleep-stage plotting helpers

Classes

segsrv

Windowed signal and annotation server for interactive EDF viewing.

Functions

stgcol(ss)

Map a sleep stage label to its canonical hex display colour.

stgn(ss)

Map a sleep stage label to its canonical numeric code.

Module Contents

lunapi.segsrv.stgcol(ss)[source]

Map a sleep stage label to its canonical hex display colour.

Thin proxy to lunapi.viz.stgcol().

Parameters:

ss (str) – Sleep stage label (e.g. 'N1', 'N2', 'N3', 'R', 'W').

Returns:

Hex colour string (e.g. '#0050C8FF').

Return type:

str

lunapi.segsrv.stgn(ss)[source]

Map a sleep stage label to its canonical numeric code.

Thin proxy to lunapi.viz.stgn().

Parameters:

ss (str) – Sleep stage label (e.g. 'N1', 'R', 'W').

Returns:

Numeric sleep stage code used internally (e.g. for hypnogram rendering).

Return type:

int

class lunapi.segsrv.segsrv(p)[source]

Windowed signal and annotation server for interactive EDF viewing.

segsrv wraps the C++ segsrv_t backend, which pre-caches a set of EDF signals and annotations in memory so that interactive viewers (e.g. lunapi.viz.scope()) can read arbitrary windows efficiently without re-querying the EDF on every update.

Typical workflow:

s = segsrv(individual)          # individual is an inst object
s.populate(chs=['EEG', 'EOG'])  # load channels into cache
s.set_epoch_size(30)
s.window(0, 30)                 # view first epoch
signal = s.get_signal('EEG')
Parameters:

p (lunapi.instance.inst) – The inst whose data will be served.

p[source]
segsrv[source]
populate(chs=None, anns=None, max_sr=None)[source]

Load signals and annotations into the segment cache.

Must be called before any windowed data retrieval. Only channels and annotation classes listed here will be available for subsequent queries.

Parameters:
  • chs (str or list of str, optional) – Channel labels to load. Defaults to all channels present in the EDF.

  • anns (str or list of str, optional) – Annotation class names to load. Defaults to all annotation classes present.

  • max_sr (int, optional) – If provided, down-sample any channel whose sample rate exceeds max_sr Hz before caching.

Return type:

None

window(a, b)[source]

Set the active viewing window.

Parameters:
  • a (float) – Window start in seconds (elapsed time from EDF start).

  • b (float) – Window stop in seconds.

Return type:

None

get_signal(ch)[source]

Return raw (possibly filtered) signal values within the current window.

Parameters:

ch (str) – Channel label to retrieve.

Returns:

1-D array of sample values for the current window.

Return type:

numpy.ndarray

get_timetrack(ch)[source]

Return the time axis for a cached channel within the current window.

Parameters:

ch (str) – Channel label.

Returns:

1-D array of time values in seconds corresponding to each sample returned by get_signal().

Return type:

numpy.ndarray

get_time_scale()[source]

Return the effective time-scaling metadata from the backend.

Returns:

Time-scale descriptor returned by the C++ segment server.

Return type:

object

get_gaps()[source]

Return discontinuity (gap) intervals in clock-time seconds.

Returns:

[(start_sec, stop_sec), ...] for each gap in the recording, or an empty list if the EDF is contiguous.

Return type:

list of tuple

apply_filter(ch, sos)[source]

Apply a second-order-sections (SOS) filter to a cached channel.

The filtered signal replaces the raw cache for that channel until clear_filter() is called.

Parameters:
  • ch (str) – Channel label to filter.

  • sos (array-like) – SOS filter coefficients as produced by scipy.signal.butter(..., output='sos').

Return type:

None

clear_filter(ch)[source]

Remove an active filter from a channel, restoring the raw cache.

Parameters:

ch (str) – Channel label whose filter should be cleared.

Return type:

None

clear_filters()[source]

Remove all active channel filters, restoring raw cached signals.

Return type:

None

set_scaling(nchs, nanns=0, yscale=1, ygroup=1, yheader=0.05, yfooter=0.05, scaling_fixed_annot=0.1, clip=True)[source]

Configure vertical layout and scaling for the signal display.

Parameters:
  • nchs (int) – Number of signal channels to lay out.

  • nanns (int, optional) – Number of annotation tracks. Default 0.

  • yscale (float, optional) – Global y-axis scale multiplier. Default 1.

  • ygroup (float, optional) – Fraction of the y-axis height allocated to the signal group. Default 1.

  • yheader (float, optional) – Fraction of the y-axis reserved for the header. Default 0.05.

  • yfooter (float, optional) – Fraction of the y-axis reserved for the footer. Default 0.05.

  • scaling_fixed_annot (float, optional) – Fixed height fraction for each annotation track. Default 0.1.

  • clip (bool, optional) – Whether to clip signal values that exceed the display bounds. Default True.

Return type:

None

get_scaled_signal(ch, n1)[source]

Return y-scaled signal samples for display row n1.

Parameters:
  • ch (str) – Channel label.

  • n1 (int) – 0-based display row index.

Returns:

Scaled sample values mapped into the normalised [0, 1] y-axis coordinate for row n1.

Return type:

numpy.ndarray

get_scaled_y(ch, y)[source]

Transform a physical amplitude value to a normalised display y-coordinate.

Parameters:
  • ch (str) – Channel label.

  • y (float) – Physical amplitude value (in the channel’s native units).

Returns:

Corresponding normalised y-coordinate in [0, 1].

Return type:

float

fix_physical_scale(ch, lwr, upr)[source]

Pin a channel’s display range to fixed physical bounds.

Overrides the auto-derived empirical scale for ch so that the viewer always uses [lwr, upr] regardless of the signal content.

Parameters:
  • ch (str) – Channel label.

  • lwr (float) – Lower bound in the channel’s physical units.

  • upr (float) – Upper bound in the channel’s physical units.

Return type:

None

empirical_physical_scale(ch)[source]

Reset a channel to its backend-derived empirical display bounds.

Parameters:

ch (str) – Channel label.

Returns:

Updated scale descriptor returned by the C++ backend.

Return type:

object

free_physical_scale(ch)[source]

Remove any fixed physical scale bounds for a channel.

After calling this, the display range reverts to auto-scaling based on the current window contents.

Parameters:

ch (str) – Channel label.

Return type:

None

make_sigmod(mod_label, mod_ch, mod_type, sr, order, lwr, upr)[source]

Create a signal modifier using a Butterworth band-pass filter.

Computes SOS coefficients via scipy.signal.butter and registers the modifier with the backend.

Parameters:
  • mod_label (str) – Unique name for this modifier (used to reference it later).

  • mod_ch (str) – Source channel label to apply the filter to.

  • mod_type (str) – Modifier type string passed to the backend (e.g. 'bandpass').

  • sr (float) – Sample rate of mod_ch in Hz (used by the filter design).

  • order (int) – Butterworth filter order.

  • lwr (float) – Lower cutoff frequency in Hz.

  • upr (float) – Upper cutoff frequency in Hz.

Return type:

None

make_sigmod_sos(mod_label, mod_ch, mod_type, sos)[source]

Create a signal modifier from an existing SOS coefficient vector.

Parameters:
  • mod_label (str) – Unique modifier name.

  • mod_ch (str) – Source channel label.

  • mod_type (str) – Modifier type string.

  • sos (array-like) – Pre-computed SOS filter coefficients (flattened to 1-D).

Return type:

None

make_sigmod_raw(mod_label, mod_ch, mod_type)[source]

Create a pass-through (unfiltered) signal modifier slot.

Parameters:
  • mod_label (str) – Unique modifier name.

  • mod_ch (str) – Source channel label.

  • mod_type (str) – Modifier type string (stored for reference; signal is passed through unchanged).

Return type:

None

apply_sigmod(mod_label, mod_ch, slot)[source]

Apply a registered signal modifier to a display slot.

Parameters:
  • mod_label (str) – Name of a modifier previously created with make_sigmod(), make_sigmod_sos(), or make_sigmod_raw().

  • mod_ch (str) – Source channel label.

  • slot (int) – Display slot index to which the modifier output will be assigned.

Return type:

None

get_sigmod_timetrack(bin)[source]

Return the time axis for a signal-modifier output bin.

Parameters:

bin (int) – Bin (output slot) index.

Returns:

Time values in seconds for the modifier output samples.

Return type:

numpy.ndarray

get_sigmod_scaled_signal(bin)[source]

Return scaled samples for a signal-modifier output bin.

Parameters:

bin (int) – Bin (output slot) index.

Returns:

Scaled amplitude values for the current window.

Return type:

numpy.ndarray

set_epoch_size(s)[source]

Set the epoch duration used by the segment server.

Parameters:

s (float) – Epoch length in seconds (e.g. 30).

Return type:

None

get_epoch_size()[source]

Return the current epoch duration in seconds.

Returns:

Epoch length in seconds.

Return type:

float

num_epochs()[source]

Return the total number of epochs at the current epoch size.

Returns:

Number of complete epochs in the recording.

Return type:

int

num_seconds_clocktime()[source]

Return the clock-time duration of the recording in seconds.

Returns:

Total clock-time duration in seconds (after any internal resampling or masking).

Return type:

float

num_seconds_clocktime_original()[source]

Return the original (unmodified) clock-time duration in seconds.

Returns:

Total clock-time duration of the EDF as originally loaded.

Return type:

float

calc_bands(chs)[source]

Pre-compute spectral band summaries for one or more channels.

Results are stored internally and retrieved with get_bands().

Parameters:

chs (str or list of str) – Channel label(s) for which to compute band power.

Return type:

None

calc_hjorths(chs)[source]

Pre-compute Hjorth parameters for one or more channels.

Hjorth parameters (activity, mobility, complexity) are computed per epoch and stored internally. Retrieve with get_hjorths().

Parameters:

chs (str or list of str) – Channel label(s) for which to compute Hjorth parameters.

Return type:

None

get_bands(ch)[source]

Return the pre-computed band-power matrix for a channel.

Requires a prior call to calc_bands() for ch.

Parameters:

ch (str) – Channel label.

Returns:

2-D matrix with one row per epoch and columns for each spectral band.

Return type:

numpy.ndarray

get_hjorths(ch)[source]

Return the pre-computed Hjorth parameter matrix for a channel.

Requires a prior call to calc_hjorths() for ch.

Parameters:

ch (str) – Channel label.

Returns:

2-D matrix with one row per epoch and three columns (activity, mobility, complexity).

Return type:

numpy.ndarray

valid_window()[source]

Return whether the current window overlaps valid (non-gap) data.

Returns:

True if the window is valid for rendering; False otherwise.

Return type:

bool

is_clocktime()[source]

Return whether this instance uses absolute clock-time semantics.

Returns:

True if the EDF uses absolute clock time (real wall-clock start time); False for relative time.

Return type:

bool

get_window_left()[source]

Return the left edge of the current viewing window in seconds.

Returns:

Window start in elapsed seconds from the EDF start.

Return type:

float

get_window_right()[source]

Return the right edge of the current viewing window in seconds.

Returns:

Window stop in elapsed seconds from the EDF start.

Return type:

float

get_window_left_hms()[source]

Return the left edge of the current window as a clock-time string.

Returns:

Window start as 'HH:MM:SS'.

Return type:

str

get_window_right_hms()[source]

Return the right edge of the current window as a clock-time string.

Returns:

Window stop as 'HH:MM:SS'.

Return type:

str

get_clock_ticks(n=6, multiday=False)[source]

Return formatted clock-time tick labels for the current window.

Parameters:
  • n (int, optional) – Maximum number of ticks to return. Default 6.

  • multiday (bool, optional) – If True, include day information in the tick labels. Default False.

Returns:

Up to n (position_sec, label_str) pairs suitable for axis tick placement.

Return type:

list

get_hour_ticks()[source]

Return hour-boundary tick positions for the full recording timeline.

Returns:

(position_sec, label_str) pairs at each whole-hour boundary.

Return type:

list

get_window_phys_range(ch)[source]

Return the physical min/max of a channel within the current window.

Parameters:

ch (str) – Channel label.

Returns:

(min_value, max_value) in the channel’s physical units.

Return type:

tuple

get_ylabel(n)[source]

Return the normalised y-axis anchor position for display row n.

Parameters:

n (int) – 0-based display row index.

Returns:

Normalised y-coordinate in [0, 1] for the centre of row n.

Return type:

float

throttle(n)[source]

Limit the number of output samples returned per window (for rendering).

Parameters:

n (int) – Maximum number of samples to return per channel per window query.

Return type:

None

input_throttle(n)[source]

Set a sample-rate cap applied when populating the cache.

Must be called before populate().

Parameters:

n (int) – Maximum sample rate (Hz) to accept when loading channels. Channels above this rate will be down-sampled.

Return type:

None

summary_threshold_mins(m)[source]

Set the summary aggregation threshold in minutes.

Parameters:

m (int or float) – Threshold in minutes used by the backend’s epoch-level summary aggregation.

Return type:

None

get_annots()[source]

Return the annotation events within the current window.

Returns:

List of (class, start_sec, stop_sec) tuples for all annotation events that overlap the current viewing window.

Return type:

list

get_all_annots(anns, hms=False)[source]

Return all events for the specified annotation classes.

Parameters:
  • anns (str or list of str) – Annotation class name(s) to retrieve.

  • hms (bool, optional) – If True, return times as 'HH:MM:SS' strings rather than floats. Default False.

Returns:

All events for the requested classes across the full recording.

Return type:

list

get_all_annots_with_inst_ids(anns, hms=False)[source]

Return all annotation events with instance IDs and exact timepoints.

Parameters:
  • anns (str or list of str) – Annotation class name(s) to retrieve.

  • hms (bool, optional) – If True, include an 'HH:MM:SS' clock-time column. Default False.

Returns:

Each row contains (hms=False):

[class, meta, start_sec, stop_sec, start_tp, stop_tp, inst_id, ch_str]
   0      1      2          3          4          5       6        7

or (hms=True):

[class, meta, hms, start_sec, duration, start_tp, stop_tp, inst_id, ch_str]
   0      1    2      3          4         5          6       7        8

start_tp and stop_tp are exact uint64_t timepoint values as strings; pass them (as int(row[4]), int(row[5])) to delete_annot() or edit_annot() for lossless round-trips.

Return type:

list of list of str

compile_windowed_annots(anns)[source]

Pre-compile annotation classes into window-ready polygon caches.

Must be called before using get_annots_xaxes() or get_annots_yaxes(). Only the classes listed here will be available for polygon-based rendering.

Parameters:

anns (str or list of str) – Annotation class name(s) to compile.

Return type:

None

set_clip_xaxes(clip)[source]

Enable or disable x-axis clipping when drawing annotation polygons.

Parameters:

clip (bool) – If True, annotation polygons are clipped to the current window boundaries.

Return type:

None

get_annots_xaxes(ann)[source]

Return x-axis polygon coordinates for annotation class ann.

Requires a prior call to compile_windowed_annots().

Parameters:

ann (str) – Annotation class name.

Returns:

X-coordinate arrays for each annotation polygon in the current window.

Return type:

list

get_annots_yaxes(ann)[source]

Return y-axis polygon coordinates for annotation class ann.

Requires a prior call to compile_windowed_annots().

Parameters:

ann (str) – Annotation class name.

Returns:

Y-coordinate arrays for each annotation polygon in the current window.

Return type:

list

set_annot_format6(b)[source]

Toggle 6-point polygon format for annotation rendering.

Parameters:

b (bool) – If True, use the 6-point (trapezoid) format; otherwise use the default rectangle format.

Return type:

None

get_annots_xaxes_ends(ann)[source]

Return end-cap x-coordinates for annotation polygons.

Parameters:

ann (str) – Annotation class name.

Returns:

End-cap x-coordinate arrays for the current window.

Return type:

list

get_annots_yaxes_ends(ann)[source]

Return end-cap y-coordinates for annotation polygons.

Parameters:

ann (str) – Annotation class name.

Returns:

End-cap y-coordinate arrays for the current window.

Return type:

list

delete_annot(aclass, inst_id, start_tp, stop_tp, ch_str='')[source]

Queue deletion of a single annotation instance.

The deletion is not applied until apply_annot_edits() is called.

Parameters:
  • aclass (str) – Annotation class name (col 0 from get_all_annots_with_inst_ids()).

  • inst_id (str) – Instance ID string (col 6).

  • start_tp (int) – Exact start timepoint in tp units (col 4, cast to int).

  • stop_tp (int) – Exact stop timepoint in tp units (col 5, cast to int).

  • ch_str (str, optional) – Channel string (col 7). Default "".

edit_annot(aclass, inst_id, start_tp, stop_tp, ch_str='', new_start=None, new_stop=None, new_ch=None, new_inst_id=None, clear_meta=False, meta=None)[source]

Queue a modification of a single annotation instance.

Any combination of action parameters may be supplied; unset ones leave the original value unchanged. The edit is not applied until apply_annot_edits() is called. aclass is the only immutable field; all others including inst_id can be changed.

Parameters:
  • aclass (str) – Annotation class name (immutable).

  • inst_id (str) – Instance ID string (current value, used to locate the instance).

  • start_tp (int) – Exact start timepoint in tp units (from get_all_annots_with_inst_ids()).

  • stop_tp (int) – Exact stop timepoint in tp units.

  • ch_str (str, optional) – Channel string. Default "".

  • new_start (int, optional) – New start timepoint (tp units). If set without new_stop, the interval is shifted preserving its original duration.

  • new_stop (int, optional) – New stop timepoint (tp units).

  • new_ch (str, optional) – New channel string.

  • new_inst_id (str, optional) – New instance ID.

  • clear_meta (bool, optional) – If True, wipe all existing metadata before applying meta updates. Default False.

  • meta (dict, optional) – {key: value} string pairs to set or override on the instance.

apply_annot_edits(classes=None)[source]

Apply all queued annotation edits and refresh the display cache.

After applying, the queue is cleared automatically.

Parameters:

classes (str or list of str, optional) – Restrict application to these annotation class names. Pass None (default) to apply to all classes that have queued edits.

Returns:

Number of instances changed or deleted.

Return type:

int

clear_annot_edits()[source]

Discard all queued annotation edits without applying them.

Return type:

None

set_psd_mode(on)[source]

Enable or disable on-the-fly PSD computation inside get_scaled_signal().

When enabled, each call to get_scaled_signal() also computes and caches a Welch PSD for that channel at native sample rate.

Parameters:

on (bool) – True to enable PSD computation, False to disable.

get_psd_mode()[source]

Return True if PSD mode is currently enabled.

get_psd_freqs(ch)[source]

Return the cached PSD frequency axis (Hz) for channel ch.

Must call get_scaled_signal() with PSD mode on first.

Parameters:

ch (str) – Channel name.

Returns:

Frequency values in Hz.

Return type:

array-like

get_psd_power(ch)[source]

Return the cached PSD power values for channel ch.

Must call get_scaled_signal() with PSD mode on first.

Parameters:

ch (str) – Channel name.

Returns:

Power spectral density values.

Return type:

array-like