Modifiers

Modifier

class crappy.modifier.modifier.MetaModifier(name: str, bases: tuple, dct: dict)

Metaclass keeping track of all the Modifiers, including the custom user-defined ones.

class crappy.modifier.modifier.Modifier

The base class for all modifier classes, simply allowing to keep track of them.

Demux

class crappy.modifier.demux.Demux(labels: str | List[str] | Tuple[str, ...], stream_label: str = 'stream', mean: bool = False, time_label: str = 't(s)', transpose: bool = False)

Modifier converting a stream into a regular Crappy dict giving for each label a single value.

The single value is either the first value of a column/row, or the average of the row/column values. This Modifier is mainly meant for linking streaming IOBlock blocks to Grapher blocks, as it is otherwise impossible to plot their data.

__init__(labels: str | List[str] | Tuple[str, ...], stream_label: str = 'stream', mean: bool = False, time_label: str = 't(s)', transpose: bool = False) → None

Sets the args and initializes the parent class.

Parameters:

• labels – The labels corresponding to the rows or columns of the stream. It can be either a single label, or a list of labels, or a tuple of labels. They must be given in the same order as they appear in the stream. If fewer labels are given than there are rows or columns in the stream, only the data from the first rows/columns will be retrieved.

• stream_label – The label carrying the stream.

• mean – If True, the returned value will be the average of the stream data. Otherwise, it will be the first value.

• time_label – The label carrying the time information.

• transpose – If True, each label corresponds to a row in the stream. Otherwise, a label corresponds to a column in the stream.

evaluate(data: Dict[str, ndarray]) → Dict[str, Any]

Retrieves for each label its value in the stream, also gets the corresponding timestamp, and returns them.

Differentiate

class crappy.modifier.differentiate.Diff(label: str, time_label: str = 't(s)', out_label: str | None = None)

This modifier differentiates the data of a label over time and adds the differentiation value to the returned data.

__init__(label: str, time_label: str = 't(s)', out_label: str | None = None) → None

Sets the args and initializes the parent class.

Parameters:

• label – The label whose data to differentiate over time.

• time_label – The label carrying the time information.

• out_label – The label carrying the differentiation value. If not given, defaults to 'd_<label>'.

evaluate(data: Dict[str, Any]) → Dict[str, Any]

Gets the data from the upstream block, updates the differentiation value and returns it.

Integrate

class crappy.modifier.integrate.Integrate(label: str, time_label: str = 't(s)', out_label: str | None = None)

This modifier integrates the data of a label over time and adds the integration value to the returned data.

__init__(label: str, time_label: str = 't(s)', out_label: str | None = None) → None

Sets the args and initializes the parent class.

Parameters:

• label – The label whose data to integrate over time.

• time_label – The label carrying the time information.

• out_label – The label carrying the integration value. If not given, defaults to 'i_<label>'.

evaluate(data: Dict[str, Any]) → Dict[str, Any]

Gets the data from the upstream block, updates the integration value and returns it.

Mean

class crappy.modifier.mean.Mean(n_points: int = 100)

Modifier waiting for a given number of data points to be received, then returning their average, and starting all over again.

Unlike Moving average, it only returns a value once every n_points points.

__init__(n_points: int = 100) → None

Sets the args and initializes the parent class.

Parameters:

n_points – The number of points on which to compute the average.

evaluate(data: Dict[str, Any]) → Dict[str, Any] | None

Receives data from the upstream block, and computes the average of every label once the right number of points have been received. Then empties the buffer and returns the averages.

If there are not enough points, doesn’t return anything.

Median

class crappy.modifier.median.Median(n_points: int = 100)

Modifier waiting for a given number of data points to be received, then returning their median, and starting all over again.

Unlike Moving med, it only returns a value once every n_points points.

__init__(n_points: int = 100) → None

Sets the args and initializes the parent class.

Parameters:

n_points – The number of points on which to compute the median.

evaluate(data: Dict[str, Any]) → Dict[str, Any] | None

Receives data from the upstream block, and computes the median of every label once the right number of points have been received. Then empties the buffer and returns the medians.

If there are not enough points, doesn’t return anything.

Moving average

class crappy.modifier.moving_avg.Moving_avg(n_points: int = 100)

Modifier replacing the data of each label with its average value over a chosen number of points.

Unlike Mean, it returns a value each time a message is received from the upstream link.

__init__(n_points: int = 100) → None

Sets the args and initializes the parent class.

Parameters:

n_points – The maximum number of points on which to compute the average.

evaluate(data: Dict[str, Any]) → Dict[str, Any]

Receives data from the upstream block, computes the average of every label and replaces the original data with it.

Moving med

class crappy.modifier.moving_med.Moving_med(n_points: int = 100)

Modifier replacing the data of each label with its median value over a chosen number of points.

Unlike Median, it returns a value each time a message is received from the upstream link.

__init__(n_points: int = 100) → None

Sets the args and initializes the parent class.

Parameters:

n_points – The maximum number of points on which to compute the median.

evaluate(data: Dict[str, Any]) → Dict[str, Any]

Receives data from the upstream block, computes the median of every label and replaces the original data with it.

Offset

class crappy.modifier.offset.Offset(labels: str | List[str] | Tuple[str, ...], offsets: float | List[float] | Tuple[float, ...])

This Modifier offsets every value of the given labels by a constant. This constant is calculated so that for each label the first returned value is equal to a user-defined target.

For example if for a given label the target is 6 and the first received value is 3, the Modifier will add 3 to each value received over this label.

This Modifier can be used for example when measuring a variable that should start at 0 (like a force) but doesn’t because of a sensor offset. It can also just be used to plot nicer figures. It is not very accurate as it is only based on a single data point for the offset calculation. The make_zero argument of the IOBlock is a better alternative if precision is required when offsetting the data.

__init__(labels: str | List[str] | Tuple[str, ...], offsets: float | List[float] | Tuple[float, ...]) → None

Sets the args and initializes the parent class.

Parameters:

• labels – The labels to offset. Can be given as a single label, a list of labels or a tuple of labels.

• offsets – For each label, the target for the first received value. Can be given as a single value, a list of values or a tuple of values.

evaluate(data: Dict[str, Any]) → Dict[str, Any]

If the compensations are not set, sets them, anf then offsets the required labels.

Trig on change

class crappy.modifier.trig_on_change.Trig_on_change(label: str)

Modifier passing the data to the downstream block only when the value of a given label changes.

It also transmits the first received data. Can be used to trig a block upon change of a label value.

__init__(label: str) → None

Sets the args and initializes the parent class.

Parameters:

label – The name of the label to monitor.

evaluate(data: Dict[str, Any]) → Dict[str, Any] | None

Compares the received value with the last sent one, and if they’re different sends the received data and stores the latest value.

Trig on value

class crappy.modifier.trig_on_value.Trig_on_value(label: str, values: Any | Tuple[Any, ...] | List[Any])

Modifier passing the data to the downstream only if the value carried by a given label matches a given set of accepted values.

Mostly useful to trig blocks.

__init__(label: str, values: Any | Tuple[Any, ...] | List[Any]) → None

Sets the args and initializes the parent class.

Parameters:

• label – The name of the label to monitor.

• values – The values of label for which the data will be transmitted. Can be a single value, a list or a tuple.

evaluate(data: Dict[str, Any]) → Dict[str, Any] | None

Checks if the value of label is in the predefined set of accepted values, and if so transmits the data.