Blocks

Autodrive

class crappy.blocks.autoDrive.AutoDrive(*args, **kwargs)[source]

To follow the spots with VideoExtenso.

This blocks takes the output of a VideoExtenso block and uses the spots coordinates to drive an actuator, on which the camera is mounted.

This will allow the camera to follow the spots along an axis.

It is simply a P loop: the difference in pixel between the barycenter of the spots and the middle of the image along the given axis is multiplied by P and set as the speed of the actuator.

__init__(actuator: Optional[dict] = None, P: float = 2000, direction: str = 'Y-', range: int = 2048, max_speed: float = 200000) → None[source]

Sets the args and initializes parent class.

Parameters

actuator (dict, optional) –
P (float, optional) – The gain for commanding the actuator.
direction (str, optional) – What axis should be considered, and in which direction ? A str whose first character is the axis (X or Y) and second character is the direction (+ or -).
range (int, optional) – The size in pixel of the image along the chosen axis.
max_speed (float, optional) – The absolute max value to send to the actuator.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Camera

class crappy.blocks.camera.Camera(*args, **kwargs)[source]

Reads images from a camera object, saves and/or sends them to another block.

It can be triggered by an other block, internally, or try to run at a given framerate.

__init__(camera: str, save_folder: Optional[str] = None, verbose: bool = False, labels: Optional[list] = None, fps_label: str = False, img_name: str = '{self.loops:06d}_{t-self.t0:.6f}', ext: str = 'tiff', save_period: int = 1, save_backend: Optional[str] = None, transform: Optional[Callable] = None, input_label: Optional[str] = None, config: bool = True, no_loop: bool = False, **kwargs) → None[source]

Sets the args and initializes parent class.

Parameters

camera (str) – The name of the camera to control. See Cameras for an exhaustive list of available ones.
save_folder (str, optional) – The directory to save images to. If it doesn’t exist it will be created. If None the images won’t be saved.
verbose (bool, optional) – If True, the block will print the number of loops/s.
labels (list, optional) – Names of the labels for respectively time and the frame.
fps_label (str, optional) – If set, self.max_fps will be set to the value received by the block with this label.
img_name (str, optional) – Template for the name of the image to save. It is evaluated as an f-string.
ext (str, optional) – Extension of the image. Make sure it is supported by the saving backend.
save_period (int, optional) – Will save only one in x images.
save_backend (str, optional) – Module to use to save the images. The supported backends are: sitk (SimpleITK), cv2 (OpenCV) and pil (Pillow). If None, will try sitk and then cv2 if not successful.
transform (function, optional) – Function to be applied on the image before sending. It will not be applied on the saved images.
input_label (str, optional) – If specified, the image will not be read from a camera object but from this label.
config (bool, optional) – Show the popup for config ?
**kwargs – Any additional specific argument to pass to the camera.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

get_img() → Optional[tuple][source]: Waits the appropriate time/event to read an image, reads it, saves it if asked to, applies the transformation and increases counter.

prepare(send_img: bool = True) → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Client Server

class crappy.blocks.client_server.Client_server(*args, **kwargs)[source]

Block for exchanging data on a local network using the MQTT protocol.

This block can send data to an MQTT broker, receive data from this broker by subscribing to its topics, and also launch the Mosquitto broker.

__init__(broker: bool = False, address: Any = 'localhost', port: int = 1148, init_output: Optional[Dict[str, Any]] = None, topics: Optional[List[Union[str, Tuple[str, str]]]] = None, cmd_labels: Optional[List[Union[str, Tuple[str, str]]]] = None, labels_to_send: Optional[List[Union[str, Tuple[str, str]]]] = None) → None[source]

Checks arguments validity and sets the instance attributes.

Parameters

broker (bool, optional) – If True, starts the Mosquitto broker during the prepare loop and stops it during the finish loop. If Mosquitto is not installed a FileNotFoundError is raised.
address (optional) – The network address on which the MQTT broker is running.
port (int, optional) – A network port on which the MQTT broker is listening.
init_output (dict, optional) – A dict containing for each label in topics the first value to be sent in the output link. Must be given if topics is not None.
topics (list, optional) – A list of str and/or tuple of str. Each string corresponds to the name of a crappy label to be received from the broker. Each element of the list is considered to be the name of an MQTT topic, to which the client subscribes. After a message has been received on that topic, the block returns for each label in the topic (i.e. each string in the tuple) the corresponding data from the message. It also returns the current timestamp in the label ‘t(s)’.
cmd_labels (list, optional) – A list of str and/or tuple of str. Each string corresponds to the name of a crappy label to send to the broker. Each element of the list is considered to be the name of an MQTT topic, in which the client publishes. Grouping labels in a same topic (i.e. strings in a same tuple) allows to keep the synchronization between signals coming from a same block, as they will be published together in a same message. This is mostly useful for sending a signal along with its timeframe.
labels_to_send (list, optional) – A list of str and/or tuple of str. Allows to rename the labels before publishing data. The structure of labels_to_send should be the exact same as cmd_labels, with each label in labels_to_send replacing the corresponding one in cmd_labels. This is especially useful for transferring several signals along with their timestamps, as the label 't(s)' should not appear more than once in the topics list of the receiving block.

Note

broker: In order for the block to run, an MQTT broker must be running at the specified address on the specified port. If not, an ConnectionRefusedError is raised. The broker can be started and stopped manually by the user independently of the execution of crappy. It also doesn’t need to be Mosquitto, any other MQTT broker can be used.
topics: The presence of the same label in multiple topics will most likely lead to a data loss.
cmd_labels: It is not possible to group signals coming from different blocks in a same topic.
labels_to_send: Differences in the structure of labels_to_send and cmd_labels will not always raise an error, but may lead to a data loss.

Single-value tuples: Single-value tuples can be shortened as strings.

topics=[('cmd1',), ('cmd2',)]
cmd_labels=[('cmd1',), ('cmd2',)]
labels_to_send=[('cmd1',), ('cmd2',)]

is equivalent to

topics=['cmd1', 'cmd2']
cmd_labels=['cmd1', 'cmd2']
labels_to_send=['cmd1', 'cmd2']

Examples

topics: If
```
topics=[('t1', 'cmd1'), 'sign']
```
the client will subscribe to the topics
```
('t1', 'cmd1')
('sign',)
```
The block will return data associated with the labels
```
't1', 'cmd1'
'sign'
```

cmd_labels: If

cmd_labels=[('t1', 'cmd1'), 'sign']

the client will publish data in the form of

[[t1_0, cmd1_0], [t1_1, cmd1_1], ...]
[[sign_0], [sign_1], ...]

in the topics

('t1', 'cmd1')
('sign',)

labels_to_send: If

cmd_labels=[('t(s)', 'cmd'), 'sign']
labels_to_send=[('t1', 'cmd1'), 'sign']

the data from labels

't(s)', 'cmd'

will be published in the topic

('t1', 'cmd1')

and the data from label

'sign'

in the topic

('sign',)

finish() → None[source]: Disconnects from the broker and stops it.

loop() → None[source]

Receives data from the broker and/or sends data to the broker.

The received data is then sent to the crappy blocks connected to this one.

prepare() → None[source]: Reorganizes the labels lists, starts the broker and connects to it.

Dashboard

class crappy.blocks.dashboard.Dashboard(*args, **kwargs)[source]

The Dashboard receives data from a Link, and prints it on a new popped window.

It can only display data coming from one block.

__init__(labels: list, nb_digits: int = 2, verbose: bool = False, freq: float = 30) → None[source]

Sets the args and initializes parent class.

Parameters

labels (list) – Values to plot on the output window.
nb_digits (int, optional) – Number of decimals to show.
verbose (bool, optional) – Display loop frequency ?
freq (float, optional) – If set, the block will loop at this frequency.

finish() → None[source]: Closes the thread.

loop() → None[source]: Simply transmits the received data to the thread.

prepare() → None[source]: Creates the window in a new thread.

class crappy.blocks.dashboard.Dashboard_window(labels: list, nb_digits: int, queue: queue.Queue)[source]

Dashboard class created, is launched in a new thread.

update() → None[source]: Method to update the output window.

Discorrel

Displayer

class crappy.blocks.displayer.Displayer(*args, **kwargs)[source]: Simple image displayer using cv2 or matplotlib.

Important

One displayer can only display images from one camera.

Disve

Drawing

class crappy.blocks.drawing.Dot_text(drawing, **kwargs)[source]

Like Text, but with a colored dot to visualize a numerical value.

__init__(drawing, **kwargs) → None[source]

Sets the args.

Parameters

drawing – The Drawing object.
**kwargs –
Contains the coordinates, the text and the label to be displayed. Also contains the type of drawing, but unused.

Important

The value received in label must be a numeric value. It will be normalized on the crange of the block and the dot will change color from blue to red depending on this value.

class crappy.blocks.drawing.Drawing(*args, **kwargs)[source]

Block to make a visual representation of data.

__init__(image, draw: Optional[list] = None, crange: Optional[list] = None, title: str = 'Drawing', window_size: tuple = (7, 5), freq: float = 2, backend: str = 'TkAgg') → None[source]

Sets the args and initializes the parent block.

Parameters

image – This image will be the background for the Canvas.
draw (dict, optional) – A list of dict defining what to draw. See below for more details.
crange –
title –
window_size –
freq –
backend –

Note

draw keys:
- type (str): Mandatory, the type of drawing to display. It can be either ‘Text’, ‘Dot_text’ or ‘’Time.
- coord (list): Mandatory, a list containing the x and y coordinates where the drawing should be displayed.
- text: Mandatory for Text and Dot_text only, the left part of the displayed string.
- label (str): Mandatory for Text and Dot_text only, the label of the data to display. It will be append to the text.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

class crappy.blocks.drawing.Text(*_, **kwargs)[source]

A simple text line.

__init__(*_, **kwargs) → None[source]

Sets the args.

Parameters

*_ – Contains the Drawing object, unused.
**kwargs – Contains the coordinates, the text and the label to be displayed. Also contains the type of drawing, but unused.

class crappy.blocks.drawing.Time(drawing, **kwargs)[source]

To print the time of the experiment.

It will print the time since the t0 of the block.

Fake machine

class crappy.blocks.fake_machine.Fake_machine(*args, **kwargs)[source]

Block to simulate the behavior of a tensile testing machine.

By default, it is supposed to represent a material with a plastic behavior.

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Generator

class crappy.blocks.generator.Generator(*args, **kwargs)[source]

This block is used to generate a signal.

It can be used to drive a machine. This block can take inputs, and each path can use these inputs to take decisions.

__init__(path: Optional[list] = None, freq: float = 200, cmd_label: str = 'cmd', cycle_label: str = 'cycle', cmd: float = 0, repeat: bool = False, trig_link: Optional[str] = None, spam: bool = False, verbose: bool = False, end_delay: float = 2) → None[source]

Sets the args and initializes parent class.

Parameters

path (list, optional) –
It must be a list of dict, each dict providing the parameters to generate the path. Each dict MUST have a key type.

Note

The Generator will then instantiate a generator path with all the other keys as kwargs, adding the current cmd and the time.

On each round, it will call Path.get_cmd() method, passing data until it raise StopIteration. It will then skip to the next path.

When all paths are over, it will stop Crappy by raising CrappyStop unless repeat is set to True. If so, it will start over indefinitely.
freq (float, optional) – The frequency of the block. If set and positive, the generator will try to send the command at this frequency (in Hz). Else, it will go as fast as possible. It relies on the Block freq control scheme.
cmd_label (str, optional) – The label of the command to send in the links
cycle_label (str, optional) –
cmd (float, optional) – The first value of the command.
repeat (bool, optional) – Loop over the paths or stop when done ?
trig_link (str, optional) – If given, the block will wait until data is received through the input link with this label. If None, it will try loop at freq.
spam (bool, optional) – If True, the value will be sent on each loop. Else, it will only send it if it was updated or we reached a new step.
verbose (bool, optional) – if True, displays a message when switching to the next path.
end_delay (float, optional) – The delay to wait for before raising the CrappyStop exception at the end of the path. This is meant to let enough time to the other blocks to properly terminate.

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

generator path

There are several types of path available for the generator block.

class crappy.blocks.generator_path.path.Path(time: float, cmd: float)[source]

Parent class for all paths.

parse_condition(condition: Union[str, bool, Callable]) → Callable[source]

This method turns a string into a function that returns a bool.

It is meant to check if a skip condition is reached.

The following syntax is supported:

myvar>myvalue
myvar<myvalue
delay=mydelay

Note

myvar must be the label of an input value, myvalue should be a float. This will return True when the data under the label myvar is larger/smaller than myvalue.

The condition will turn True after mydelay seconds.

Any other syntax will return True.

constant

class crappy.blocks.generator_path.constant.Constant(time: float, cmd: float, condition: Union[str, bool, Callable], send_one: bool = True, value: Optional[float] = None)[source]

Simplest condition. It will send value until condition is reached.

__init__(time: float, cmd: float, condition: Union[str, bool, Callable], send_one: bool = True, value: Optional[float] = None) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition (str) – Representing the condition to end this path. See generator path for more info.
send_one (str, optional) – If True, this condition will send the value at least once before checking the condition.
value – What value must be sent.

custom

class crappy.blocks.generator_path.custom.Custom(time: float, cmd: float, filename: str, delimiter: str = '\t')[source]

To generate a custom path from a file.

__init__(time: float, cmd: float, filename: str, delimiter: str = '\t') → None[source]

Loads the file and sets the args.

Parameters

time –
cmd –
filename –
Name of the .csv file.

Note

It must contain two columns: one with time, the other with the value.
delimiter –

cyclic ramp

class crappy.blocks.generator_path.cyclic_ramp.Cyclic_ramp(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], speed1: float, speed2: float, cycles: float = 1, verbose: bool = False)[source]

A “boosted” ramp path: will take TWO values and conditions.

Note

It will make a ramp of speed speed1, switch to the second when the first condition is reached and return to the first when the second condition is reached.

This will be done cycles times (supporting half cycles for ending after the first condition)

__init__(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], speed1: float, speed2: float, cycles: float = 1, verbose: bool = False) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition1 (str) – Representing the condition to switch to speed2. See generator path for more info.
condition2 (str) – Representing the condition to switch to speed1. See generator path for more info.
speed1 – Speed of the first ramp.
speed2 – Speed of the second ramp.
cycles –
Number of time we should be doing this.

Note

cycles = 0 will make it loop forever.
verbose –

Note

[{'type': 'cyclic_ramp', 'speed1': 5, 'condition1': 'AIN0>2',
'speed2': -2, 'condition2': 'AIN1<1', 'cycles': 5}]

is equivalent to

[{'type': 'ramp', 'speed': 5,'condition': 'AIN0>2'},
{'type': 'ramp', 'value': -2, 'condition': 'AIN1<1'}] * 5

cyclic

class crappy.blocks.generator_path.cyclic.Cyclic(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], value1: float, value2: float, cycles: float = 1, verbose: bool = False)[source]

A “boosted” constant path: will take TWO values and conditions.

Note

It will set the first value, switch to the second when the first condition is reached and return to the first when the second condition is reached.

This will be done cycles times (supporting half cycles for ending after the first condition).

__init__(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], value1: float, value2: float, cycles: float = 1, verbose: bool = False) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition1 (str) – Representing the condition to switch to value2. See generator path for more info.
condition2 (str) – Representing the condition to switch to value1. See generator path for more info.
value1 – First value to send.
value2 – Second value to send.
cycles –
Number of time we should be doing this.

Note

cycles = 0 will make it loop forever.
verbose –

Note

[{'type': 'cyclic', 'value1': 1, 'condition1': 'AIN0>2',
'value2': 0, 'condition2': 'AIN1<1', 'cycles': 5}]

is equivalent to

[{'type': 'constant', 'value': 1,'condition': 'AIN0>2'},
{'type': 'constant', 'value': 0, 'condition': 'AIN1<1'}] * 5

inertia

class crappy.blocks.generator_path.inertia.Inertia(time: float, cmd: float, condition: Union[str, bool, Callable], inertia: float, flabel: str, const: float = 9.549296585513721, tlabel: str = 't(s)', value: Optional[float] = None)[source]

Used to lower/higher the output command by integrating an input over time.

Let f(t) be the input signal, v(t) the value of the output, m the inertia and t0 the beginning of this path. K is a chosen constant.

Then the output value for this path will be:

v(t) = v(t0) - K * [I(t0 -> t)f(t)dt] / m

__init__(time: float, cmd: float, condition: Union[str, bool, Callable], inertia: float, flabel: str, const: float = 9.549296585513721, tlabel: str = 't(s)', value: Optional[float] = None) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition (str) – Condition to meet to end this path. See generator path for more info.
inertia (float) – This is the virtual inertia of the process. The higher it is, the slower the (in/de) crease will be. In the above formula, it is the value of m.
flabel (str) – The name of the label of the value to integrate.
const (float, optional) – The value of K in the formula above. The default value is meant to send rpm with inertia in kg.m² and torque in N.m. If sending rad/s, use const=1.
tlabel (str, optional) –
The name of the label of time for the integration.

Note

The data received by flabel and tlabel must correspond. In other word, there must be exactly the same number of values received by these two labels at any instant (i.e. they must come from the same parent block).
value –

protection

class crappy.blocks.generator_path.protection.Protection(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], value1: float, value2: float, value0: float = 0, verbose: bool = False)[source]

Useful to protect samples from being pulled apart when setting up a test.

__init__(time: float, cmd: float, condition1: Union[str, bool, Callable], condition2: Union[str, bool, Callable], value1: float, value2: float, value0: float = 0, verbose: bool = False) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition1 (str) – Representing the first condition. See generator path for more info.
condition2 (str) – Representing the second condition. See generator path for more info.
value1 – Value to send when condition1 is met.
value2 – Value to send when condition2 is met.
value0 – Value to send when no condition is reached.
verbose –

Note

By default will send value0.

While condition1 is met, will return value1.

While condition2 is met, will return value2.

If condition1 and condition2 are met simultaneously, the first one met will prevail. If met at the same time, condition1 will prevail.

ramp

class crappy.blocks.generator_path.ramp.Ramp(time: float, cmd: float, condition: Union[str, bool, Callable], speed: float)[source]

Will make a ramp from previous value until condition is reached.

__init__(time: float, cmd: float, condition: Union[str, bool, Callable], speed: float)[source]

Sets the args and initializes parent class.

Parameters

time –
cmd – If specified, will be the starting value of the ramp.
condition (str) – Representing the condition to end this path. See generator path for more info.
speed – The speed of the ramp in units/s.

sine

class crappy.blocks.generator_path.sine.Sine(time: float, cmd: float, condition: Union[str, bool, Callable], freq: float, amplitude: float, offset: float = 0, phase: float = 0)[source]

To generate a sine wave.

__init__(time: float, cmd: float, condition: Union[str, bool, Callable], freq: float, amplitude: float, offset: float = 0, phase: float = 0) → None[source]

Sets the args and initializes parent class.

Parameters

time –
cmd –
condition (str) – Representing the condition to end this path. See generator path for more info.
freq – Frequency of the sine in Hz.
amplitude – Amplitude of the sine wave.
offset (optional) – Offset of the sine.
phase (optional) – Phase of the sine.

GPUCorrel

class crappy.blocks.gpucorrel.GPUCorrel(*args, **kwargs)[source]

This block uses the GPU Correl class.

See the documentation of GPU Correl to have more information about the arguments specific to this class.

It will try to identify the deformation parameters for each fields. If you use custom fields, you can use labels=(...) to name the data sent through the link. If no labels are specified, custom fields will be named by their position.

The reference image is only taken once, when the start() method is called (after dropping the first image).

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare(*_, **__) → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

GPUve

class crappy.blocks.gpuve.GPUVE(*args, **kwargs)[source]

An optical Videoextensometry measuring the displacement of small areas using GPU accelerated DIC.

This block simply returns the displacement of each region along x and y in pixel. It will not return the strain as it does not know how the patches are arranged. It should be done by another block or a condition if necessary.

Important

patches must be a list of tuple of length 4. Each tuple contains the origin and the size of each patch along Y and X respectively (i.e. Oy, Ox, Ly, Lx).

prepare(*_, **__) → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Grapher

class crappy.blocks.grapher.Grapher(*args, **kwargs)[source]

The grapher receive data from a block (via a Link) and plots it.

__init__(*labels: tuple, length: int = 0, freq: float = 2, maxpt: int = 20000, window_size: tuple = (8, 8), window_pos: Optional[tuple] = None, interp: bool = True, backend: str = 'TkAgg', verbose: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters

*labels (tuple) – Tuples of the columns labels of input data for plotting. You can add as much as you want, depending on your performances. The first value is the x label, the second is the y label.
length (int, optional) – If 0 the graph is static and displays all data from the start of the assay. Else only displays the last length received chunks, and drops the previous ones.
freq (float, optional) – The refresh rate of the graph. May cause high memory consumption if set too high.
maxpt (int, optional) – The maximum number of points displayed on the graph. When reaching this limit, the block deletes one point out of two but this is almost invisible to the user.
window_size (tuple, optional) – The size of the graph, in inches.
window_pos (tuple, optional) – The position of the graph in pixels. The first value is for the x direction, the second for the y direction. The origin is the top left corner. Works with multiple screens.
interp (bool, optional) – If True, the points of data will be linked to the following by straight lines. Else, each value wil be displayed as constant until the next update.
backend (int, optional) – The matplotlib backend to use.
verbose (bool, optional) – To display the loop frequency of the block.

Example

graph = Grapher(('t(s)', 'F(N)'), ('t(s)', 'def(%)'))

will plot a dynamic graph with two lines plot (F=f(t) and def=f(t)).

graph = Grapher(('def(%)', 'F(N)'), length=0)

will plot a static graph.

graph = Grapher(('t(s)', 'F(N)'), length=30)

will plot a dynamic graph displaying the last 30 chunks of data.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

GUI

class crappy.blocks.gui.GUI(*args, **kwargs)[source]

Block to send a signal based on a user input.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

HDF Recorder

class crappy.blocks.hdf_recorder.Hdf_recorder(*args, **kwargs)[source]

To save data efficiently in a hdf5 file.

This block is is meant to save data coming by arrays at a high rate (>1kHz). It uses the module tables.

Important

Do not forget to specify the type of data to be saved (see atom parameter) to avoid casting the data into another type, as this could result in data loss or inefficient saving.

__init__(filename: str, node: str = 'table', expected_rows: int = 100000000, atom=None, label: str = 'stream', metadata: Optional[dict] = None) → None[source]

Sets the args and initializes the parent class.

Parameters

filename (str) – Path of the file where the data should be saved.
node (str, optional) – The name of the node where the data will be saved.
expected_rows (int, optional) – The number of expected rows in the file. It is used to optimize the dumping.
atom (optional) – This represent the type of data to be stored in the table. It can be given as a tables.Atom instance, as a numpy.array or as a str.
label (str, optional) – The key of the dict that contains the array to save.
metadata (dict, optional) – A dict containing additional info to save in the hdf5 file.

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

class crappy.blocks.hdf_recorder.Hdf_saver(*args, **kwargs)[source]

IOBlock

class crappy.blocks.ioblock.IOBlock(*args, **kwargs)[source]

This block is used to communicate with In / Out objects.

They can be used as sensor, actuators or both.

__init__(name: str, freq: Optional[float] = None, verbose: bool = False, labels: Optional[list] = None, cmd_labels: Optional[list] = None, trigger: Optional[int] = None, streamer: bool = False, initial_cmd: Union[float, list] = 0, exit_values: Optional[list] = None, spam: bool = False, **kwargs) → None[source]

Sets the args and initializes the parent class.

Parameters

name (str) – The name of the In / Out class to instantiate.
freq (float, optional) – The looping frequency of the block, if None will go as fast as possible.
verbose (bool, optional) – Prints extra information if True.
labels (list, optional) – A list of the output labels.
cmd_labels (list, optional) – The list of the labels considered as inputs for this block. Will call set_cmd() in the In / Out object with the values received on this labels.
trigger (int, optional) – If the block is triggered by another block, this must specify the index of the input considered as a trigger. The data going through this link is discarded, add another link if the block should also consider it as an input.
streamer (bool, optional) – If False, will call get_data() else, will call get_stream() in the In / Out object (only if it has these methods, of course).
initial_cmd (list, optional) – The initial values for the outputs, sent during prepare(). If it is a single value, then it will send this same value for all the output labels.
exit_values (list, optional) – If not None, the outputs will be set to these values when Crappy is ending (or crashing).
spam (bool, optional) – If False (default), the block will only call set_cmd on the InOut object if the command changed
**kwargs – The arguments to be passed to the In / Out class.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

read() → None[source]: Will read the device and send the data.

Machine

class crappy.blocks.machine.Machine(*args, **kwargs)[source]

To drive a machine with a one or more Actuators.

Takes a list of dict, containing the information to create each actuator. Each key will either stand for a parameter, or else they are transferred to the actuator.

__init__(actuators: list, common: Optional[dict] = None, freq: float = 200, time_label: str = 't(s)', spam: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters

actuators (list) – The list of the Actuators of the machine. It contains one or several dict, whose mandatory keys are described below. The other keys will be passed to the actuator as arguments.
common (dict, optional) – The keys of this dict will be common to all of the actuators. However if this conflicts with an already existing key for an actuator, the latter will prevail.
freq (float, optional) – The looping frequency of the block.
time_label (str, optional) – If reading data from one or more actuators, the time will be returned under this label.
spam (bool, optional) – If True, a command is sent on each loop of the block, else it is sent every time a value is received.

Note

actuators keys:
- type (str): The name of the actuator to instantiate.
- cmd (str): The label of the input to drive the axis.
- mode (str, default: ‘speed’): Can be either ‘speed’ or ‘position’. Will either call set_speed() or set_position() to drive the actuator.
- speed (float): If mode is ‘position’, the speed of the axis.
- pos_label (str): If set, the block will return the value of get_position() with this label.
- speed_label (str): If set, the block will return the value of get_speed() with this label.

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Mean_block

class crappy.blocks.mean.Mean_block(*args, **kwargs)[source]

Can take multiple inputs, makes an average and sends the result every delay s.

__init__(delay: float, tlabel: str = 't(s)', out_labels: Optional[list] = None, freq: float = 50) → None[source]

Sets the args and initializes the parent class.

Parameters

delay (float) – The averaged data will be sent each delay seconds.
tlabel (str, optional) – The label containing the time information.
out_labels (list, optional) – If given, only the listed labels and the time will be returned. Otherwise all of them are returned.
freq – The block will loop at this frequency.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

Multiplex

class crappy.blocks.multiplex.Multiplex(*args, **kwargs)[source]

This block interpolates data.

It is used to read data and return the reading from multiple sensors at the same instants and a constant frequency. This block uses linear interpolation whenever possible, else the nearest neighbor.

Warning

This block needs a delay to make sure all the required data for interpolation has already been received, so do not use this block as the input of a decision block!

__init__(key: str = 't(s)', freq: float = 200) → None[source]

Sets the args and initializes the parent class.

Parameters

key (str, optional) – The label carrying the time information.
freq (float, optional) – The block will loop at this frequency.

begin() → None[source]: We need to receive the first bit of data from each input to know the labels and make lists of what we will read in the main loop.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

PID

class crappy.blocks.pid.PID(*args, **kwargs)[source]

A PID corrector.

A PID will continuously adjust its output based on the target value and the actual measured value, to try to actually reach the target.

__init__(kp: float, ki: float = 0, kd: float = 0, freq: float = 500, out_max: float = inf, out_min: float = - inf, target_label: str = 'cmd', input_label: str = 'V', time_label: str = 't(s)', labels: Optional[list] = None, reverse: bool = False, i_limit: Union[float, tuple] = 1, send_terms: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters

kp (float) – P gain.
ki (float) – I gain.
kd (float) – D gain.
freq (float, optional) – The block will loop at this frequency.
out_max (float, optional) – A value the output can never be superior to.
out_min (float, optional) – A value the output can never be inferior to.
target_label (str, optional) – The label of the setpoint.
input_label (str, optional) – The reading of the actual value to be compared with the setpoint.
time_label (str, optional) – The label of the time.
labels (list, optional) – The labels of the output of the block. It must contain two str : the time label and the actual output.
reverse (bool, optional) – To reverse the retro-action.
i_limit (tuple, optional) –
To avoid over-integration. If given as a tuple of two values, they will be the boundaries for the I term. If given as a single float the boundaries will be:
```
i_limit * out_min, i_limit * out_max
```
send_terms (bool, optional) – To get the weight of each term in the output value. It will add ['p_term', 'i_term', 'd_term'] to the labels. This is particularly useful to tweak the gains.

begin() → None[source]: If main() is not overridden, this method will be called first, before entering the main loop.

Reader

class crappy.blocks.reader.Reader(*args, **kwargs)[source]

Reads and prints the input Link.

Creates a reader that prints the input data continuously.

__init__(reader_name: Optional[str] = None) → None[source]

Sets the arg and initializes the parent class.

Parameters: reader_name (str, optional) – If set, will be printed to identify the reader.

Recorder

class crappy.blocks.recorder.Recorder(*args, **kwargs)[source]

Will save the incoming data to a file (default .csv).

Important

Can only take ONE input, i.e. save labels from only one block. If you want multiple readings in a single file use the Multiplex block.

__init__(filename: str, delay: float = 2, labels: Union[str, list] = 't(s)') → None[source]

Sets the args and initializes the parent class.

Parameters

filename (str) – Path and name of the output file. If the folders do not exist, they will be created. If the file already exists, the actual file will be named with a trailing number to avoid overriding it.
delay (float, optional) – Delay between each write in seconds.
labels (list, optional) – What labels to save. Can be either a str to save all labels but this one first, or a list to save only these labels.

begin() → None[source]: This is meant to receive data once and adapt the label list.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare() → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.

class crappy.blocks.recorder.Saver(*args, **kwargs)[source]

Sink

class crappy.blocks.sink.Sink(*args, **kwargs)[source]: Test block used to get data and do nothing.

UController

class crappy.blocks.ucontroller.UController(*args, **kwargs)[source]

Block for interfacing over serial with an external device, written mostly for communication with microcontrollers.

It can send labeled commands to the device, and/or receive labeled data from it. This block is meant to be used along with the MicroController.py MicroPython template located in the tool folder of Crappy, even though it is not mandatory. A given syntax needs to be followed for any data to be exchanged.

__init__(labels: Optional[list] = None, cmd_labels: Optional[list] = None, init_output: Optional[dict] = None, post_process: Optional[dict] = None, t_device: bool = False, port: str = '/dev/ttyUSB0', baudrate: int = 115200, verbose: bool = False, freq: float = 100) → None[source]

Checks the validity of the arguments.

Parameters

labels (list, optional) – The list of the labels to get from the device. Only these labels should be given as argument to the send_to_pc() method in the MicroPython script. If this argument is not None, then the``init_output`` argument should be given as well. No more than 9 labels should be given.
cmd_labels (list, optional) – The list of the command labels that will be sent to the device upon reception from an upstream block. The variables in the MicroPython script should have these exact names. No more than 9 cmd_labels should be given.
init_output (dict, optional) – If the labels argument is not None, the values to output to downstream blocks for each label as long as no value has been received from the device. An initial output value must be given for each label.
post_process (dict, optional) – Optionally allows applying a function to the data of a label before transmitting it to downstream blocks. It is possible to give functions for only part of the labels.
t_device (bool, optional) – It True, the timestamp returned under the label ‘t(s)’ is the one of the device, not the one of Crappy. It may reduce the maximum achievable sample rate, as more bytes have to be transmitted, but it is also far more precise.
port (str, optional) – The serial port to open for communicating with the device. In Windows, they are usually called COMx, whereas in Linux and Mac they’re called /dev/ttyxxxx.
baudrate (int, optional) – The baudrate for serial communication. It depends on the capabilities of the device.
verbose (bool, optional) – If True, prints debugging information.
freq (float, optional) – The looping frequency of the block.

finish() → None[source]: Closes the serial port, and sends a ‘stop!’ message to the device.

loop() → None[source]: First sends the commands from upstream blocks to the device, then reads the data from the device and sends it to the downstream blocks.

Important

The precision of the commands sent to the device is limited to 3 digits after the decimal point, to limit the traffic on the bus. Adapt the range of the command values consequently.

Note

Commands are sent as text, because some boards cannot read bytes from the stdin buffer in MicroPython. Data is however received on the PC from the device as bytes.

prepare() → None[source]

Opens the serial port, and sends a ‘go’ message to the device.

Also shares with the device two tables associating each cmd_label and label with an integer. This allows reducing the traffic on the serial bus.

Note

The commands are sent as text because some boards cannot read bytes from the stdin buffer in MicroPython.

VideoExtenso

class crappy.blocks.videoExtenso.Video_extenso(*args, **kwargs)[source]

Measures the deformation from video by tracking dots on the sample.

This requires the user to select the ROI to make the spot detection. Once done, it will return the deformation (in %) along X and Y axis. Optionally, it can save images.

It also returns a list of :obj`tuple`, which are the coordinates (in pixel) of the barycenter of the spots.

The initial length is reset when receiving data from a parent block.

__init__(camera: str, save_folder: Optional[str] = None, verbose: bool = False, labels: Optional[list] = None, fps_label: bool = False, img_name: str = '{self.loops:06d}_{t-self.t0:.6f}', ext: str = 'tiff', save_period: int = 1, save_backend: Optional[str] = None, transform: Optional[Callable] = None, input_label: Optional[str] = None, config: bool = False, show_image: bool = False, wait_l0: bool = False, end: bool = True, white_spots: bool = False, update_thresh: bool = False, num_spots: Union[str, int] = 'auto', safe_mode: bool = False, border: int = 5, min_area: float = 150, blur: float = 5, **kwargs) → None[source]

Sets the args and initializes the camera.

Parameters

camera (str) – The name of the camera to control. See Cameras for an exhaustive list of available ones.
save_folder (str, optional) – The directory to save images to. If it doesn’t exist it will be created. If None the images won’t be saved.
verbose (bool, optional) – If True, the block will print the number of loops/s.
labels (list, optional) – Names of the labels for the output.
fps_label (str, optional) – If set, self.max_fps will be set to the value received by the block with this label.
img_name (str, optional) – Template for the name of the image to save. It is evaluated as an f-string.
ext (str, optional) – Extension of the image. Make sure it is supported by the saving backend.
save_period (int, optional) – Will save only one in x images.
save_backend (str, optional) – Module to use to save the images. The supported backends are: sitk (SimpleITK), cv2 (OpenCV) and pil (Pillow). If None, will try sitk and then cv2 if not successful.
transform (function, optional) – Function to be applied on the image before sending. It will not be applied on the saved images.
input_label (str, optional) – If specified, the image will not be read from a camera object but from this label.
config (bool, optional) – Show the popup for config ?
show_image –
wait_l0 (bool, optional) – If set to True, the block sends only zeros until the initial length is reset by receiving data from an input.
end (bool, optional) – If True, the block will stop the Crappy program when the spots are lost, else it will just stop sending data.
white_spots – Set to True if the spots are lighter than the surroundings, else set to False.
update_thresh – Should the threshold be updated in each round ? If so there are lower chances to lose the spots but there will be more noise in the measurement.
num_spots –
The number of spots to detect. Helps for spot detection and allows to force detection of a given number of spots (“auto” works fine most of the time). Can be set to:
```
"auto", 2, 3, 4
```
safe_mode – If set to False, it will try hard to catch the spots when losing them. Could result in incoherent values without crash. Set to True when security is a concern.
border – The number of pixels that will be added to the limits of the boundingbox.
min_area – Filters regions with an area smaller than this value among the selected regions.
blur – Median blur to be added to the image to smooth out irregularities and make detection more reliable.
**kwargs – Any additional specific argument to pass to the camera.

finish() → None[source]: If main() is not overridden, this method will be called upon exit or after a crash.

prepare(*_, **__) → None[source]

This will be run when creating the process, but before the actual start.

The first code to be run in the new process, will only be called once and before the actual start of the main launch of the blocks.

It can remain empty and do nothing.