ripyl package

Submodules

ripyl.config module

Manage Ripyl configuration data

class ripyl.config.ConfigSettings

Bases: object

Container for general ripyl library settings

__init__()
cython_active

Identify if Cython modules have been monkeypatched

Returns:True if any patches have been applied
cython_status()

Report summary of Cython status

find_patch_obj(obj_path)

Search for a PatchObject by name

Parameters:obj_path (string) – Full path to the object in dotted notation (ex: ‘ripyl.module.object’)
Returns:A patchobject if found or none.
class ripyl.config.PatchObject(py_mname, obj_name, obj, orig_obj)

Bases: object

Represent a monkeypatch object

This class provides a means to track and control pairs of class and function objects that can be swapped at will. This allows simplified monkeypatching for replacing python code with Cython equivalents.

__init__(py_mname, obj_name, obj, orig_obj)
Parameters:
  • py_mname (string) – The name of the module containing the original python object in dotted notation
  • obj_name (string) – The name of the object to be patched
  • obj (function or class object) – Reference to the replacement object patched over the original
  • orig_obj (function or class object) – Reference to the original object
activate()

Apply the monkeypatch over the orignal object

revert()

Restore the original object is it was previously monkeypatched

ripyl.config.write_config(cfg_path, use_cython, cython_prebuild)

Write a file for the Ripyl build configuration

ripyl.decode module

General routines shared between decoders

class ripyl.decode.EdgeSequence(edges, time_step, start_time=None)

Bases: object

Utility class to walk through an edge iterator in arbitrary time steps

__init__(edges, time_step, start_time=None)
Parameters:
  • edges (sequence of (float, int) tuples) – An iterable of 2-tuples representing each edge transition. The 2-tuples must be in the absolute time form (time, logic level).
  • time_step (float) – The default time step for advance() when it is called without an argument.
  • start_time (float) – The initial starting time for the sequence.

:param : :raises: StreamError when there are less than two elements to the edges iterable

advance(time_step=None)

Move forward through edges by a given amount of time.

Parameters:time_step (float) – The amount of time to move forward. If None, the default time_step from the constructor is used.
advance_to_edge()

Advance to the next edge in the iterator after the current time

Returns:The amount of time advanced as a float.
at_end()
Returns:True when the edge iterator has terminated
cur_state()

The logic level of the edge iterator at the current time

class ripyl.decode.MultiEdgeSequence(edge_sets, time_step, start_time=None)

Bases: object

Utility class to walk through a group of edge iterators in arbitrary time steps

__init__(edge_sets, time_step, start_time=None)
Parameters:
  • edge_sets (dict) – A dict of edge sequence iterators keyed by the string name of the channel
  • time_step (float) – The default time step for advance() when it is called without an argument.
  • start_time (float) – The initial starting time for the sequence.
advance(time_step=None)

Move forward through edges by a given amount of time.

Parameters:time_step (float) – The amount of time to move forward. If None, the default time_step from the constructor is used.
advance_to_edge(channel_name=None)

Advance to the next edge among the edge sets or in a named channel after the current time

Parameters:channel_name (string) – If None, the edge sets are advanced to the closest edge after the current time. if a valid channel name is provided the edge sets are advanced to the closest edge on that channel.
Returns:A tuple (time, channel_name) representing the amount of time advanced as a float and the name of the channel containing the edge. If there are no unterminated edge sequences then the tuple (0,0, ‘’) is returned.
Raises:ValueError if channel_name is invalid
at_end(channel_name=None)

Test if the sequences have ended

Parameters:channel_name (string) – The name of the channel to test for termination

:param : :returns: True when the named edge iterator has terminated. if channel_name is

None, returns True when all channels in the set have terminated.
Raises:ValueError if channel_name is invalid
cur_state(channel_name=None)

Get the current state of the edge sets

Parameters:channel_name (string) – Name of the channel to retrieve state from

:param : :returns: The value of the named channel’s state. if channel_name is none

the state of all channels is returned as a list.
Raises:ValueError if channel_name is invalid
cur_time()

Get the current time of the edge sets

ripyl.decode.check_logic_levels(samples, max_samples=20000, buf_size=2000)

Automatically determine the binary logic levels of a digital signal.

This is a wrapper for find_logic_levels() that handles teeing off a buffered sample stream and raising AutoLevelError when detection fails.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream. Each element is a SampleChunk containing an array of samples. This iterator is internally tee’d and becomes invalidated for further use. The return value includes a new sample stream to retrieve samples from.
  • max_samples (int) – The maximum number of samples to consume from the samples iterable. This should be at least 2x buf_size and will be coerced to that value if it is less.

:param : :type buf_size: int :param buf_size:

The maximum size of the sample buffer to analyze for logic levels. This should be less than max_samples.
Returns:A 2-tuple (sample steam, logic_levels) representing the buffered sample stream and a tuple of the detected logic levels of the samples.
Raises:AutoLevelError if less than two peaks are found in the sample histogram.
ripyl.decode.expand_logic_levels(logic_levels, count)

Generate evenly spaced logic levels

Parameters:
  • logic_levels ((float, float)) – A 2-tuple (low, high) representing the min and max logic level to expand on
  • count (int) – The number of logic levels in the result. If the value is less than 3, the result is the same as the sequence passed as logic_levels.
Returns:

A list of logic levels with count length representing each logic level evenly spaced between logic_levels[0] and logic_levels[1].
ripyl.decode.find_bot_top_hist_peaks(raw_samples, bins, use_kde=False, kde_bw=0.05)

Find the bottom and top peaks in a histogram of data sample magnitudes. These are the left-most and right-most of the two largest peaks in the histogram.

Parameters:
  • raw_samples (sequence of numbers) – A sequence representing the population of data samples that will be analyzed for peaks
  • bins (int) – The number of bins to use for the histogram
  • use_kde (bool) – Boolean indicating whether to construct the histogram from a Kernel Density Estimate. This is useful for approximating normally distributed peaks on synthetic data sets lacking noise.

:param : :type kde_bw: float :param kde_bw:

Float providing the bandwidth parameter for the KDE

:param : :returns: A 2-tuple (bot, top) representing the bottom and top peaks. the value for

each peak is the center of the histogram bin that represents the midpoint of the population for that peak.
Returns:None if less than two peaks are found in the histogram
Raises:ValueError if a KDE cannot be constructed
ripyl.decode.find_edges(samples, logic, hysteresis=0.4)

Find the edges in a sampled digital waveform

This is a generator function that can be used in a pipeline of waveform procesing operations.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream. Each element is a SampleChunk containing an array of samples.
  • logic ((float, float)) – A 2-tuple (low, high) representing the mean logic levels in the sampled waveform

:param : :type hysteresis: float :param hysteresis:

A value between 0.0 and 1.0 representing the amount of hysteresis the use for detecting valid edge crossings.

:param : :returns: An iterator yielding a series of 2-tuples (time, value) representing the time and

logic value (0 or 1) for each edge transition. The first tuple yielded is the initial state of the sampled waveform. All remaining tuples are detected edges.
Raises:StreamError if the stream is empty
ripyl.decode.find_hist_peaks(hist, thresh_scale=1.0)

Find all peaks in a histogram This uses a modification of the method employed by the “peaks” function in LeCroy digital oscilloscopes. The original algorithm is described in various manuals such as the 9300 WP03 manual or WavePro manual RevC 2002 (p16-14).

This algorithm works well for real world data sets where the histogram peaks are normally distributed (i.e. there is some noise present in the data set). For synthetic waveforms lacking noise or any intermediate samples between discrete logic levels, the statistical measures used to determine the threshold for a peak are not valid. The threshold t2 ends up being too large and valid peaks may be excluded. To avoid this problem the histogram can be sampled from a KDE instead or the thresh_scale parameter can be set to a lower value.

Parameters:
  • hist (sequence of int) – A sequence representing the histogram bin counts. Typically the first parameter returned by numpy.histogram() or a KDE from scipy.stats.gaussian_kde().
  • thresh_scale (float) – Apply a scale factor to the internal threshold for peak classification.

:param : :returns: A list of peaks where each peak is a 2-tuple representing the

start and end indices of the peak in hist.
ripyl.decode.find_logic_levels(samples, max_samples=20000, buf_size=2000)

Automatically determine the binary logic levels of a digital signal.

This function consumes up to max_samples from samples in an attempt to build a buffer containing a representative set of samples at high and low logic levels. Less than max_samples may be consumed if an edge is found and the remaining half of the buffer is filled before the max_samples threshold is reached.

Warning: this function is insensitive to any edge transition that occurs within the first 100 samples. If the distribution of samples is heavily skewed toward one level over the other None may be returned. To be reliable, a set of samples should contain more than one edge or a solitary edge after the 400th sample.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream. Each element is a SampleChunk containing an array of samples.
  • max_samples (int) – The maximum number of samples to consume from the samples iterable. This should be at least 2x buf_size and will be coerced to that value if it is less.

:param : :type buf_size: int :param buf_size:

The maximum size of the sample buffer to analyze for logic levels. This should be less than max_samples.

:param : :returns: A 2-tuple (low, high) representing the logic levels of the samples :returns: None if less than two peaks are found in the sample histogram.

ripyl.decode.find_multi_edges(samples, hyst_thresholds)

Find the multi-level edges in a sampled digital waveform

This is a generator function that can be used in a pipeline of waveform procesing operations.

Note that the output of this function cannot be used directly without further processing. Transitions across multiple states cannot be easily distinguished from transitions incliding intermediate states. For the case of three states (-1, 0, 1), Short periods in the 0 state should be removed but this requires knowledge of the minimum time for a 0 state to be valid. This is performed by the remove_transitional_states() function.

The logic state encoding is formulated to balance the number of positive and negative states around 0 for odd numbers of states and with one extra positive state for even state numbers. For 2 states the encoding is the usual (0,1). For 3: (-1, 0, 1). For 4: (-1, 0, 1, 2). For 5: (-2, -1, 0, 1, 2), etc.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream. Each element is a SampleChunk containing an array of samples.
  • hyst_thresholds (sequence of float) – A sequence containing the hysteresis thresholds for the logic states. For N states there should be (N-1) * 2 thresholds. The gen_hyst_thresholds() function can compute these values from more usual logic parameters. The numbers must be sorted in ascending order. Every pair of numbers in the sequence forms the bounds of a hysteresis band. Samples within these bands are considered transient states. Samples outside these bands are the valid logic states.
Returns:

An iterator yielding a series of 2-tuples (time, int) representing the time and logic value for each edge transition. The first tuple yielded is the initial state of the sampled waveform. All remaining tuples are detected edges.
Raises:

StreamError if the stream is empty
ripyl.decode.find_symbol_rate(edges, sample_rate=1.0, spectra=2, auto_span_limit=True, max_span_limit=None)

Determine the base symbol rate from a set of edges

This function depends on the edge data containing a variety of spans between edges all related to the fundamental symbol rate. The Harmonic Product Spectrum (HPS) of the edge span values is calculated and used to isolate the fundamental symbol rate. This function will not work properly on a clock signal containing a single time span between edges due to the lack of higher fundementals needed by the HPS unless spectra=1 which effectively disables the HPS operation.

Parameters:edges ([(float, int)..] or [(int, int)..]) –

An iterable of 2-tuples representing each edge transition. The tuples are in one of two forms:

  • absolute time (time, logic level)
  • sample indexed (index, logic level)

This function will consume all elements of the edges iterable. It must have a finite length

:param : :type sample_rate: float :param sample_rate:

An adjustment to convert the raw symbol rate from samples to time. If the edges parameter is based on absolute time units then this should remain the default value of 1.0.

:param : :type spectra: int :param spectra:

The number of spectra to include in the calculation of the HPS. This number should not larger than the highest harmonic in the edge span data.

:param : :type auto_span_limit: bool :param auto_span_limit:

Excessively long edge spans can impair the symbol rate detection by reducing the resolution of the HPS. They are typically the result of long idle periods between the interesting parts we want to estimate the symbol rate from. When this parameter is True, an attempt is made to find the ideal limit for the spans included in the HPS.

:param : :type max_span_limit: int :param max_span_limit:

An optional upper limit for span length to include in the HPS. auto_span_limit must be False for this to take effect.
Returns:The estimated symbol rate of the edge data set as an int
Raises:ValueError if there are not enough edge spans to evaluate a HPS.
ripyl.decode.gen_histogram(raw_samples, bins, use_kde=False, kde_bw=0.05)

Generate a histogram using either normal binning or a KDE

Parameters:
  • raw_samples (sequence of numbers) – A sequence representing the population of data samples that will be analyzed for peaks
  • bins (int) – The number of bins to use for the histogram
  • use_kde (bool) – Boolean indicating whether to construct the histogram from a Kernel Density Estimate. This is useful for approximating normally distributed peaks on synthetic data sets lacking noise.

:param : :type kde_bw: float :param kde_bw:

Float providing the bandwidth parameter for the KDE
Returns:A tuple (hist, bin_centers) containing lists of the histogram bins and the center value of each bin.
Raises:ValueError if a KDE cannot be constructed
ripyl.decode.gen_hyst_thresholds(logic_levels, expand=None, hysteresis=0.1)

Generate hysteresis thresholds for find_multi_edges()

This function computes the hysteresis thresholds for multi-level edge finding with find_multi_edges().

Parameters:
  • logic_levels (sequence of float) – A sequence of the nominal voltage levels for each logic state sorted in ascending order or the (low, high) pair when expansion is used.
  • expand (int or None) – When not None, the number of logic levels to expand the provided logic_levels into.
  • hysteresis (float) – A value between 0.0 and 1.0 representing the amount of hysteresis the use for detecting valid edge crossings.
Returns:

A list of floats. every pair of numbers represents a hysteresis band.
ripyl.decode.remove_transitional_states(edges, min_state_period)

Filter out brief transitional states from an edge stream

This is a generator function that can be used in a pipeline of waveform procesing operations.

Parameters:
  • edges (iterable of (float, int) tuples) – An iterable of 2-tuples representing each edge transition. The 2-tuples must be in the absolute time form (time, logic level).
  • min_state_period (float) – The threshold for transitional states. A transition lasting less than this threshold will be filtered out of the edge stream.
Returns:

An iterator yielding a series of 2-tuples (time, value) representing the time and logic value for each edge transition. The first tuple yielded is the initial state of the sampled waveform. All remaining tuples are detected edges.
Raises:

StreamError if the stream is empty

ripyl.manchester module

Manchester encoding utilities

class ripyl.manchester.ManchesterStates

Bases: ripyl.util.enum.Enum

Enumeration of manchester states

High = 3
Idle = 4
Low = 2
One = 1
Zero = 0
ripyl.manchester.diff_encode(bits)

Convert from unipolar Manchester states (0, Idle, 1) to differential (-1, 0, 1) encoding

Parameters:bits (iterable of (float, int)) – An edge stream to convert to differential form.
Returns:An iterator yielding an edge stream of differentially coded states.
ripyl.manchester.manchester_decode(edges, bit_period, falling=0, combine_bits=False)

Convert differential coded edges to manchester states

Parameters:
  • edges (iterable of (float, int)) – An edge stream of differential states (-1, 0, 1) representing Manchester coded bits.
  • bit_period (float) – The period of a single bit.
  • falling (int) – The bit encoding for a falling edge. Either 0 or 1.
  • combine_bits (bool) – When True, the output states will only be yielded when a different state occurs. When False, a state is yielded for each bit or half-bit.
Returns:

An iterator yielding a stream of Manchester states in (float, int) form.
ripyl.manchester.manchester_encode(states, bit_period, falling=0, idle_start=0.0, idle_end=0.0)

Convert Manchester states to a stream of unipolar (0, 1, Idle) states

Parameters:
  • states (iterable of ManchesterState) – A series of ManchesterStates to encode.
  • bit_period (float) – The period of a single bit.
  • falling (int) – The bit encoding for a falling edge. Either 0 or 1.
  • idle_start (float) – The amount of idle time before the transmission of states begins.
  • idle_end (float) – The amount of idle time after the last state.
Returns:

An iterator yielding an edge stream of encoded states including Idle.

ripyl.sigproc module

General routines for signal processing of streaming waveforms

ripyl.sigproc.amplify(samples, gain=1.0, offset=0.0)

Apply gain and offset to a sample stream

This modifies samples such that output = input * gain + offset.

This is a generator function.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream to modify.
  • gain (float) – The gain multiplier for the samples
  • offset (float) – The additive offset for the samples
Returns:

An iterator yielding a sample stream.
ripyl.sigproc.approximate_bandwidth(rise_time)

Determine an approximate bandwidth for a signal with a known rise time

Parameters:rise_time (float) – A rise time to compute bandwidth from
Returns:The bandwidth for the rise time.
ripyl.sigproc.capacify(samples, capacitance, resistance=1.0, iterations=80)

Simulate an RC filter on a waveform:

: samples >--R--+--> out
:               |
:               C
:              _|_

Warning: This function becomes unstable for small time constants (C * R). It is implemented with a simple application of the RC difference equations. The time step (dt) is taken from the sample period of each sample chunk divided by the number of iterations. The results will be inaccurate for large values of dt but still largely representative of the expected behavior.

This is a generator function.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream to modify.
  • capacitance (float) – The capacitance value. If resistance is kept at the default value of 1.0 Ohms this parameter is equivalent to the time constant tau (c = tau / r).
  • resistance (float) – The resistance value
  • iterations (int) – The number of iterations to calculate each sample. You can experience numeric instability if this value is too low.
Returns:

An iterator yielding a sample stream.
ripyl.sigproc.chain(stream_gap_time, *streams)

Combine a sequence of sample streams together.

A set of sample streams are concatenated together with updated time stamps to maintain monotonically increasing time.

Parameters:
  • stream_gap_time (float) – The time interval added between successive streams
  • streams (sequence of iterables containing SampleChunk objects) – A sequence of streams
Returns:

An iterator yielding a sample stream representing the data from each stream in order
ripyl.sigproc.chain_edges(stream_gap_time, *streams)

Combine a sequence of edge streams together.

A set of edge streams are concatenated together with updated time stamps to maintain monotonically increasing time.

Parameters:
  • stream_gap_time (float) – The time interval added between successive streams
  • streams (sequence of sequences containing (float, number) tuples) – A sequence of streams
Returns:

An iterator yielding an edge stream representing the data from each stream in order
ripyl.sigproc.dropout(samples, start_time, end_time, val=0.0)

Force a sample stream to a fixed level

This simulates the effect of a dropout in data transmission by repacing samples with a fixed value.

This is a generator function.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream to modify.
  • start_time (float) – Start of the dropout phase
  • end_time (float) – End of the dropout phase
  • val (float) – The sample value to substitute during the dropout phase
Returns:

An iterator yielding a sample stream.
ripyl.sigproc.edges_to_sample_stream(edges, sample_period, logic_states=(0, 1), end_extension=None, chunk_size=10000)

Convert an edge stream to a sample stream

The output samples are scaled to the range of 0.0 to 1.0 regardless of the number of logic states.

Parameters:edges (iterable of (float, int) tuples) – An edge stream to sample

:param : :type sample_period: float :param sample_period:

The sample period for converting the edge stream
Parameters:logic_states (sequence of int) – The coded state values for the lowest and highest state in the edge stream. For 2-level states these will be (0,1). For 3-level: (-1, 1). For 5-level: (-2, 2).

:param : :type end_extension: float :param end_extension:

Optional amount of time to add to the end after the last edge transition
Parameters:chunk_size (int) – Number of samples in each SampleChunk
Returns:An iterator yielding a stream of SampleChunk objects.
ripyl.sigproc.filter_waveform(samples, sample_rate, rise_time, ripple_db=60.0, chunk_size=10000)

Apply a bandwidth limiting low-pass filter to a sample stream

This is a generator function.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream to be filtered.
  • sample_rate (float) – The sample rate for converting the sample stream.
  • rise_time (float) – Rise (and fall) time for the filtered samples.
  • ripple_db (float) – Noise suppression in dB for the bandwidth filter stop band. This should be a positive value.

:param : :type chunk_size: int :param chunk_size:

Internal FIR filter sample pool size. This can generally be ignored. To support streaming of samples, the FIR filter operation is done piecewise so we don’t have to consume the entire input before producing filtered output. Larger values will reduce the number of filter operations performed. Excessively small values will waste time due to the reprocessing of overlapping samples between successive pools.
Returns:An iterator yielding a stream of SampleChunk objects.
ripyl.sigproc.invert(stream)

Invert the polarity of stream values

This is a generator function.

Parameters:samples (iterable of SampleChunk objects) – An iterable sample stream to modify.
Returns:An iterator yielding a sample stream.
ripyl.sigproc.min_rise_time(sample_rate)

Compute the minimum rise time for a sample rate

This function is useful to determine the minimum rise time acceptable as parameters to filter_waveform() and synth_wave(). You should use a scale factor to incease the rise time at least slightly (e.g. rt * 1.01) to avoid raising a ValueError in those functions due to floating point inaccuracies.

Parameters:sample_rate (number) – The sample rate to determine rise time from
Returns:A float for the minimum acceptable rise time.
ripyl.sigproc.noisify(samples, snr_db=30.0)

Add noise to a sample stream

This modifies samples with additive, normally distributed noise

This is a generator function.

Parameters:
  • samples (iterable of SampleChunk objects) – An iterable sample stream to modify.
  • snr_db (float) – The Signal to Noise Ratio in dB. This value is only accurate if the input samples are normalized to the range 0.0 to 1.0. Any amplification should be applied after noisify() for the SNR to be correct.
Returns:

An iterator yielding a sample stream.
ripyl.sigproc.quantize(samples, full_scale, bits=8)

Quantize samples to simulate oscilloscope quantization error

This should be applied to a noisy signal to have notable results.

Parameters:samples (iterable of SampleChunk objects) – An iterable sample stream to modify.

:param : :type full_scale: float :param full_scale:

The full scale range for digitizer being emulated. For example, a scope with 8 vertical divisions set at 2V/div. will have a full scale range of 2V*8 = 16V
Parameters:bits (int) – The number of bits to quantize to
Returns:An iterator yielding a sample stream.
ripyl.sigproc.remove_excess_edges(edges)

Remove incorrect edge transitions from an edge stream

This is a generator function.

This function is most useful for conditioning the outputs of the protocol synthesizers. For those protocols with multiple signals, the synthesizers myst yield a new output set for all signals when any one of them changes. This results in non-conforming edge streams that contain multiple consecutive pairs with a non-changing value.

Parameters:edges (iterable of (float, int) tuples) – An edge stream to filter for extraneous non-edges

:param : :returns: An iterator yielding an edge stream.

ripyl.sigproc.sum_streams(stream1, stream2)

Add two sample streams together

The time elements of each stream will not be aligned if they do not match. Instead the time values from stream1 are used for the result. The iterator terminates when either of the two streams ends.

This is a generator function.

Parameters:
  • stream1 (iterable of SampleChunk objects) –
  • stream2 (iterable of SampleChunk objects) – The two sample streams to have their corresponding values added together.

:param : :returns: An iterator yielding a sample stream.

ripyl.sigproc.synth_wave(edges, sample_rate, rise_time, tau_factor=0.0, logic_states=(0, 1), ripple_db=60.0, chunk_size=10000)

Convert an edge stream to a sampled waveform with band limited rise/fall times

This is a convenience function combining edges_to_sample_stream(), filter_waveform(), and (optionally) capacify().

Parameters:
  • edges (sequence of (float, int) tuples) – An edge stream to be sampled
  • sample_rate (float) – The sample rate for converting the edge stream
  • rise_time (float) – Rise (and fall) time for the filtered samples
  • tau_factor (float) – The scale factor used to derive a capacify() time constant from the rise_time such that tau = rise_time * tau_factor. The capacify operation is skipped if tau_factor is < 0.01.
  • logic_states (sequence of int) – The coded state values for the lowest and highest state in the edge stream. For 2-level states these will be (0,1). For 3-level: (-1, 1). For 5-level: (-2, 2).
  • ripple_db (float) – Noise suppression in dB for the bandwidth filter stop band. This should be a positive value.
  • chunk_size (int) – Number of samples in each SampleChunk
Returns:

An iterator for the synthesized sample stream

ripyl.streaming module

Data streaming common classes

class ripyl.streaming.AnnotationFormat

Bases: ripyl.util.enum.Enum

Enumeration of annotation data formats

Bin = 7
Enum = 8
General = 2
Hex = 6
Hidden = 0
Int = 5
Invisible = 1
Small = 9
String = 3
Text = 4
exception ripyl.streaming.AutoLevelError(msg='Unable to find avg. logic levels of waveform')

Bases: ripyl.streaming.StreamError

Custom exception class auto-level errors

__init__(msg='Unable to find avg. logic levels of waveform')
class ripyl.streaming.ChunkExtractor(stream)

Bases: object

Utility class that pulls arbitrarily sized chunks from a sample stream

__init__(stream)
buffered_chunk()

Get all remaining buffered samples

next_chunk(chunk_size=10000)

Get a new chunk of samples from the stream

Parameters:chunk_size (int) – The number of samples for this chunk.
Returns:A samplechunk object. if the stream had fewer than chunk_size samples remaining then the SampleChunk.samples array is sized to hold only those samples.
Returns:None if the stream has ended.
next_samples(sample_count=10000)

Get a new set of raw samples from the stream

Parameters:sample_count (int) – The number of samples for the array.
Returns:A numpy array of float. if the stream had fewer than sample_count samples remaining then the array is sized to hold only those samples.
Returns:None if the stream has ended.
class ripyl.streaming.SampleChunk(samples, start_time, sample_period)

Bases: object

‘Sample stream object

This represents a “chunk” of samples contained in a numpy array stored in the samples attribute.

__init__(samples, start_time, sample_period)
exception ripyl.streaming.StreamError

Bases: exceptions.RuntimeError

Custom exception class for edge and sample streams

class ripyl.streaming.StreamEvent(time, data=None, kind='unknown event', status=0)

Bases: ripyl.streaming.StreamRecord

A stream element that occurs at a specific point in time

__init__(time, data=None, kind='unknown event', status=0)
summary(a=None, depth=0)

Yield string(s) summarizing this segment and all of its subrecords :type a: StreamRecord or None :param a:

StreamRecord to produce summary from. Uses self if None.
Parameters:depth (int) – Indentation level for this summary
class ripyl.streaming.StreamRecord(kind='unknown', status=0)

Bases: object

Base class for protocol decoder output stream objects

Variables:
  • kind – A string identifying the kind of record
  • status – An integer status code
  • stream_id – A unique numeric stream identifier
  • subrecords – A list of child StreamRecord objects

:param :

__init__(kind='unknown', status=0)
annotate(style=None, fields=None, data_format=2)

‘Set annotation attributes

Parameters:style (string or None) – The name of a style to use for drawing a rectangle representing this record

:type : as defined in ripyl.util.plot.annotation_styles :param :

Parameters:
  • fields (dict of string:value) – A set of arbitrary info fields that may be displayed as attributes of this record. The special field ‘_bits’ identifies the number of data bits in this record The special field ‘_enum’ identifies an enumeration type for this record’s data attribute The special field ‘value’ is a string that will override any other source of label text
  • data_format (AnnotationFormat) – The format for the text label
nested_status()
Returns:The highest status value from this record and its subrecords
classmethod status_text(status)
Returns:The string representation of a status code
text(default_format=3)

Generate a string representation of this segment’s data

Parameters:default_format (AnnotationFormat) – Set the format to use when the data_format attribute is General
class ripyl.streaming.StreamSegment(time_bounds, data=None, kind='unknown segment', status=0)

Bases: ripyl.streaming.StreamRecord

A stream element that spans two points in time

__init__(time_bounds, data=None, kind='unknown segment', status=0)
end_time
start_time
summary(a=None, depth=0)

Yield string(s) summarizing this segment and all of its subrecords :type a: StreamRecord or None :param a:

StreamRecord to produce summary from. Uses self if None.
Parameters:depth (int) – Indentation level for this summary
class ripyl.streaming.StreamStatus

Bases: ripyl.util.enum.Enum

Enumeration for standard stream status codes

Error = 200
Ok = 0
Warning = 100
class ripyl.streaming.StreamType

Bases: ripyl.util.enum.Enum

Enumeration for stream types

Edges = 0
Samples = 1
ripyl.streaming.extract_all_samples(samples)

Get all samples from a sample stream along with parameter information.

Parameters:samples (iterable of SampleChunk objects) – The sample stream to extract samples from.
Returns:A tuple containing a numpy sample array of float, the start time, and the sample period.
ripyl.streaming.extract_samples(samples, sample_count=10000)

Create a new gerator that yields sample arrays of the desired size

This is a generator function. Its send() method can be used to change the value of sample_count mid-stream.

Parameters:
  • samples (iterable of SampleChunk objects) – The sample stream to extract samples from.
  • sample_count (int) – The number of samples for the arrays.
Returns:

An iterator yielding a series of numpy arrays. If the stream has fewer than sample_count samples remaining then the array is sized to hold only those samples.
ripyl.streaming.load_stream(fh)

Restore a stream of StreamRecord objects from a file

Parameters:fh (file-like object or a string) – File to load records from. If a file handle is passed it should have been opened in ‘rb’ mode. If a string is passed it is the name of a file to read from.

:param : :returns: A list of streamrecord objects

ripyl.streaming.merge_streams(records_a, records_b, id_a=0, id_b=1)

Combine two streams of StreamRecord objects. Records with time signatures from each input stream are kept in chronological order. :param : :type records_a: StreamRecord :param records_a:

Source records from stream a

:param : :type records_b: StreamRecord :param records_b:

Source records from stream b

:param : :type id_a: int :param id_a:

stream_id assigned to records from records_a

:param : :type id_b: int :param id_b:

stream_id assigned to records from records_b

:param : :returns: An iterator yielding a stream of StreamRecord objects.

ripyl.streaming.rechunkify(samples, chunk_size=10000)

Create a new gerator that yields SampleChunk objects of the desired size

This is a generator function. Its send() method can be used to change the value of chunk_size mid-stream.

Parameters:
  • samples (iterable of SampleChunk objects) – The sample stream to extract SampleChunk objects from.
  • chunk_size (int) – The number of samples for the SampleChunk objects.
Returns:

An iterator yielding a series of SampleChunk objects. If the stream has fewer than chunk_size samples remaining then the SampleChunk.samples array is sized to hold only those samples.
ripyl.streaming.sample_stream_to_samples(samples)

Get all samples from a sample stream as an array

Parameters:samples (iterable of SampleChunk objects) – The sample stream to extract samples from.
Returns:A numpy array of float.
ripyl.streaming.samples_to_sample_stream(raw_samples, sample_period, start_time=0.0, chunk_size=10000)

Convert raw samples to a chunked sample stream

This is a generator function that can be used in a pipeline of waveform procesing operations

Parameters:
  • raw_samples (iterable of numbers) – The samples to convert to a sample stream.
  • sample_period (float) – The time interval between samples
  • start_time (float) – The time for the first sample
  • chunk_size (int) – The maximum number of samples for each chunk
Returns:

An iterator yielding a series of SampleChunk objects representing the time and sample value for each input sample. This can be fed to functions that expect a chunked sample stream as input.
ripyl.streaming.save_stream(records, fh)

Save a stream of StreamRecord objects to a file

Parameters:
  • records (StreamRecord sequence) – The StreamRecord objects to save.
  • fh (file-like object or a string) – File to save records to. If a file handle is passed it should have been opened in ‘wb’ mode. If a string is passed it is the name of a file to write to.
Raises:

TypeError when records parameter is not a sequence.

ripyl.wave_synth module

General routines for synthesizing basic waveforms

ripyl.wave_synth.sine_synth(frequency=1000000.0, phase=0.0, sample_period=5e-08, samples=None)

Sine generator

Will generate a continuous stream of samples from a sine wave. This generator will terminate if the number of output samples is specified. :param : :type frequency: float :param frequency:

Frequency in Hz of the sine wave
Parameters:
  • phase (float) – Phase in radians of the sine wave
  • sample_period (float) – Time interval between samples
  • samples (int) – Optional number of samples to generate

:param : :returns: An iterator yielding a series of raw samples.

ripyl.wave_synth.square_synth(frequency=1000000.0, duty=0.5, phase=0.0, sample_period=5e-08, samples=None)

Square wave generator

Will generate a continuous stream of samples from a square wave. This generator will terminate if the number of output samples is specified. :param : :type frequency: float :param frequency:

Frequency in Hz of the square wave

:param : :type duty: float :param duty:

Duty cycle of the square wave. Should be between 0.0 and 1.0.
Parameters:
  • phase (float) – Phase of the square wave in fractions of a period. Should be between 0.0 and 1.0.
  • sample_period (float) – Time interval between samples
  • samples (int) – Optional number of samples to generate

:param : :returns: An iterator yielding a series of raw samples.

Module contents

Ripyl protocol decode library