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 as str, bytes, or os.PathLike.

Methods raise ValueError for invalid arguments, such as an undefined signal_id, and RuntimeError for errors reported by the JLS library.

Not thread-safe: call each Reader’s methods from one thread at a time, because all reads share one buffer and file position. Calling methods concurrently may return wrong data or crash the interpreter. To read from multiple threads, open a separate Reader in each thread.

close() raises RuntimeError while a method is in progress, either in another thread or from a callback passed to annotations(), user_data(), or utc(). An exception raised by a callback stops the iteration and propagates from the method. Garbage collection closes a Reader left open.

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 release all resources.

Safe to call repeatedly. Afterwards, the other methods raise ValueError.

Raises:: RuntimeError – If called from a callback or while another thread is inside a method. The Reader remains open.

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.float64.

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.

The standard deviation is the sample standard deviation (ddof=1). Non-finite samples (NaN, inf) are skipped.

For length 1, a window shorter than 25 * sample_decimate_factor samples is computed from the samples, so its statistics are sample-accurate. A longer window computes its partial first and last summary entries from the samples and combines the stored summary entries between them. Its min and max are exact. Its mean and std are accurate to the summary precision: float64 for f64, u32, u64, i32, and i64 data and float32 for all other types, which limits the std of a signal whose mean is large relative to its std (about 2e-5 relative for 15 V with a 4 mV std). A summary entry with some NaN samples counts with its full sample count.

For larger lengths, the external boundaries for index 0 (first) and index -1 (last) are computed as for length 1. The internal boundaries are approximated, perfect for waveform display, but perhaps not suitable for other use cases. For sample-accurate statistics over a long window, read the samples with fsr() and compute them.

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 or iterable of sample_ids to convert.

Returns:

The JLS timestamp corresponding to sample_id, as an int for a scalar sample_id or an np.int64 array otherwise.

Raises:

ValueError – on invalid arguments.
RuntimeError – on error.

signal_lookup(spec) → SignalDef#

Look up a signal.

Parameters:: spec – The signal specification as one of: * signal name (str) * signal id (numeric or str convertable using int()) * ‘<source_name>.<signal_name>’ * [<source_spec>, ‘<signal_name>’]
Returns:: The signal definition.
Raises:: ValueError – If not found.

signals#: Return the dict mapping signal_id to SignalDef.

source_lookup(spec) → SourceDef#

Look up a source.

Parameters:: spec – The source specification as one of: * source name (str). * source id (numeric or str convertable using int())
Returns:: The source definition:
Raises:: ValueError – If not found.

sources#: Return the dict mapping source_id to SourceDef.

time_map_get(signal_id, index=None)#

Get the time map data.

Parameters:

signal_id – The signal id.
index –
The index for the time map which is one of:
- None: return all time maps in the instance.
- int index: Return only this index.
- (int start_index, int end_index): Return this range. start_index is inclusive, end_index is exclusive.

Returns:

Return an np.ndarray with the the named columns [‘sample_id’, ‘timestamp’]. Access as either a[0][‘timestamp’] or a[0][1].

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 or iterable of timestamps to convert.

Returns:

The sample_id corresponding to utc_timestamp, as an int for a scalar utc_timestamp or an np.int64 array otherwise.

Raises:

ValueError – on invalid arguments.
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.TimeMap(reader, signal_id)#: A time map class compatible with pyjoulescope_driver.TimeMap.

class pyjls.binding.Writer#

Create a new JLS writer.

Parameters:: path – The output JLS file path as str, bytes, or os.PathLike.

Methods raise ValueError for invalid arguments, such as an undefined signal_id, and RuntimeError for errors reported by the JLS library.

Thread-safe: multiple threads may call the methods concurrently, such as fsr() for different signals. close() raises RuntimeError while another thread is inside a method, so call close() after the other threads have finished calling this Writer. Garbage collection and interpreter exit close a Writer left open, but call close() or use a with block to finalize the file at a known time.

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.
group_id – The optional group identifier. If unused, set to 0.
data – The data for the annotation.

Raise:

On error.

close()#

Close the JLS file and release all resources.

Safe to call repeatedly. Afterwards, the other methods raise ValueError.

Raises:

RuntimeError – If another thread is inside a method. The Writer remains open.
RuntimeError – If any write failed, such as on a full disk. The Writer is closed, but the file may be incomplete.

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 as a numpy array with the dtype for the signal data_type. u1 and u4 data is packed in little endian order into np.uint8, so each call writes a multiple of 8 (u1) or 2 (u4) samples. This is a known limitation: to store a total sample count that is not such a multiple, pad the final byte, and readers will see the padding samples. The C API, jls_twr_fsr(), accepts any sample count.

Raise:

On error.

A forward skip in sample_id fills the missing samples with NaN for float types and 0 otherwise. Samples that duplicate already written sample ids are dropped, so a late write cannot fill a gap.

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.

fsr_omit_data(signal_id, enable)#

Omit constant level 0 data chunks from an FSR signal.

Parameters:

signal_id – The FSR signal id.
enable – True to omit constant data chunks, False to write all data chunks (default).

Raise:

On error.

See jls_twr_fsr_omit_data() for the data types this affects. On read, omitted data is reconstructed from the summaries.

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 each status message str. Return True to cancel; the copy then raises RuntimeError ABORTED. An exception raised by msg_fn cancels the copy and propagates.
progress_fn – The optional function to call with progress from 0.0 to 1.0. Return True to cancel; the copy then raises RuntimeError ABORTED. An exception raised by progress_fn cancels the copy and propagates.

Raises:

RuntimeError – on error.