Python API

Python binding for the native JLS implementation.

class pyjls.binding.AnnotationType: The annotation type enumeration.

class pyjls.binding.DataType

The signal data type enumeration.

unsigned integers: U1, U4, U8, U16, U32, U64 signed integer: I4, I8, I16, I32, I64 floating point: F32, F64

class pyjls.binding.Reader

Open a JLS v2 file for reading.

Parameters:: path – The path to the JLS file.

annotations(signal_id, timestamp, cbk_fn)

Read annotations from a signal.

Parameters:

signal_id – The signal id.
timestamp – The starting timestamp. FSR uses sample_id. VSR uses utc.
cbk – The function(timestamp, y, annotation_type, group_id, data) to call for each annotation. Return True to stop iteration over the annotations or False to continue iterating.

close(): Close the JLS file and free all resources.

fsr(signal_id, start_sample_id, length)

Read the FSR data.

Parameters:

signal_id – The signal id.
start_sample_id – The starting sample id to read.
length – The number of samples to read.

Returns:

The data, which varies depending upon the FSR data type.

u1 and u4 data will be packed in little endian order.

For u1, unpack with:: np.unpackbits(y, bitorder=’little’)[:len(x)]
For u4, unpack with: d = np.empty(len(y) * 2, dtype=np.uint8) d[0::2] = np.bitwise_and(y, 0x0f) d[1::2] = np.bitwise_and(np.right_shift(y, 4), 0x0f)

fsr_statistics(signal_id, start_sample_id, increment, length)

Read FSR statistics (mean, stdev, min, max).

Parameters:

signal_id – The signal id for a fixed sampling rate (FSR) signal.
start_sample_id – The starting sample id to read. The sample_id of the first recorded sample in a signal is 0.
increment – The number of samples represented per return value.
length – The number of return values to generate.

Returns:

The 2-D array[summary][stat] of np.float32. * Each summary entry represents the statistics computed

approximately over increment samples starting from start_sample_id + <index> * increment. It has length given by the length argument.

stat is length 4 with columns defined by SummaryFSR which are mean (average), standard deviation, minimum, and maximum.

For length 1, the return statistics are sample-accurate. For larger lengths, the external boundaries for index 0 (first) and index -1 (last) are computed exactly. The internal boundaries are approximated, perfect for waveform display, but perhaps not suitable for other use cases. If you need sample accurate statistics over multiple increments, call this function repeatedly with length 1.

sample_id_to_timestamp(signal_id, sample_id)

Convert sample_id to UTC timestamp for FSR signals.

Parameters:

signal_id – The signal id.
sample_id – The sample_id to convert.

Returns:

The JLS timestamp corresponding to sample_id.

Raises:

RuntimeError – on error.

signal_lookup(spec) → SignalDef

Look up a signal.

Parameters:: spec – The signal id or name.
Returns:: The signal definition:
Raises:: ValueError – If not found.

signals: Return the dict mapping signal_id to SignalDef.

sources: Return the dict mapping source_id to SourceDef.

timestamp_to_sample_id(signal_id, utc_timestamp)

Convert UTC timestamp to sample_id for FSR signals.

Parameters:

signal_id – The signal id.
utc_timestamp – The UTC timestamp to convert.

Returns:

The sample_id corresponding to utc_timestamp.

Raises:

RuntimeError – on error.

user_data(cbk_fn)

Get the user data.

Parameters:: cbk_fn – The callable(chunk_meta_u16, data) called for each user_data entry. Return True to stop iterating over subsequent user data entries or False to continue iterating.

utc(signal_id, sample_id, cbk_fn)

Read the sample_id / utc pairs from a FSR signal.

Parameters:

signal_id – The signal id.
sample_id – The starting sample_id.
cbk – The function(entries) to call for each annotation. Entries is an Nx2 numpy array of [sample_id, utc_timestamp]. Return True to stop iteration over the annotations or False to continue iterating.

class pyjls.binding.SignalDef(signal_id: int, source_id: int, signal_type: int = 0, data_type: int = 0, sample_rate: int = 0, samples_per_data: int = 0, sample_decimate_factor: int = 0, entries_per_summary: int = 0, summary_decimate_factor: int = 0, annotation_decimate_factor: int = 0, utc_decimate_factor: int = 0, sample_id_offset: int = 0, name: str = None, units: str = None, length: int = 0)[source]

Define a signal.

Variables:

signal_id – The source identifier. 0 reserved for global annotations, must be unique per instance.
source_id – The source identifier. Must match a SourceDef entry.
signal_type – The pyjls.SignalType for this signal.
data_type – The pyjls.DataType for this signal.
sample_rate – The sample rate per second (Hz). 0 for VSR.
samples_per_data – The number of samples per data chunk. (write suggestion)
sample_decimate_factor – The number of samples per summary level 1 entry.
entries_per_summary – The number of entries per summary chunk. (write suggestion)
summary_decimate_factor – The number of summaries per summary, level >= 2.
annotation_decimate_factor – The annotation decimate factor for summaries.
utc_decimate_factor – The UTC decimate factor for summaries.
sample_id_offset – The sample id offset for the first sample. (FSR only)
name – The signal name string.
units – The signal units string.
length – The length in samples.

class pyjls.binding.SignalType: The signal type enumeration.

class pyjls.binding.SourceDef(source_id: int, name: str = None, vendor: str = None, model: str = None, version: str = None, serial_number: str = None)[source]

Define a source.

Variables:

source_id – The source identifier.
name – The source name string.
vendor – The vendor string.
model – The model string.
version – The version string.
serial_number – The serial number string.

class pyjls.binding.SummaryFSR: The FSR column enumeration.

class pyjls.binding.Writer

Create a new JLS writer.

Parameters:: path – The output JLS file path.

annotation(signal_id, timestamp, y, annotation_type, group_id, data)

Add an annotation to a signal.

Parameters:

signal_id – The signal id.
timestamp – The x-axis timestamp in sample_id for FSR and UTC for VSR.
y – The y-axis value or NAN to automatically position.
annotation_type – The annotation type.
data – The data for the annotation.

Raise:

On error.

close(): Close the JLS file and release all resources.

flush(): Flush any pending data to the JLS file.

fsr(signal_id, sample_id, data)

Add FSR data to a signal.

Parameters:

signal_id – The signal id for the data.
sample_id – The sample id for the first sample in data.
data – The data to add.

Raise:

On error.

fsr_f32(signal_id, sample_id, data)

Add 32-bit floating-point FSR data to a signal.

Parameters:

signal_id – The signal id for the data.
sample_id – The sample id for the first sample in data.
data – The 32-bit floating point data to add.

signal_def(signal_id, source_id, signal_type=None, data_type=None, sample_rate=None, samples_per_data=None, sample_decimate_factor=None, entries_per_summary=None, summary_decimate_factor=None, annotation_decimate_factor=None, utc_decimate_factor=None, name=None, units=None): Define a signal.

signal_def_from_struct(s: SignalDef): Define a signal.

source_def(source_id, name=None, vendor=None, model=None, version=None, serial_number=None): Define a source.

source_def_from_struct(s: SourceDef): Define a source.

user_data(chunk_meta, data)

Add user data to the file.

Parameters:

chunk_meta – The arbitrary u16 metadata value.
data – The bytes to store.

Raise:

On error.

utc(signal_id, sample_id, utc_i64): Add a mapping from sample_id to UTC timestamp for an FSR signal. :param signal_id: The signal id. :param sample_id: The sample_id for FSR. :param utc: The UTC timestamp. :raise: On error.

pyjls.binding.copy(src, dst, msg_fn=None, progress_fn=None)

Copy a JLS file.

Parameters:

src – The source path.
dst – The destination path.
msg_fn – The optional function to call with status messages.
progress_fn – The optional function to call with progress from 0.0 to 1.0.