Blocks

Regular Blocks

Auto Drive

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

This Block is meant to drive an Actuator on which a Camera performing video-extensometry is mounted, so that the spots stay centered on the image.

It takes the output of a VideoExtenso Block and uses the coordinates of the spots to drive the Actuator. The Actuator can only be driven in speed, not in position. The label carrying the coordinates of the tracked spots must be 'Coord(px)'.

It also outputs the difference between the center of the image and the middle of the spots, along with a timestamp, over the 't(s)' and 'diff(pix)' labels. It can then be used by downstream Blocks.

New in version 1.4.0.

Changed in version 2.0.0: renamed from AutoDrive to AutoDriveVideoExtenso

__init__(actuator: Dict[str, Any], gain: float = 2000, direction: str = 'Y-', pixel_range: int = 2048, max_speed: float = 200000, ft232h_ser_num: str | None = None, freq: float | None = 200, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

actuator – A dict for initializing the Actuator to drive. Unlike for the Machine Block, only the 'type' key is mandatory here. All the other keys will be considered as kwargs to pass to the Actuator.
gain –
The gain for driving the Actuator in speed. The speed command is simply the difference in pixels between the center of the image and the center of the spots, multiplied by this gain.

Changed in version 1.5.10: renamed from P to gain
direction – Indicates which axis to consider for driving the Actuator, and whether the action should be inverted. The first character is the axis (X or Y) and second character is the inversion (+ or -). The inversion depends on whether a positive speed will bring the spots closer or farther.
pixel_range –
The size of the image (in pixels) along the chosen axis.

Changed in version 1.5.10: renamed from range to pixel_range
max_speed – The absolute maximum speed value that can be sent to the Actuator.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block.

New in version 2.0.0.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None[source]: Checks the consistency of the linking and initializes the Actuator to drive.

loop() → None[source]: Receives the latest data from the VideoExtenso Block, calculates the center coordinate in the chosen direction, and sets the Actuator speed accordingly.

finish() → None[source]: Stops the Actuator and closes it.

Button

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

This Block allows the user to send a signal to downstream Blocks upon clicking on a button in a Graphical User Interface.

It sends an integer value, that starts from 0 and is incremented every time the user clicks on the button. This Block relies on a Tk window for the graphical interface.

This Block is mostly useful for incorporating user feedback in a script, i.e. triggering actions based on an experimenter’s decision. It can be handy for taking pictures at precise moments, or when an action should only begin after the experimenter has completed a task, for example.

New in version 1.4.0.

Changed in version 2.0.0: renamed from GUI to Button

__init__(send_0: bool = False, label: str = 'step', time_label: str = 't(s)', freq: float | None = 50, spam: bool = False, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

send_0 –
If True, the value 0 will be sent automatically when starting the Block. Otherwise, 1 will be sent at the first click. Only relevant when spam is False.

New in version 1.5.10.
label – The label carrying the information on the number of clicks, default is 'step'.
time_label –
The label carrying the time information, default is 't(s)'.

New in version 1.5.10.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
spam – If True, sends the current step value at each loop, otherwise only sends it at each click.
display_freq –
If True, displays the looping frequency of the Block.

New in version 2.0.0.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None[source]: Creates the graphical interface and sets its layout and callbacks.

begin() → None[source]: Sends the value of the first step (0) if required.

loop() → None[source]: Updates the interface, and sends the current step value if spam is True.

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

Camera Block

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

This Block can drive a Camera object. It can acquire images, display them and record them. It can only drive one Camera at once.

It takes no input Link in a majority of situations, and usually doesn’t have output Links neither. The only situations when it can accept input Links is when an image_generator is defined, or when defining a software_trig_label. If save_images is set to True, and if an output Link is present, a message is sent to downstream Blocks at each saved image, containing the timestamp, the index, and the metadata of the image. They are respectively carried by the ‘t(s)’, ‘img_index’ and ‘meta’ labels. This is useful for performing an action conditionally at each new saved image.

Most of the time, this Block is used for recording to the desired location the images it acquires. Optionally, the images can also be displayed in a dedicated window. Both of these features are however optional, and it is possible to acquire images and not do anything with them. Several options are available for tuning the record and the display.

Before a test starts, this Block can also display a CameraConfig window in which the user can visualize the acquired images, and interactively tune all the CameraSetting available for the instantiated Camera.

Internally, this Block is only in charge of the image acquisition, and the other tasks are parallelized and delegated to CameraProcess objects. The display is handled by the Displayer, and the recording by the ImageSaver. This Block manages the instantiation, the synchronisation and the termination of all the CameraProcess it controls.

New in version 1.4.0.

__init__(camera: str, transform: Callable[[ndarray], ndarray] | None = None, config: bool = True, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, display_freq: bool = False, debug: bool | None = False, freq: float | None = 200, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, img_shape: Tuple[int, int] | Tuple[int, int, int] | None = None, img_dtype: str | None = None, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
config –
If True, a CameraConfig window is displayed before the test starts. There, the user can interactively adjust the different CameraSetting available for the selected Camera, and visualize the acquired images. The test starts when closing the configuration window. If not enabled, the img_dtype and img_shape arguments must be provided.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
display_freq –
If True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of this Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
img_shape –
The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
img_dtype –
The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

New in version 1.5.2: no_loop argument

Removed in version 1.5.10: fps_label, ext, input_label and no_loop arguments

Removed in version 2.0.0: img_name argument

prepare() → None[source]

Preparing the save folder, opening the camera and displaying the configuration GUI.

This method calls the crappy.camera.Camera.open() method of the Camera object.

begin() → None[source]

This method waits for all the CameraProcess to be ready, then releases them all at once to make sure they’re synchronized.

A Barrier is used for forcing the CameraProcesses to wait for each other.

loop() → None[source]

This method receives data from upstream Blocks, acquires a frame from the Camera object, and transmits it to all the CameraProcess.

The image is acquired by calling the get_image() method of the Camera object. If only a timestamp is returned by this method, and not a complete dict of metadata, some basic metadata is generated here and transmitted to the CameraProcesses.

This method also manages the software trigger if this option was set, applies the image transformation function if one was given, and displays the FPS of the acquisition if required.

finish() → None[source]

This method stops the image acquisition on the Camera, as well as all the CameraProcess that were started.

If the CameraProcesses do not gently stop, they are terminated. Also stops the Manager in charge of handling the metadata.

For stopping the image acquisition, the close() method is called.

Canvas

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

This Block allows displaying a real-time visual representation of data.

It displays the data on top of a background image and updates it according to the values received through the incoming Link. The background image and the data overlay are displayed in a new window.

It is possible to display a simple text, a time counter, or text associated with a color dot evolving depending on a predefined color bar and the received values.

This Block is mostly useful for displaying a user-friendly and fine-tuned representation of data. For simpler displays, the Dashboard, Grapher and LinkReader Blocks should be preferred.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Drawing to Canvas

__init__(image_path: str, draw: Iterable[Dict[str, Any]] | None = None, color_range: Tuple[float, float] = (20, 300), title: str = 'Canvas', window_size: Tuple[int, int] = (7, 5), backend: str = 'TkAgg', freq: float | None = 2, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

image_path – Path to the image that will be the background of the canvas, as a str.
draw – An iterable (like a list or a tuple) of dict defining what to draw. See below for more details.
color_range –
A tuple containing the lowest and highest values for the color bar.

Changed in version 1.5.10: renamed from crange to color_range
title – The title of the window containing the drawing.
window_size – The x and y dimension of the window, following matplotlib nomenclature.
backend – The matplotlib backend to use.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

Note

Information about the draw keys:
- type: Mandatory, the type of drawing to display. It can be either ‘text’, ‘dot_text’ or ‘time’.
- coord: Mandatory, a tuple containing the x and y coordinates where the element should be displayed on the drawing.
- text: Mandatory for ‘text’ and ‘dot_text’ only, the text to display on the drawing. It must follow the %-formatting, and contain exactly one %-field. Ex: ‘T0 = %f’. This field will be updated using the value carried by label.
- label: Mandatory for ‘text’ and ‘dot_text’ only, the label of the data to display. It will try to retrieve this data in the incoming Links. The text will then be updated with this data.

prepare() → None[source]: Initializes the different elements of the drawing.

loop() → None[source]: Receives the latest data from upstream Blocks and updates the drawing accordingly.

finish() → None[source]: Closes the window containing the drawing.

Client Server

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

This Block can exchange data on a local network using the MQTT protocol.

It communicates with an MQTT broker, from which it can receive data by subscribing to topics, and to which it can send data by publishing in topics. This Block can also manage the execution of a Mosquitto broker, so that the broker doesn’t need to be manually started before running the Crappy script.

This Block is intended for communication with remote devices, that may run Crappy scripts or other scripts handling data with the correct syntax. Potential uses include acquisition from battery-powered devices (e.g. microcontrollers) acquiring data in enclosed areas or rotating parts, data transfer between machines to delegate processing, communication with other programs on a same machine, etc.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Client_server to ClientServer

Checks the validity of the arguments and sets the instance attributes.

Parameters:

broker – 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 – The network address on which the MQTT broker is running, as a str.
port – A network port on which the MQTT broker is listening, as an int.
init_output – A dict containing for each label in topics the first value to be sent in the output Links. Should be given in case the data comes from several sources and data for all labels may not be available during the first loops. Must also be given is spam is set to True.
topics – An iterable (like a list or a tuple) containing str and/or iterables of str. Each string corresponds to the name of a label in Crappy. Each element in the iterable (string or iterable of strings) 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 (just the given string or each string in the iterable) the corresponding data from the message. It also returns the current timestamp in the label ‘t(s)’.
cmd_labels – An iterable (like a list or a tuple) containing str and/or iterables of str. Each string corresponds to the name of a label in Crappy. Each element in the iterable (string or iterable of strings) 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 iterable) 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 – An iterable (like a list or a tuple) containing str and/or iterables 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.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
spam –
If True, sends the last received values at each loop even if no new values were received from the broker. When set to True, the init_output must be provided.

New in version 1.5.10.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

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 iterables: Single-value iterables 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',)

prepare() → None[source]: Starts the broker and connects to 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.

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

Dashboard

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

This Block generates an interface displaying data as text in a dedicated window.

It relies on a Tk window for the graphical interface.

In the window, the left column contains the names of the labels to display and the right column contains the latest received values for these labels. For each label, only the last value is therefore displayed.

This Block provides a nicer display than the raw LinkReader Block. For displaying the evolution of a label over time, the Grapher Block should be used instead.

New in version 1.4.0.

__init__(labels: str | Iterable[str], nb_digits: int = 2, display_freq: bool = False, freq: float | None = 30, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

labels – Only the data from these labels will be displayed on the window.
nb_digits – Number of decimals to show.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.7.

Changed in version 2.0.0: renamed from verbose to display_freq
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.7.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None[source]: Checks that there’s at least one incoming Link, and starts the GUI.

New in version 1.5.7.

loop() → None[source]: Receives the data from the incoming Link and displays it.

New in version 1.5.7.

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

New in version 1.5.7.

DIS Correl

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

This Block can perform Dense Inverse Search on a sub-frame (patch) of images acquired by a Camera object, and project the result on various fields.

It is mostly used for computing the displacement and the strain over the given patch, but other fields are also available. refer to the fields and labels arguments for more details. It relies on OpenCV’s DISFlow algorithm, and offers the possibility to adjust many of its settings.

This Block takes no input Link in a majority of situations, and outputs the results of image correlation. It is a subclass of the Camera Block, and inherits of all its features. That includes the possibility to record and to display images in real-time, simultaneously to the image acquisition and processing. Refer to the documentation of the Camera Block for more information on these features.

This Block is very similar to the GPUCorrel Block, except this latter uses GPU-acceleration to perform the image correlation and does not use DISFlow. The DICVE Block also relies on image correlation for computing the displacement and strain on images, but it tracks multiple patches and uses video-extensometry.

Similar to the CameraConfig window that can be displayed by the Camera Block, this Block can display a DISCorrelConfig window before the test starts. Here, the user can also select the patch to track if it was not already specified as an argument.

New in version 1.4.0.

__init__(camera: str, transform: Callable[[ndarray], ndarray] | None = None, config: bool = True, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, display_freq: bool = False, freq: float | None = 200, debug: bool | None = False, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, img_shape: Tuple[int, int] | None = None, img_dtype: str | None = None, patch: Tuple[int, int, int, int] | None = None, fields: str | Iterable[str] | None = None, labels: str | Iterable[str] | None = None, alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, init: bool = True, patch_size: int = 8, patch_stride: int = 3, residual: bool = False, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
config –
If True, a DISCorrelConfig window is displayed before the test starts. There, the user can interactively adjust the different CameraSetting available for the selected Camera, visualize the acquired images, and select the patch to track if it hasn’t been given in the patch argument. The test starts when closing the configuration window. If not enabled, the img_dtype, img_shape and patch arguments must be provided.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format. In addition to the acquired frames, the tracked patch is also displayed on the image as an overlay.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
display_freq –
If True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of the Camera Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
img_shape –
The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
img_dtype –
The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
patch –
The coordinates of the patch to track, as a tuple of exactly 4 int. These integers correspond to the y position of the top-left corner of the patch, the x position of the top-left corner of the patch, the height of the patch, and the width of the patch. Only one patch can be tracked. This argument must be provided if config is False.

New in version 2.0.0.
fields –
The several fields to calculate on the acquired images. They should be given as an iterable containing str. Each string represents one field to calculate, so the more fields are given the heavier the computation is. The possible fields are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
For each field, one single value is computed, corresponding to the norm of the field values over the patch area. If not provided, the default fields are 'x', 'y', 'exx', 'eyy'.
labels – The labels to use for sending data to downstream Blocks. If not given, the default labels are 't(s)', 'meta', 'x(pix)', 'y(pix)', 'Exx(%)', 'Eyy(%)'. They carry for each image its timestamp, a dict containing its metadata, and then for each field the computed value as a float. These default labels are compatible with the default fields, but must be changed if the number of fields changes. When setting this argument, make sure to give at least 2 labels for the time and the metadata, and one label per field. The 'res' label containing the residuals if residual is True should not be included here, it will be automatically added.
alpha – Weight of the smoothness term in DISFlow, as a float.
delta – Weight of the color constancy term in DISFlow, as a float.
gamma – Weight of the gradient constancy term in DISFlow , as a float.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DISFlow, as an int. Zero level corresponds to the original image resolution. The final flow is obtained by bilinear upscaling.
iterations – The number of fixed point iterations of variational refinement per scale in DISFlow, as an int. Set to zero to disable variational refinement completely. Higher values will typically result in more smooth and high-quality flow.
gradient_iterations –
The maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. Higher values may improve the quality.

Changed in version 1.5.10: renamed from gditerations to gradient_iterations
init – If True, the last calculated optical flow is used for initializing the calculation of the next one.
patch_size – The size of an image patch for matching in DISFlow, in pixels as an int.
patch_stride – The stride between two neighbor patches in DISFlow, in pixels as an int. Must be less than the patch_size. Lower values correspond to higher flow quality.
residual – If True, the residuals of the optical flow calculation are computed for each image. They are then returned under the 'res' label, that should not be included in the given labels. This option is mainly intended as a debug feature, to monitor the quality of the image correlation.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

New in version 1.5.10: img_name argument

Removed in version 1.5.10: residual_full argument

Removed in version 2.0.0: img_name argument

prepare() → None[source]

This method mostly calls the prepare() method of the parent class.

In addition to that it instantiates the DISCorrelProcess object that performs the image correlation and the tracking.

Changed in version 1.5.5: now accepting args and kwargs

Changed in version 1.5.10: not accepting arguments anymore

DIC VE

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

This Block can perform video-extensometry on images acquired by a Camera object, by tracking patches using Digital Image Correlation techniques.

It takes no input Link in a majority of situations, and outputs the results of the video-extensometry. It is a subclass of the Camera Block, and inherits of all its features. That includes the possibility to record and to display images in real-time, simultaneously to the image acquisition and processing. Refer to the documentation of the Camera Block for more information on these features.

This Block is quite similar to the VideoExtenso Block, except this latter tracks spots instead of patches with a texture. Both Blocks output similar information, although the default labels and the data format are slightly different. The DISCorrel Block also relies on image correlation techniques for estimating the strain and the displacement on acquired images, but it only performs correlation on a single patch and is designed to have a much greater accuracy on this single patch. The GPUVE Block also performs video-extensometry based on digital image correlation, but the correlation is GPU-accelerated. The algorithm used for the correlation is also different from the ones available in this Block.

For tracking the provided patches, several image correlation techniques are available. The most effective one is DISFlow, for which many parameters can be tuned. The other techniques are lighter on the CPU but also less precise. For each image, several values are computed and sent to the downstream Blocks. See the labels argument for a complete list.

Similar to the CameraConfig window that can be displayed by the Camera Block, this Block can display a DICVEConfig window before the test starts. Here, the user can also select the patches to track if they were not already specified as an argument.

New in version 1.4.0.

Changed in version 2.0.0: renamed from DISVE to DICVE

__init__(camera: str, transform: Callable[[ndarray], ndarray] | None = None, config: bool = True, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, display_freq: bool = False, freq: float | None = 200, debug: bool | None = False, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, img_shape: Tuple[int, int] | None = None, img_dtype: str | None = None, patches: Iterable[Tuple[int, int, int, int]] | None = None, labels: str | Iterable[str] | None = None, method: str = 'Disflow', alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, patch_size: int = 8, patch_stride: int = 3, border: float = 0.2, safe: bool = True, follow: bool = True, raise_on_patch_exit: bool = True, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
config –
If True, a DICVEConfig window is displayed before the test starts. There, the user can interactively adjust the different CameraSetting available for the selected Camera, visualize the acquired images, and select the patches to track if they haven’t been given in the patches argument. The test starts when closing the configuration window. If not enabled, the img_dtype, img_shape and patches arguments must be provided.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format. In addition to the acquired frames, the tracked patches are also displayed on the image as an overlay.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
display_freq –
If True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of the Camera Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
img_shape –
The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
img_dtype –
The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'. This argument is mandatory in case config is False. It is otherwise ignored.

New in version 2.0.0.
patches – The coordinates of the several patches to track, as an iterable (like a list or a tuple) containing one or several tuple of exactly int values. These integers correspond to the y position of the top-left corner of the patch, the x position of the top-left corner of the patch, the height of the patch, and the width of the patch. Up to 4 patches can be given and tracked. This argument must be provided if config is False.
labels – The labels to use for sending data to downstream Blocks. If not given, the default labels are 't(s)', 'meta', 'Coord(px)', 'Eyy(%)', 'Exx(%)', 'Disp(px)'. They carry for each image its timestamp, a dict containing its metadata, a list containing for each patch the coordinates of its center in a tuple of int, the y and x strain values calculated from the displacement and the initial position of the patches, and finally a list containing for each patch its displacement in the y and x direction in a tuple of int. If different labels are desired, they should all be provided at once in an iterable of str containing the correct number of labels (6).
method –
The method to use for performing the digital image correlation. Should be one of :
```
`Disflow`, 'Pixel precision', 'Parabola', 'Lucas Kanade'
```
'Disflow' uses OpenCV’s DISOpticalFlow and 'Lucas Kanade' uses OpenCV’s calcOpticalFlowPyrLK, while all other methods are based on a basic cross-correlation in the Fourier domain. 'Pixel precision' calculates the displacement by getting the position of the maximum of the cross-correlation, and has thus a 1-pixel resolution. It is mainly meant for debugging. 'Parabola' refines the result of 'Pixel precision' by interpolating the neighborhood of the maximum, and has thus a sub-pixel resolution.

New in version 1.5.9.
alpha – Weight of the smoothness term in DISFlow, as a float. Ignored if method is not 'Disflow'.
delta – Weight of the color constancy term in DISFlow, as a float. Ignored if method is not 'Disflow'.
gamma – Weight of the gradient constancy term in DISFlow , as a float. Ignored if method is not 'Disflow'.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DISFlow, as an int. Zero level corresponds to the original image resolution. The final flow is obtained by bilinear upscaling. Ignored if method is not 'Disflow'.
iterations – The number of fixed point iterations of variational refinement per scale in DISFlow, as an int. Set to zero to disable variational refinement completely. Higher values will typically result in more smooth and high-quality flow. Ignored if method is not 'Disflow'.
gradient_iterations –
The maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. Higher values may improve the quality. Ignored if method is not 'Disflow'.

Changed in version 1.5.10: renamed from gditerations to gradient_iterations
patch_size – The size of an image patch for matching in DISFlow, in pixels as an int. Ignored if method is not 'Disflow'.
patch_stride – The stride between two neighbor patches in DISFlow, in pixels as an int. Must be less than the patch_size. Lower values correspond to higher flow quality. Ignored if method is not 'Disflow'.
border – The ratio of the patch that is kept for calculating the displacement, if method is 'Disflow'. For example if a value of 0.2 is given, only the center 80% of the image is used for calculating the average displacement, in both directions. Ignored if method is not 'Disflow'.
safe –
If True, checks at each new image if the patches are not exiting the frame. Otherwise, the patches might exit the image which can lead to an unexpected behavior without raising an error.

New in version 1.5.7.
follow –
If True, the position of each patch on the images is adjusted at each new image based on the previous computed displacement of this patch. If a displacement of 1 in the x direction was calculated on the previous image, and the patch is located at position (x0, y0), the patch will be moved to position (x0 + 1, y0) for the next image. It “follows” the texture to track. Recommended if the expected displacement in pixels is big compared to the patch size. The only downside is that the patches may exit the frame if something goes wrong with the tracking.

New in version 1.5.7.
raise_on_patch_exit –
If True, raises an exception when a tracked patch exits the border of the image, which stops the entire test. Otherwise, just logs a warning message and sleeps until the test is stopped in another way.

New in version 2.0.0.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

Removed in version 1.5.9: fields argument

Removed in version 2.0.0: img_name argument

prepare() → None[source]

This method mostly calls the prepare() method of the parent class.

In addition to that it instantiates the DICVEProcess object that performs the image correlation and the tracking.

Changed in version 1.5.5: now accepting args and kwargs

Changed in version 1.5.10: not accepting arguments anymore

Fake Machine

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

This Block emulates the behavior of a tensile test machine.

It can emulate tensile tests, not compression tests. By default, it assumes an elasto-plastic behavior of the tested sample. The main mechanical parameters of the material are tunable.

This Block is meant to be driven like a Machine Block. It receives speed or position commands from upstream Blocks, and modifies the behavior of the emulated machine accordingly. Its outputs are however different from the Machine Blocks, as it outputs the current force, position, and strain of the emulated tensile test machine. The labels carrying this data are : t(s), F(N), x(mm), Exx(%), Eyy(%).

This Block was originally designed for proposing examples that do not require any hardware to run, but still display the possibilities of Crappy. It can also be used to test a script without actually interacting with hardware.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Fake_machine to FakeMachine

__init__(rigidity: float = 8400000.0, l0: float = 200, max_strain: float = 1.51, sigma: ~typing.Dict[str, float] | None = None, nu: float = 0.3, plastic_law: ~typing.Callable[[float], float] = <function plastic>, max_speed: float = 5, mode: str = 'speed', cmd_label: str = 'cmd', freq: float | None = 100, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

rigidity –
The rigidity of the material, in N, so that \(force = rigidity * strain\).

Changed in version 2.0.0: renamed from k to rigidity
l0 – The initial length of the fake sample to test, in mm.
max_strain –
The maximum strain the material can withstand before breaking.

Changed in version 1.5.10: renamed from maxstrain to max_strain
mode – Whether the command sent to the fake machine is a speed or a position command. Can be 'speed' or 'position'.
plastic_law – A callable taking the maximum reached strain and returning the proportion of the current strain caused by plastic deformation.
sigma – A dict containing for each label the standard deviation for adding noise to the signal. Can be given for part or all of the labels. The deviation should be given not normalized, in the same unit as the label to which it applies.
nu – Poisson’s ratio of the material.
cmd_label – The label carrying the command of the fake machine.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

begin() → None[source]: Sends a first value that should be 0.

loop() → None[source]: Receives the latest command value, calculates the new speed and position from it, checks whether the sample broke and what the plastic elongation is, and finally returns the data.

prepare() → None

This method should perform any action required for initializing the Block before the test starts.

For example, it can open a network connection, create a file, etc. It is also fine for this method not to be overriden if there’s no particular action to perform.

Note that this method is called once the Process associated to the Block has been started.

Generator

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

This Block generates a signal following a user-defined assembly of Path.

The generated signal is just a waveform that can serve any purpose. It can for example be used for driving a Machine Block, or for triggering a Camera Block.

One Generator Block can only generate one signal. Use multiple Blocks if several signals are needed. Note that the default behavior of a Generator is to stop the entire script when it reaches the end of all the Paths.

This Block can also accept inputs from other Blocks, as these inputs may be used by a Path. The most common use of this feature is to have the stop condition of a Path depend on the received values of a label.

New in version 1.4.0.

__init__(path: Iterable[Dict[str, Any]], freq: float | None = 200, cmd_label: str = 'cmd', path_index_label: str = 'index', repeat: bool = False, spam: bool = False, display_freq: bool = False, end_delay: float | None = 2, safe_start: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

path – An iterable (like a list or a tuple) of dict, each dict providing the parameters to generate a Path. The Paths are generated in the order in which they are given, and the stop condition of each Path is used for determining when to switch to the next one. The 'type' key of each dict gives the name of the Path to use, and all the other keys correspond to the arguments to give to this Path. Refer to the documentation of the chosen Paths to know which keys to provide.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
cmd_label – The label of the signal sent to the downstream Blocks.
path_index_label – In addition to the cmd_label, this label holds the index of the current Path. Useful to trigger a Block when the current Path changes, as the output value might not necessarily change.
repeat – If True, the path will loop forever instead of stopping when it reaches the last Path.
spam – If True, the signal value will be sent on each loop. Else, it will only be sent if it is different from the previous or if the Block switched to the next Path.
display_freq –
if True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
end_delay – When all the Paths are exhausted, waits this many seconds before stopping the entire script. Can be set to None, in which case the Generator won’t stop the program when finishing.
safe_start –
Ensures the first Path waits for at least one data point from upstream Blocks before sending the first value of the signal. Otherwise, the first value might be sent without checking the associated condition if its depends on labels from other Blocks.

New in version 1.5.10.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

begin() → None[source]: Initializes the first Path.

loop() → None[source]

First reads data from upstream Blocks, then gets the next command to send, and finally sends it to downstream Blocks.

It also manages the transitions between the Path.

prepare() → None

This method should perform any action required for initializing the Block before the test starts.

For example, it can open a network connection, create a file, etc. It is also fine for this method not to be overriden if there’s no particular action to perform.

Note that this method is called once the Process associated to the Block has been started.

GPU Correl

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

This Block can perform GPU-accelerated image correlation on images acquired by a Camera object, and project the result on various fields.

It is mostly used for computing the displacement and the strain over the given patch, but other fields are also available. refer to the fields and labels arguments for more details.

This Block takes no input Link in a majority of situations, and outputs the results of image correlation. It is a subclass of the Camera Block, and inherits of all its features. That includes the possibility to record and to display images in real-time, simultaneously to the image acquisition and processing. Refer to the documentation of the Camera Block for more information on these features.

This Block is very similar to the DISCorrel Block, except this latter uses DISFlow to perform the image correlation and is not GPU-accelerated. The GPUVE Block also relies on GPU-accelerated image correlation for computing the displacement and strain on images, but it tracks multiple patches and uses video-extensometry.

No Region Of Interest can be specified to this Block, so by default the correlation is performed on the entire image. It is however possible to set a mask, so that only part of the image is considered when running the correlation.

Warning

This Block cannot run with CUDA versions greater than 11.3 ! This is due to a deprecation in pycuda, and is unlikely to be fixed anytime soon in Crappy or pycuda.

New in version 1.4.0.

__init__(camera: str, fields: str | Iterable[str], img_shape: Tuple[int, int], img_dtype: str, transform: Callable[[ndarray], ndarray] | None = None, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, verbose: int = 0, freq: float | None = 200, debug: bool | None = False, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: str | Iterable[str] | None = None, discard_limit: float = 3, discard_ref: int = 5, img_ref: ndarray | None = None, levels: int = 5, resampling_factor: float = 2, kernel_file: str | Path | None = None, iterations: int = 4, mask: ndarray | None = None, mul: float = 3, res: bool = False, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
fields –
The several fields to calculate on the acquired images. They should be given as an iterable containing str. Each string represents one field to calculate, so the more fields are given the heavier the computation is. The possible fields are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
For each field, one single value is computed, corresponding to the norm of the field values over the patch area.
img_shape – The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape.
img_dtype – The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is displayed, and at level 3 so much information is displayed that it slows the code down. This argument allows to adjust the precision of the log messages, while the debug argument is for enabling or disabling logging.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of the Camera Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
labels – The labels to use for sending data to downstream Blocks. If not given, the default labels are 't(s)', 'meta' followed by the names of the given fields. They carry for each image its timestamp, a dict containing its metadata, and then for each field the computed value as a float. When setting this argument, make sure to give at least 2 labels for the time and the metadata, and one label per field. The 'res' label containing the residuals if res is True should not be included here, it will be automatically added.
discard_limit –
If res is True, the result of the correlation is not sent to the downstream Blocks if the residuals for the current image are greater than discard_limit times the average residual for the last discard_ref images.

Changed in version 1.5.10: renamed from discard_lim to discard_limit
discard_ref – If res is True, the result of the correlation is not sent to the downstream Blocks if the residuals for the current image are greater than discard_limit times the average residual for the last discard_ref images.
img_ref –
A reference image, as a 2D numpy.array with dtype float32. If given, it will be set early on the correlation class and the test can start from the first acquired frame. If not given, the first acquired image will be set as the reference, which will slow down the beginning of the test.

Changed in version 1.5.10: renamed from imgref to img_ref
levels –
Number of levels of the pyramid. More levels may help converging on images with large strain, but may fail on images that don’t contain low spatial frequency. Fewer levels mean that the program runs faster.

New in version 1.5.10.
resampling_factor –
The factor by which the resolution is divided between each stage of the pyramid. A low factor ensures coherence between the stages, but is more computationally intensive. A high factor allows reaching a finer detail level, but may lead to a coherence loss between the stages.

New in version 1.5.10.
kernel_file –
The path to the file containing the kernels to use for the correlation. Can be a pathlib.Path object or a str. If not provided, the default GPU Kernels are used.

New in version 1.5.10.
iterations –
The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing.

New in version 1.5.10.
mask –
The mask used for weighting the region of interest on the image. It is generally used to prevent unexpected behavior on the border of the image. Also allows to select a sub-region of the image if the correlation should not be performed on the entire image, as this Block does not accept a patch argument.

New in version 1.5.10.
mul –
The scalar by which the direction will be multiplied before being added to the solution. If it’s too high, the convergence will be fast but there’s a risk to go past the solution and to diverge. If it’s too low, the convergence will be slower and require more iterations. 3 was found to be an acceptable value in most cases, but it is recommended to tune this value for each application so that the convergence is neither too slow nor too fast.

New in version 1.5.10.
res –
If True, calculates the residuals after performing the correlation and returns the residuals along with the correlation data. The residuals are always returned under the label 'res', and this label should not be included in the labels argument.

New in version 1.5.10.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

Removed in version 1.5.10: fps_label, ext, input_label, config and cam_kwargs arguments

Removed in version 2.0.0: img_name argument

prepare() → None[source]

This method mostly calls the prepare() method of the parent class.

In addition to that it instantiates the GPUCorrelProcess object that performs the GPU-accelerated image correlation.

Changed in version 1.5.5: now accepting args and kwargs

Changed in version 1.5.10: not accepting arguments anymore

GPU VE

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

This Block can perform GPU-accelerated video-extensometry on images acquired by a Camera object, by tracking patches and computing the strain based on their displacement.

It takes no input Link in a majority of situations, and outputs the results of the video-extensometry. It is a subclass of the Camera Block, and inherits of all its features. That includes the possibility to record and to display images in real-time, simultaneously to the image acquisition and processing. Refer to the documentation of the Camera Block for more information on these features.

This Block is quite similar to the DICVE Block, except this latter is not GPU-accelerated and uses OpenCV’s DISFlow. The GPUCorrel Block also relies on GPU-accelerated image correlation for estimating the strain and the displacement on acquired images, but it performs correlation on the entire image and is designed to have a much greater accuracy. The VideoExtenso Block also performs video-extensometry, but it does so by tracking spots instead of textured patches, and it is not GPU-accelerated.

Warning

This Block cannot run with CUDA versions greater than 11.3 ! This is due to a deprecation in pycuda, and is unlikely to be fixed anytime soon in Crappy or pycuda.

New in version 1.4.0.

__init__(camera: str, patches: Iterable[Tuple[int, int, int, int]], img_shape: Tuple[int, int], img_dtype: str, transform: Callable[[ndarray], ndarray] | None = None, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, verbose: bool = False, freq: float | None = 200, debug: bool | None = False, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: str | Iterable[str] | None = None, img_ref: ndarray | None = None, kernel_file: str | Path | None = None, iterations: int = 4, mul: float = 3, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
patches – The coordinates of the several patches to track, as an iterable (like a list or a tuple) containing one or several tuple of exactly int values. These integers correspond to the y position of the top-left corner of the patch, the x position of the top-left corner of the patch, the height of the patch, and the width of the patch. Up to 4 patches can be given and tracked.
img_shape – The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape.
img_dtype – The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format. In addition to the acquired frames, the tracked patches are also displayed on the image as an overlay.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is displayed, and at level 3 so much information is displayed that it slows the code down. This argument allows to adjust the precision of the log messages, while the debug argument is for enabling or disabling logging.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of the Camera Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
labels – The labels to use for sending data to downstream Blocks. If not given, the default labels are 't(s)', 'meta' followed for each given patch by 'p<i>x', 'p<i>y' with '<i>' the number of the patch. They carry for each image its timestamp, a dict containing its metadata, and then for each patch the x and y positions of its centroid. When setting this argument, make sure to give 2 labels for the time and the metadata, and two labels per patch.
img_ref –
A reference image, as a 2D numpy.array with dtype float32. If given, it will be set early on the correlation class and the test can start from the first acquired frame. If not given, the first acquired image will be set as the reference, which will slow down the beginning of the test.

New in version 1.5.10.
kernel_file –
The path to the file containing the kernels to use for the correlation. Can be a pathlib.Path object or a str. If not provided, the default GPU Kernels are used.

New in version 1.5.10.
iterations –
The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing.

New in version 1.5.10.
mul –
The scalar by which the direction will be multiplied before being added to the solution. If it’s too high, the convergence will be fast but there’s a risk to go past the solution and to diverge. If it’s too low, the convergence will be slower and require more iterations. 3 was found to be an acceptable value in most cases, but it is recommended to tune this value for each application so that the convergence is neither too slow nor too fast.

New in version 1.5.10.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

Removed in version 1.5.10: fps_label, ext, input_label, config and cam_kwargs arguments

Removed in version 2.0.0: img_name argument

prepare() → None[source]

This method mostly calls the prepare() method of the parent class.

In addition to that it instantiates the GPUVEProcess object that performs the GPU-accelerated image correlation.

Changed in version 1.5.5: now accepting args and kwargs

Changed in version 1.5.10: not accepting arguments anymore

Grapher

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

This Block can display data in a 2D graph in a persistent way.

The graph is displayed in an independent window. It iis updated each time new data is received from upstream Blocks. The displayed data can come from different Blocks.

The user can choose which labels are plotted on the x and y axes. It is therefore possible to plot a label versus time, or a label versus another label. A single graph is displayed, but multiple curves can be plotted on this graph.

The Grapher Block is known for being very CPU-intensive. For displaying only the last values of given labels, the LinkReader and Dashboard Blocks are simpler solutions that go much easier on the CPU.

New in version 1.4.0.

__init__(*labels: Tuple[str, str], length: int = 0, freq: float | None = 2, max_pt: int = 20000, window_size: Tuple[int, int] = (8, 8), window_pos: Tuple[int, int] | None = None, interp: bool = True, backend: str = 'TkAgg', display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

*labels – Each tuple corresponds to a curve to plot, and should contain two values: the first will be the label of the x values, the second the label of the y values. There’s no limit to the number of curves. Note that all the curves are displayed in a same graph.
length – 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 – The target looping frequency for the Block. If None, loops as fast as possible.
max_pt –
The maximum number of points displayed on the graph. When reaching this limit, the Block deletes one point out of two to avoid using too much memory and CPU.

Changed in version 2.0.0: renamed from maxpt to max_pt
window_size – The size of the graph, in inches.
window_pos – 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 – If True, the data points are linked together by straight lines. Else, only the points are displayed.
backend – The matplotlib backend to use. Performance may vary according to the chosen backend. Also, every backend may not be available depending on your machine.
display_freq –
if True, displays the looping frequency of the Block.

New in version 1.5.6.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

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.

prepare() → None[source]: Configures the figure for displaying data.

loop() → None[source]: Receives the upcoming data, puts in the display buffer and updates the graph.

finish() → None[source]: Closes all the opened matplotlib windows.

HDF Recorder

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

This Block records data efficiently into a HDF5 file.

It expects data as numpy.array from exactly one upstream Block, that should be an IOBlock in streamer mode. It then saves this data in a HDF5 file using the tables module.

This Block is intended for high-speed data recording from InOut in streamer mode. For regular data recording, the Recorder Block should be used instead.

Warning

Corrupted HDF5 files are not readable at all ! If anything goes wrong during a test, especially during the finish phase, it is not guaranteed that the recorded data will be readable.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Hdf_recorder to HDFRecorder

__init__(filename: str | Path, node: str = 'table', expected_rows: int = 100000000, atom=None, label: str = 'stream', metadata: dict | None = None, freq: float | None = None, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

filename – Path to the output file, either relative or absolute. If the parent folders of the file do not exist, they will be created. If the file already exists, the actual file where data will be written will be renamed with a trailing index to avoid overriding it.
node – The name of the array to create in the HDF5 file, as a str.
expected_rows – The number of expected rows in the file. It is used to optimize the dumping.
atom – This represents 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 – The label carrying the data to be saved
metadata – A dict containing additional information to save in the HDF5 file.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
display_freq –
if True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None[source]: Checks that the Block has the right number of inputs, creates the folder containing the file if it doesn’t already exist, changes the name of the file if it already exists, and initializes the HDF file.

loop() → None[source]

Receives data from the upstream Block and saves it.

Also creates the array for recording data when the first values are received.

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

IOBlock

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

This Block is meant to drive InOut objects. It can acquire data, and/or set commands. One IOBlock can only drive a single InOut.

If it has incoming Link, it will set the commands received over the labels given in cmd_labels by calling the set_cmd() method of the InOut. Additional commands to set at the very beginning or the very end of the test can also be specified.

If it has outgoing Link, it will acquire data using the get_data() method of the InOut and send it downstream over the labels given in labels. It is possible to trigger the acquisition using a predefined label.

The streamer argument allows using the “streamer” mode of InOuts supporting it, instead of the regular acquisition mode. Finally, the make_zero_delay argument allows offsetting the acquired values to zero at the beginning of the test. Refer to the documentation of each argument for a more detailed description.

New in version 1.4.0.

Sets the arguments and initializes the parent class.

Parameters:

name – The name of the InOut class to instantiate.
labels – An iterable (e.g. a list or a tuple) containing the output labels for InOuts that acquire data. They correspond to the values returned by the InOut’s get_data() method, so there should be as many labels as returned values, and given in the appropriate order. The first label must always be the time label, preferably called 't(s)'. This argument can be omitted if get_data() returns a dict. Ignored if the Block has no output Link.
cmd_labels – An iterable (e.g. a list or a tuple) containing the labels considered as inputs of this Block, for InOuts that set commands. The values received from these labels will be passed to the InOut’s set_cmd() method, in the same order as the labels are given. Usually, time is not part of the cmd_labels. Ignored if the Block has no input Link.
trigger_label –
If given, the Block will only read data whenever a value is received on this label (can be any value). Ignored if the Block has no output Link. A trigger label can also be a cmd label.

Changed in version 1.5.10: renamed from trigger to trigger_label
streamer – If False, the get_data() method of the InOut is called for acquiring data, else it is the get_stream() method. Refer to the documentation of these methods for more information.
initial_cmd – An initial command for the InOut, set during prepare(). If given, there must be as many values as in cmd_labels. Must be given as an iterable (e.g. a list or a tuple).
exit_cmd –
A final command for the InOut, set during finish(). If given, there must be as many values as in cmd_labels. Must be given as an iterable (e.g. a list or a tuple).

Changed in version 1.5.10: renamed from exit_values to exit_cmd
make_zero_delay –
If set, will acquire data before the beginning of the test and use it to offset all the labels to zero. The data will be acquired during the given number of seconds. Ignored if the Block has no output Links. Does not work for InOuts that acquire values other than numbers (str for example).

New in version 1.5.10.
spam – If False, the Block will call set_cmd() on the InOut object only if the current command is different from the previous. Otherwise, it will call the method each time a command is received.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block while running.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
**kwargs – The arguments to be passed to the InOut.

prepare() → None[source]

Checks the consistency of the Link layout, opens the InOut and sets the initial command if required.

This method mainly calls the open() method of the driven InOut.

loop() → None[source]

Reads data from the InOut and/or sets the received commands.

Data is read from the InOut only if this Block has outgoing Links. If the trigger_label is given, data is read only if a trigger is received over the given trigger label.

A command is set on the InOut only if this Block has incoming Links, and if data is received over these Links. Depending on the value of the spam argument, a command might not be set if it is similar to the previous one.

The data is read from the InOut either by calling its return_data() or its return_stream() method, depending if the streamer argument is True of False. The commands are always set by calling the set_cmd() method.

finish() → None[source]

Stops the stream, sets the exit command if necessary, and closes the InOut.

This method mainly calls the close() method of the driven InOut.

Link Reader

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

This Block reads and displays the data it receives.

It can display data received from any number of Link. For each new received data point, a message is displayed in the console with the received values.

This Block is the most basic way of displaying data in Crappy. The Dashboard Block can be used for a nicer layout, and the Grapher Block should be used for plotting data in a persistent way.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Reader to LinkReader

__init__(name: str | None = None, freq: float | None = 50, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

name –
If set, will be displayed to identify the LinkReader. Otherwise, the block is automatically named based on the number of its instances already running.

Changed in version 1.5.5: renamed from name to reader_name

Changed in version 1.5.5: renamed from reader_name to name
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
display_freq –
if True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

loop() → None[source]: Flushes the incoming Link and displays their data.

Machine

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

This Block is meant to drive one or several Actuator. It can set speed or position commands on hardware actuators.

The possibility to drive several Actuators from a unique Block is given so that they can be driven in a synchronized way. If synchronization is not needed, it is preferable to drive the Actuators from separate Machine Blocks.

This Block takes the speed or position commands for the Actuators as inputs, and can optionally read and output the current speed and/or positions of the Actuators. The speed and position commands are set respectively by calling the set_position() and set_speed() methods of the Actuators, and the current speed and position values are acquired by calling the get_position() and get_speed() methods of the Actuators.

It is possible to tune for each Actuator the label over which it receives its commands, and optionally the labels over which it sends its current speed and/or position. The driving mode (‘speed’ or ‘position’) can also be set independently for each Actuator.

New in version 1.4.0.

__init__(actuators: Iterable[Dict[str, Any]], common: Dict[str, Any] | None = None, time_label: str = 't(s)', ft232h_ser_num: str | None = None, spam: bool = False, freq: float | None = 200, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

actuators – An iterable (like a list or a tuple) of all the Actuator this Block needs to drive. It contains one dict for every Actuator, with mandatory and optional keys. The keys providing information on how to drive the Actuator are listed below. Any other unrecognized key will be passed to the Actuator as argument when instantiating it.
common – The keys of this dict will be common to all the Actuators. If one key conflicts with an existing key for an Actuator, the common one will prevail.
time_label – If reading speed or position from one or more Actuators, the time information will be carried by this label.
ft232h_ser_num –
Serial number of the FT232H device to use for driving the controlled Actuator.

New in version 2.0.0.
spam – If True, a command is sent to the Actuators at each loop of the Block, else it is sent every time a new command is received.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

Note

actuators keys:
- type: The name of the Actuator class to instantiate. This key is mandatory.
- cmd_label: The label carrying the command for driving the Actuator. It defaults to ‘cmd’.
- mode: Can be either ‘speed’ or ‘position’. Either set_speed() or set_position() is called to drive the Actuator, depending on the selected mode. When driven in ‘position’ mode, the speed of the actuator can also be adjusted, see the speed and speed_cmd_label keys. The default mode is ‘speed’.
- speed: If mode is ‘position’, the speed at which the Actuator should move. This speed is passed as second argument to the set_position() method of the Actuator. If the speed_cmd_label key is not specified, this speed will remain the same for the entire test. This key is not mandatory.
- position_label: If given, the Block will return the value of get_position() under this label. This key is not mandatory.
- speed_label: If given, the Block will return the value of get_speed() under this label. This key is not mandatory.
- speed_cmd_label: The label carrying the speed to set when driving in ‘position’ mode. Each time a value is received, the stored speed value is updated. It will also overwrite the speed key if given.

prepare() → None[source]

Checks the validity of the linking and initializes all the Actuator objects to drive.

This method calls the open() method of each Actuator.

loop() → None[source]

Sets the received position and speed commands, and reads the current speed and position from the Actuator.

For each Actuator, a command is set only if a new one was received or if the spam argument is True. It is set using either set_position() or set_speed() depending on the selected driving mode.

For each Actuator, a speed and/or position value is read only if the speed_label and/or the position_label was set. If so, these values are read at each loop and sent to downstream Blocks over the given labels. This is independent of the chosen driving mode. The get_position() and get_speed() are called for acquiring the position and speed values respectively.

finish() → None[source]

Stops and closes all the Actuators to drive.

This method calls the stop() and close() method of each Actuator.

Mean Block

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

This Block can compute the average values of given labels over a given delay.

It can take any number of inputs, provided that they share a common time label. If the same label (except time) is received from several Blocks, it may lead to unexpected results.

The output of this Block is very similar to that of the Mean and MovingAvg Modifiers, but not exactly similar. While these Modifiers calculate the average of a label over a fixed number of data points, the MeanBlock calculates the average of the values received over a given delay. This behavior could, however, also be achieved using a Modifier.

Warning

If the delay for averaging is too short compared with the looping frequency of the upstream Blocks, this Block may not always return the same number of labels ! This can cause errors in downstream Blocks expecting a fixed number of labels.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Mean_block to MeanBlock

__init__(delay: float, time_label: str = 't(s)', out_labels: str | Iterable[str] | None = None, display_freq: bool = False, freq: float | None = 50, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

delay – The averaged data will be sent each delay seconds.
time_label –
The label containing the time information. It must be common to all the incoming Links.

Changed in version 1.5.10: renamed from t_label to time_label
out_labels – An iterable (like a list or a tuple) containing all the labels to average, as str. If not given, all the received labels are averaged and returned. The time label should not be included, as it is already given in time_label. If there is only one label to average, it can be directly given as a str, i.e. not in an iterable.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
freq – The target looping frequency for the Block. If None, loops as fast as possible.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None

This method should perform any action required for initializing the Block before the test starts.

For example, it can open a network connection, create a file, etc. It is also fine for this method not to be overriden if there’s no particular action to perform.

Note that this method is called once the Process associated to the Block has been started.

begin() → None[source]: Initializes the time counter.

New in version 2.0.0.

loop() → None[source]: Receives all available data from the upstream Blocks, averages it and sends it if the time delay is reached.

Multiplexer

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

This Block takes data from upstream Blocks as input and interpolates it to output all the labels in a common time basis.

This Block is very similar to the Synchronizer Block, but the Synchronizer takes the timestamps of a reference label as a time base whereas this one performs the interpolation on a time base independent of the received labels.

It can take any number of inputs, provided that they all share a common time label. It is also possible to choose which labels are considered for multiplexing and which are dropped. The interpolation is performed using the numpy.interp method.

This Block is useful for synchronizing data acquired from different sensors, e.g. to plot a real-time stress-strain curve with position data coming from a Machine Block and force data coming from a IOBlock Block. Multiplexing is however quite resource-consuming, so it is preferable to perform interpolation when post-processing if real-time is not needed.

Note

This Block doesn’t truly output data in real-time as it needs to wait for data from all the upstream Blocks before performing the interpolation. It should only be used with care as an input to a decision-making Block. This is especially true when the upstream Blocks have very different sampling rates.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Multiplex to Multiplexer

__init__(time_label: str = 't(s)', out_labels: str | Iterable[str] | None = None, interp_freq: float = 200, freq: float | None = 50, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

time_label –
The label carrying the time information.

Changed in version 1.5.10: renamed from key to time_label
out_labels –
An iterable (like a list or a tuple) containing the labels to multiplex, except for the time label that is given separately in the time_label argument. The Block also doesn’t output anything until data has been received on all these labels. If left to None, all the received labels are considered. It is recommended to always set this argument ! It is also possible to give this argument as a single str (i.e. not in an iterable), although multiplexing a single label is of limited interest.

New in version 2.0.0.
interp_freq –
The target frequency for performing the interpolation. In the output data, there will be one interpolated data point each \(1 / interp_freq\) seconds. Independent of the freq argument, but it is no use setting freq higher than interp_freq otherwise there will be void loops.

New in version 2.0.0.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.9.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

loop() → None[source]: Receives data, interpolates it, and sends it to the downstream Blocks.

PID

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

This Block implements a basic PID corrector.

It takes a target value and a measured value as an input, and outputs a command value for reaching the target on the controlled system. The command is updated at each loop. The given P, I and D gains can be adjusted at runtime, by sending the new desired values. It is also possible to set boundaries to the output command, or to the integral corrector only.

This Block is generally used in combination with a Generator Block for generating the target value, and a Machine or IOBlock Block driving a physical system. This latter Block takes the command as input, and returns to the PID the measured value to be compared to the target.

New in version 1.4.0.

__init__(kp: float, ki: float = 0, kd: float = 0, out_max: float = inf, out_min: float = -inf, setpoint_label: str = 'cmd', input_label: str = 'V', time_label: str = 't(s)', kp_label: str = 'kp', ki_label: str = 'ki', kd_label: str = 'kd', labels: Tuple[str, str] | None = None, reverse: bool = False, i_limit: Tuple[float | None, float | None] = (None, None), send_terms: bool = False, freq: float | None = 500, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

kp – The initial P gain. It can be tuned while running by sending the new value over the given kp_label. No matter if a positive or a negative value is given, the definitive sign will be set by the reverse argument.
ki – The initial I gain. It can be tuned while running by sending the new value over the given ki_label. No matter if a positive or a negative value is given, the definitive sign will be set by the reverse argument.
kd – The initial D gain. It can be tuned while running by sending the new value over the given kd_label. No matter if a positive or a negative value is given, the definitive sign will be set by the reverse argument.
out_max – Ensures the output is always inferior to this value.
out_min – Ensures the output is always superior to this value.
setpoint_label –
The label carrying the setpoint value.

Changed in version 2.0.0: renamed from target_label to setpoint_label
input_label – The label carrying the reading of the actual value, to be compared with the setpoint.
time_label – The label carrying the time information in the incoming Links.
kp_label –
The label to use for changing the P gain on the fly. If a value is received over this label, it will overwrite the one given in the kp argument.

New in version 2.0.0.
ki_label –
The label to use for changing the I gain on the fly. If a value is received over this label, it will overwrite the one given in the ki argument.

New in version 2.0.0.
kd_label –
The label to use for changing the D gain on the fly. If a value is received over this label, it will overwrite the one given in the kd argument.

New in version 2.0.0.
labels – The two labels that will be sent to downstream Blocks. The first one is the time label, the second one is the output of the PID. If this argument is not given, they default to 't(s)' and 'pid'.
reverse – If True, reverses the action of the PID.
i_limit – A tuple containing respectively the lower and upper boundaries for the I term.
send_terms – If True, returns the weight of each term in the output value. It adds 'p_term', 'i_term', 'd_term' to the output labels. This is particularly useful to tweak the gains.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

loop() → None[source]: Receives the latest target and input values, calculates the P, I and D terms and sends the output to the downstream Blocks.

Recorder

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

This Block saves data from an upstream Block to a text file, with values separated by a coma and lines by a newline character.

The first row of the file contains the names of the saved labels. This Block can only save data coming from exactly one upstream Block. To save data from multiple Blocks, use several instances of Recorder (recommended) or a Multiplexer Block.

This Block cannot directly record data from “streams”, i.e. coming from an IOBlock Block with the 'streamer' argument set to True. To do so, the HDFRecorder Block should be used instead. Alternatively, a Demux Modifier can be placed between the IOBlock and the Recorder, but most of the acquired data won’t be saved.

New in version 1.4.0.

Sets the arguments and initializes the parent class.

Parameters:

file_name –
Path to the output file, either relative or absolute. If the parent folders of the file do not exist, they will be created. If the file already exists, the actual file where data will be written will be renamed with a trailing index to avoid overriding it.

Changed in version 2.0.0: renamed from filename to file_name
delay – Delay between each write in seconds.
labels – If provided, only the data carried by these labels will be saved. Otherwise, all the received data is saved.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

prepare() → None[source]: Checks that the Block has the right number of inputs, creates the folder containing the file if it doesn’t already exist, and changes the name of the file if it already exists.

loop() → None[source]: Receives data from the upstream Block and saves it.

Sink

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

This Block drops all the data it receives, and does nothing else.

It is only useful for debugging, e.g. with Blocks like the IOBlock that have a different behavior when they have output Links.

New in version 1.4.0.

__init__(display_freq: bool = False, freq: float | None = 10, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

display_freq –
If True, displays the looping frequency of the Block.

New in version 1.5.10.

Changed in version 2.0.0: renamed from verbose to display_freq
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

loop() → None[source]: Drops all the received data.

Stop Block

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

This Block parses the data it receives and checks if this data meets the given stop criteria. If so, its stops the test.

Along with the StopButton Block, it allows to stop a test in a clean way without resorting to CTRL+C.

New in version 2.0.0.

Sets the arguments and initialize the parent class.

Parameters:

criteria –
A str, a Callable, or an Iterable (like a tuple or a list) containing such objects. Each str or Callable represents one stop criterion. There is no limit to the given number of stop criteria. If a criterion is given as an Callable, it should accept as its sole argument the output of the crappy.blocks.Block.recv_all_data() method and return True if the criterion is met, and False otherwise. If the criterion is given as a str, it should follow the following syntax :
```
'<lab> > <threshold>'
'<lab> < <threshold>'
```
With <lab> and <threshold> to be replaced respectively with the name of a received label, and a threshold value. The spaces in the string are ignored.
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq – if True, displays the looping frequency of the Block.
debug – If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

loop() → None[source]: Receives data from upstream Blocks, checks if this data meets the criteria, and stop the test if that’s the case.

Stop Button

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

This Block allows the user to stop the current Crapy script by clicking on a button in a GUI.

Along with the StopBlock, it allows to stop a test in a clean way without resorting to CTRL+C.

New in version 2.0.0.

__init__(freq: float | None = 50, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq – If True, displays the looping frequency of the Block.
debug – If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

prepare() → None[source]: Creates the graphical interface and sets its layout and callbacks.

loop() → None[source]: Updates the interface at each loop.

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

Synchronizer

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

This Block takes data from upstream Blocks as input and interpolates it to output all the labels on the same timestamps as a reference label.

This Block is very similar to the Multiplexer Block, but the Multiplexer interpolates data in a time base independent of the labels whereas this one takes one label as a reference.

It can take any number of inputs, provided that they all share a common time label. It is also possible to choose which labels are considered for interpolation and which are dropped. The interpolation is performed using the numpy.interp method.

This Block is useful for synchronizing data acquired from different sensors, in the context when one label should be treated as a reference. This is for example the case when synchronizing signals with the output of an image processing, to be able to compare all the signals in the time base of the image acquisition.

New in version 2.0.5.

__init__(reference_label: str, time_label: str = 't(s)', labels_to_sync: str | Iterable[str] | None = None, freq: float | None = 50, display_freq: bool = False, debug: bool | None = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

reference_label – The label whose timestamps will be taken as a time base for performing the interpolation.
time_label – The label carrying the time information. Should be common to all the input Blocks.
labels_to_sync – An iterable (like a list or a tuple) containing the labels to interpolate on the reference label’s time base, except for the time label that is given separately in the time_label argument. The Block also doesn’t output anything until data has been received on all these labels. If left to None, all the received labels are considered. It is recommended to always set this argument ! It is also possible to give this argument as a single str (i.e. not in an iterable).
freq – The target looping frequency for the Block. If None, loops as fast as possible.
display_freq – If True, displays the looping frequency of the Block.
debug – If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

loop() → None[source]: Receives data, interpolates it, and sends it to the downstream Blocks.

UController

class crappy.blocks.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.

New in version 1.5.8.

__init__(labels: str | Iterable[str] | None = None, cmd_labels: str | Iterable[str] | None = None, init_output: Dict[str, float] | None = None, post_process: Dict[str, Callable[[float], float]] | None = None, t_device: bool = False, port: str = '/dev/ttyUSB0', baudrate: int = 115200, display_freq: bool = False, freq: float | None = 100, debug: bool | None = False) → None[source]

Checks the validity of the arguments.

Parameters:

labels – An iterable (like a list or a tuple) containing the labels to get from the device (as str). 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. If there’s only one label to acquire, it can be given directly as a str and not in an iterable.
cmd_labels – An iterable (like a list or a tuple) containing 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. Not more than 9 cmd_labels should be given. If there’s only one command label, it can be given directly as a str and not in an iterable.
init_output – 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 – 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 – 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 – 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 – The baudrate for serial communication. It depends on the capabilities of the device.
display_freq –
If True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
freq – The target looping frequency for the Block. If None, loops as fast as possible.
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.

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.

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.

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

Video Extenso

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

This Block can perform video-extensometry on images acquired by a Camera object, by tracking spots on the images.

It takes no input Link in a majority of situations, and outputs the results of the video-extensometry. It is a subclass of the Camera Block, and inherits of all its features. That includes the possibility to record and to display images in real-time, simultaneously to the image acquisition and processing. Refer to the documentation of the Camera Block for more information on these features.

This Block is quite similar to the DICVE Block, except this latter tracks patches with a texture instead of spots. Both Blocks output similar information, although the default labels and the data format are slightly different. See the labels argument for more detail on the output values. Similar to the DICVE, the GPUVE Block also performs video-extensometry based on GPU-accelerated image correlation.

Similar to the CameraConfig window that can be displayed by the Camera Block, this Block can display a VideoExtensoConfig window before the test starts. Here, the user can also detect and select the spots to track. It is currently not possible to specify the coordinates of the spots to track as an argument, so the use of the configuration window is mandatory. This might change in the future.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Video_extenso to VideoExtenso

__init__(camera: str, transform: Callable[[ndarray], ndarray] | None = None, config: bool = True, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, software_trig_label: str | None = None, display_freq: bool = False, freq: float | None = 200, debug: bool | None = False, save_images: bool = False, img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, img_shape: Tuple[int, int] | None = None, img_dtype: str | None = None, labels: str | Iterable[str] | None = None, raise_on_lost_spot: bool = True, white_spots: bool = False, update_thresh: bool = False, num_spots: int | None = None, safe_mode: bool = False, border: int = 5, min_area: int = 150, blur: int | None = 5, **kwargs) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

camera – The name of the Camera object to use for acquiring the images. Arguments can be passed to this Camera as kwargs of this Block. This argument is ignored if the image_generator argument is provided.
transform –
A callable taking an image as an argument, and returning a transformed image as an output. Allows applying a post-processing operation to the acquired images. This is done right after the acquisition, so the original image is permanently lost and only the transformed image is displayed and/or saved and/or further processed. The transform operation is not parallelized, so it might negatively affect the acquisition framerate if it is too heavy.

New in version 1.5.10.
config –
If True, a VideoExtensoConfig window is displayed before the test starts. There, the user can interactively adjust the different CameraSetting available for the selected Camera, visualize the acquired images, and detect and select the spots to track. The test starts when closing the configuration window. It is currently not possible to set this argument to False ! This might change in the future.

New in version 1.5.10.
display_images –
If True, displays the acquired images in a dedicated window, using the backend given in displayer_backend and at the frequency specified in displayer_framerate. This option should be considered as a debug or basic follow-up feature, it is not intended to be very fast nor to display high-quality images. The maximum resolution of the displayed images in 640x480, the images might be downscaled to fit in this format. In addition to the acquired frames, the tracked spots are also displayed on the image as an overlay.

Changed in version 1.5.10: renamed from show_image to display_images
displayer_backend –
The backend to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 (OpenCV) or matplotlib. 'cv2' usually allows achieving a higher display frequency. Ignored if display_images is False. If not given and display_images is True, 'cv2' is tried first and 'mpl' second, and the first available one is used.

New in version 1.5.10.
displayer_framerate –
The maximum update frequency of the image displayer, as an int. This value usually lies between 5 and 30Hz, the default is 5. The achieved update frequency might be lower than requested. Ignored if display_images is False.

New in version 1.5.10.
software_trig_label –
The name of a label used as a software trigger for the Camera. If given, images will only be acquired when receiving data over this label. The received value does not matter. This software trigger is not meant to be very precise, it is recommended not to rely on it for a trigger frequency greater than 10Hz, in which case a hardware trigger should be preferred if available on the camera.

New in version 2.0.0.
display_freq –
If True, displays the looping frequency of the Block.

Changed in version 2.0.0: renamed from verbose to display_freq
debug –
If True, displays all the log messages including the DEBUG ones. If False, only displays the log messages with INFO level or higher. If None, disables logging for this Block.

New in version 2.0.0.
freq –
The target looping frequency for the Block. If None, loops as fast as possible.

New in version 1.5.10.
save_images –
If True, the acquired images are saved to the folder specified in save_folder, in the format specified in img_extension, using the backend specified in save_backend, and at the frequency specified in save_period. Each image is saved with the name : <frame_nr>_<timestamp>.<extension>, and can thus easily be identified. Along with the images, a metadata.csv file records the metadata of all the saved images. This metadata is either the one returned by the get_image() method of the Camera object, or the default one generated in the loop() method of the Camera Block. Depending on the framerate of the camera and the performance of the computer, it is not guaranteed that all the acquired images will be recorded.

New in version 1.5.10.
img_extension –
The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc. Depending on the used save_backend, some extensions might not be available. It is currently not possible to customize the save parameters further than choosing the file extension. Ignored if save_images is False.

New in version 2.0.0.
save_folder –
Path to the folder where to save the images, either as a str or as a pathlib.Path. Can be an absolute or a relative path, pointing to a folder. If the folder does not exist, it will be created (if the user has permission). If the given folder already contains a metadata.csv file (and thus likely images from Crappy), images are saved to another folder with the same name except a suffix is appended. Ignored if save_images is False. If not provided and save_images is True, the images are saved to the folder Crappy_images, created next to the running script.

New in version 1.5.10.
save_period –
Must be given as an int. Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of recorded images if saving them all is not needed. Ignored if save_images is False.

New in version 1.5.10.
save_backend –
If save_images is True, the backend to use for recording the images. It should be one of:
```
'sitk', 'cv2', 'pil', 'npy'
```
They correspond to the modules SimpleITK, cv2 (OpenCV), PIL (Pillow Fork), and numpy. Note that the 'npy' backend saves the images as raw numpy.array, and thus ignores the img_extension argument. Depending on the machine, some backends may be faster or slower. For using each backend, the corresponding Python must of course be installed. If not provided and save_images is True, the backends are tried in the same order as given above and the first available one is used. 'npy' is always available.

New in version 1.5.10.
image_generator –
A callable taking two float as arguments and returning an image as a numpy.array. This argument is intended for use in the examples of Crappy, to apply an artificial strain on a base image. Most users should ignore it. When given, the camera argument is ignored and the images are acquired from the generator. To apply a strain on the image, strain values (in %) should be sent to the Camera Block over the labels 'Exx(%)' and 'Eyy(%)'.

New in version 1.5.10.
img_shape –
The shape of the images returned by the Camera object as a tuple of int. It should correspond to the value returned by numpy.shape. This argument is always ignored as config cannot be set to False. This might change in the future.

New in version 2.0.0.
img_dtype –
The dtype of the images returned by the Camera object, as a str. It should correspond to a valid data type in numpy, e.g. 'uint8'. This argument is always ignored as config cannot be set to False. This might change in the future.

New in version 2.0.0.
labels – The labels to use for sending data to downstream Blocks. If not given, the default labels are 't(s)', 'meta', 'Coord(px)', 'Eyy(%)', 'Exx(%)'. They carry for each image its timestamp, a dict containing its metadata, a list containing for each spot the coordinates of its center in a tuple of int, and the y and x strain values calculated from the displacement and the initial position of the spots. If different labels are desired, they should all be provided at once in an iterable of str containing the correct number of labels (5).
raise_on_lost_spot –
If True, raises an exception when losing the spots to track, which stops the test. Otherwise, stops the tracking but lets the test go on and silently sleeps.

Changed in version 1.5.10: renamed from end to raise_on_lost_spot
white_spots – If True, detects white objects over a black background, else black objects over a white background.
update_thresh – If True, the grey level threshold for detecting the spots is re-calculated at each new image. Otherwise, the first calculated threshold is kept for the entire test. The spots are less likely to be lost with adaptive threshold, but the measurement will be more noisy. Adaptive threshold may also yield inconsistent results when spots are lost.
num_spots – The number of spots to detect, as an int between 1 and 4. If given, will try to detect exactly that number of spots and will fail if not enough spots can be detected. If left to None, will detect up to 4 spots, but potentially fewer.
safe_mode – If True, the Block will stop and raise an exception as soon as overlapping spots are detected. Otherwise, it will first try to reduce the detection window to get rid of overlapping. This argument should be used when inconsistency in the results may have critical consequences.
border – When searching for the new position of a spot, the Block will search in the last known bounding box of this spot plus a few additional pixels in each direction. This argument sets the number of additional pixels to use. It should be greater than the expected “speed” of the spots, in pixels / frame. But if it’s set too high, noise or other spots might hinder the detection.
min_area – The minimum area an object should have to be potentially detected as a spot. The value is given in pixels, as a surface unit. It must of course be adapted depending on the resolution of the camera and the size of the spots to detect.
blur – The size in pixels (as an odd int greater than 1) of the kernel to use when applying a median blur filter to the image before the spot detection. If not given, no blurring is performed. A slight blur improves the spot detection by smoothening the noise, but also takes a bit more time compared to no blurring.
**kwargs – Any additional argument will be passed to the Camera object, and used as a kwarg to its open() method.

Removed in version 1.5.10: ext, fps_label, wait_l0 and input_label arguments

Removed in version 2.0.0: img_name argument

prepare() → None[source]

This method mostly calls the prepare() method of the parent class.

In addition to that it instantiates the VideoExtensoProcess object that performs the video-extensometry and the tracking.

Changed in version 1.5.5: now accepting args and kwargs

Changed in version 1.5.10: not accepting arguments anymore

Generator Paths

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

Conditional

class crappy.blocks.generator_path.Conditional(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], value1: float, value2: float, value0: float = 0)[source]

This Path returns one of three possible output values, based on two given conditions.

It is especially useful for controlling processes that need to behave differently based on input values, e.g. for preventing a heating element from overheating, or a motor from driving too far.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Protection to Conditional

__init__(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], value1: float, value2: float, value0: float = 0) → None[source]

Sets the args and initializes the parent class.

Parameters:

condition1 – The first condition checked by the Path. Refer to Path for more information.
condition2 – The second condition checked by the path. Refer to Path for more information.
value1 – Value to send when condition1 is met.
value2 – Value to send when condition2 is met and condition1 is not met.
value0 – Value to send when neither condition1 nor condition2 are met.

Note

This Generator Path never ends, it doesn’t have a stop condition.

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 1.5.10: verbose argument

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]: Sends either value1 if condition1 is met, or value2 if only condition2 is met, or value0 if none of the conditions are met.

Constant

class crappy.blocks.generator_path.Constant(condition: str | Callable[[Dict[str, list]], bool], value: float | None = None)[source]

The simplest Path, outputs the same constant value until the stop condition is met.

New in version 1.4.0.

__init__(condition: str | Callable[[Dict[str, list]], bool], value: float | None = None) → None[source]

Sets the args and initializes the parent class.

Parameters:

condition – The condition for switching to the next Path. Refer to Path for more information.
value – The value to output.

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 1.5.10: send_one argument

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]: Returns the value to send or raises StopIteration if the stop condition is met.

New in version 1.5.10.

Custom

class crappy.blocks.generator_path.Custom(file_name: str | Path, delimiter: str = ',')[source]

Generates a custom Path from a text file, until the file is exhausted.

The file can be in any text format, including the most common .csv and .txt extensions.

New in version 1.4.0.

__init__(file_name: str | Path, delimiter: str = ',') → None[source]

Loads the file and sets the arguments.

The stop condition is simply to reach the last timestamp given in the file.

Parameters:

file_name –
Path to the file containing the information on the Generator Path. Can be either a str or a pathlib.Path. The file must contain two columns: the first one containing timestamps (starting from 0), the other one containing the values.

Changed in version 2.0.0: renamed from filename to file_name
delimiter – The delimiter between columns in the file, usually a coma.

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(_: Dict[str, list]) → float[source]

Returns the value to send or raises StopIteration if the stop condition is met.

The value is interpolated from the given file.

Cyclic

class crappy.blocks.generator_path.Cyclic(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], value1: float, value2: float, cycles: float = 1)[source]

This Path cyclically alternates between two constant values, based on two different conditions.

It can for example be used as a trigger, or used to drive an actuator cyclically. It is equivalent to a succession of Constant Paths.

New in version 1.4.0.

__init__(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], value1: float, value2: float, cycles: float = 1) → None[source]

Sets the arguments and initializes the parent class.

The Path always starts with value1, and then switches to value2.

Parameters:

condition1 – The condition for switching to value2. Refer to Path for more information.
condition2 – The condition for switching to value1. Refer to Path for more information.
value1 – First value to send.
value2 – Second value to send.
cycles – Number of cycles. Half cycles are accepted. If 0, loops forever.

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

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 1.5.10: verbose argument

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]

Returns either the first or second value depending on the current state of the cycle. Raises StopIteration when the cycles are exhausted.

Also manages the switch between the values and conditions 1 and 2.

Cyclic Ramp

class crappy.blocks.generator_path.CyclicRamp(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], speed1: float, speed2: float, cycles: float = 1, init_value: float | None = None)[source]

This Pth cyclically alternates between two ramps with different slopes, based on two different conditions.

It is equivalent to a succession of Ramp Paths.

New in version 1.4.0.

Changed in version 2.0.0: renamed from Cyclic_ramp to CyclicRamp

__init__(condition1: str | Callable[[Dict[str, list]], bool], condition2: str | Callable[[Dict[str, list]], bool], speed1: float, speed2: float, cycles: float = 1, init_value: float | None = None) → None[source]

Sets the arguments and initializes the parent class.

The path always starts with speed1, and then switches to speed2.

Parameters:

condition1 – The condition for switching to speed2. Refer to Path for more information.
condition2 – The condition for switching to speed1. Refer to Path for more information.
speed1 – Slope of the first generated ramp, in units/s.
speed2 – Slope of the second generated ramp, in units/s.
cycles – Number of cycles. Half cycles are accepted. If 0, loops forever.
init_value –
If given, overwrites the last value of the signal as the starting point for the first ramp. In the specific case when this Path is the first one in the Generator Paths, this argument must be given !

New in version 1.5.10.

Note

[{'type': 'CyclicRamp', '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

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 1.5.10: verbose argument

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]

Returns the current value of the signal and raises StopIteration when the cycles are exhausted.

Also manages the switch between the speeds and conditions 1 and 2.

Integrator

class crappy.blocks.generator_path.Integrator(condition: str | Callable[[Dict[str, list]], bool], inertia: float, func_label: str, time_label: str = 't(s)', init_value: float | None = None)[source]

This Path integrates an incoming label over time and returns the integration as an output signal.

Let f(t) be the input signal, v(t) the value of the output, m the inertia and t0 the timestamp of the beginning of this Path.

Then the output value for this Path will be \(v(t) = v(t0) - [I(t0 -> t)f(t)dt] / m\).

New in version 1.4.0.

Changed in version 2.0.0: renamed from Inertia to Integrator

__init__(condition: str | Callable[[Dict[str, list]], bool], inertia: float, func_label: str, time_label: str = 't(s)', init_value: float | None = None) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

condition – The condition for switching to the next Path. Refer to Path for more information.
inertia – Value of the equivalent inertia to use for driving the signal. In the above formula, it is the value of m. The larger this value, the slower the changes in the signal value.
func_label –
The name of the label of the input value to integrate.

Changed in version 1.5.10: renamed from flabel to func_label
time_label –
The name of the time label for the integration.

Changed in version 1.5.10: renamed from tlabel to time_label
init_value –
If given, overwrites the last value of the signal as the starting point for the inertia path. In the specific case when this path is the first one in the Generator Paths, this argument must be given !

Changed in version 1.5.10: renamed from value to init_value

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 1.5.10: const argument

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]

Gets the latest values of the incoming label, integrates them and changes the output accordingly.

Also raises StopIteration in case the stop condition is met.

Ramp

class crappy.blocks.generator_path.Ramp(condition: str | Callable[[Dict[str, list]], bool], speed: float, init_value: float | None = None)[source]

Sends a ramp signal varying linearly over time, until the stop condition is met.

New in version 1.4.0.

__init__(condition: str | Callable[[Dict[str, list]], bool], speed: float, init_value: float | None = None)[source]

Sets the arguments and initializes the parent class.

Parameters:

condition – The condition for switching to the next Path. Refer to Path for more information.
speed – The slope of the ramp, in units/s.
init_value –
If given, overwrites the last value of the signal as the starting point for the ramp. In the specific case when this path is the first one in the Generator Paths, this argument must be given !

New in version 1.5.10.

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]: Returns the value to send or raises StopIteration if the stop condition is met.

Sine

class crappy.blocks.generator_path.Sine(condition: str | Callable[[Dict[str, list]], bool], freq: float, amplitude: float, offset: float = 0, phase: float = 0)[source]

This Path generates a sine wave varying with time until the stop condition is met.

New in version 1.4.0.

__init__(condition: str | Callable[[Dict[str, list]], bool], freq: float, amplitude: float, offset: float = 0, phase: float = 0) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

condition – The condition for switching to the next Path. Refer to Path for more information.
freq – The frequency of the sine in Hz.
amplitude – The amplitude of the sine wave (peak to peak).
offset – The offset of the sine (average value).
phase – The phase of the sine (in radians).

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float[source]: Returns the value to send or raises StopIteration if the stop condition is met.

Parent Path

Meta Path

class crappy.blocks.generator_path.meta_path.MetaPath(name: str, bases: tuple, dct: dict)[source]

Metaclass ensuring that two Paths don’t have the same name, and that all Paths define the required methods. Also keeps track of all the Path classes, including the custom user-defined ones.

New in version 2.0.0.

__init__(name: str, bases: tuple, dct: dict) → None[source]

Path

class crappy.blocks.generator_path.meta_path.Path(*_, **__)[source]

Base class for all the Generator Path objects.

The Path object are used by the Generator Block to generate signals.

New in version 1.4.0.

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

Here, the arguments given to the Path should be handled.

If the Path accepts one or more conditions, (e.g. stop conditions for switching to the next Path), they can be parsed here using the parse_condition() method. See the code of Constant for an example.

The self.t0 attribute stores the time when the last command of the previous Path was sent, and the self.last_cmd stores the value of the last command of the previous Path.

Changed in version 1.5.10: renamed time argument to _last_time

Changed in version 1.5.10: renamed cmd argument to _last_cmd

Removed in version 2.0.0: _last_time and _last_cmd arguments

get_cmd(data: Dict[str, list]) → float | None[source]

This method is called by the Generator Block to get the next command to send.

It takes as input a dict containing the data received by the Generator Block since the last command was sent. Refer to recv_all_data() for more information on the format of this dict.

This method should output the next command that will be sent by the Generator Block, as a numeric value. This value can be calculated based on one or several of :

the input data (see the code of Conditional)
the current time and the self.t0 attribute (see the code of Ramp)
the last sent value, using the self.last_cmd attribute
any other criteria

Alternatively, if the Path is done and should hand over to the next one, it must raise a StopIteration exception. Again, the choice to raise that exception can be motivated by the current time, a condition as generated by parse_condition(), or any other criteria.

It is also fine for this method to return None if no value should be output by the Generator Block for this loop.

log(level: int, msg: str) → None[source]

Records log messages for the Path.

Also instantiates the Logger when logging the first message.

Parameters:

level – An int indicating the logging level of the message.
msg – The message to log, as a str.

New in version 2.0.0.

parse_condition(condition: str | Callable[[Dict[str, list]], bool] | None) → Callable[[Dict[str, list]], bool][source]

This method returns a function allowing to check whether a given condition is met or not.

This returned function takes as an input a dict containing the data received by the Generator Block since it sent the last command. See recv_all_data() for information on the structure of this data. Based on the input data, it returns True if the condition is met, and False otherwise.

The condition can be given already as a function (or a callable), as None or more conveniently as a str to parse. If it is given as a function / callable, this callable is directly returned. If it is given as None, a function always returning False is returned.

If the condition is given as a string, the supported condition types are :

'<var> > <threshold>'
'<var> < <threshold>'
'delay = <your_delay>'

With <var>, <threshold> and <your_delay> to be replaced respectively with the label on which the condition applies, the threshold for the condition to become True, and the delay before switching to the next Path.

In the case when a str to parse is given as the condition, a function performing the check is generated and returned. This way, the user doesn’t have to understand the internals of data transfers in Crappy to handle custom conditions.

Camera Processes

Camera Process

class crappy.blocks.camera_processes.CameraProcess[source]

This Process is the base class for all the helper Processes of the Camera Block.

It defines a base architecture that all the children classes use, and on top of which they add the specific actions they perform. This class is not meant to be instantiated as is, it should always be subclassed.

This class and its children can be seen as a tool for the Camera Block and its children. Several instances can be started by the Camera Block, each instance performing a different action with the acquired images. This allows to parallelize image acquisition, display, processing and recording on multiple CPU cores, to increase the achieved FPS.

The Camera Block performs the acquisition, and makes the latest captured image available to all the CameraProcess Process through an Array. They are then free to run at their own rhythm, but are sure to always grab the latest available frame.

The instantiation, startup and termination of the CameraProcesses is all managed by the parent Camera Block, depending on the provided arguments. Users should normally not need to call this class themselves.

New in version 2.0.0.

__init__() → None[source]: Initializes the parent class and all the instance attributes.

set_shared(array: SynchronizedArray, data_dict: DictProxy, lock: RLock, barrier: Barrier, event: Event, shape: Tuple[int, int] | Tuple[int, int, int], dtype, to_draw_conn: Connection | None, outputs: List[Link], labels: List[str] | None, log_queue: Queue, log_level: int | None = 20, display_freq: bool = False) → None[source]

Method allowing the Camera Block to share multiprocessing synchronization objects with this class.

Parameters:

array – The Array containing the last frame acquired by the Camera Block.
data_dict – A dict managed by a Manager and containing the metadata of the last acquired frame.
lock – A RLock ensuring that the CameraProcess and the Camera Block do not try to access the shared array at the same time.
barrier – A Barrier ensuring that all the CameraProcesses wait for a start signal from the Camera Block before starting to run.
event – A Event indicating to the CameraProcess when to stop running. It is either set by the Camera Block, or by a CameraProcess.
shape – The expected shape of the image, as a tuple. It is necessary as the frames are shared as a one-dimensional array.
dtype – The expected dtype of the image. It is necessary for reconstructing the image from the one-dimensional shared array.
to_draw_conn – A Connection for sending or receiving Overlay objects to draw on top of the displayed image.
outputs – The Link objects for sending data to downstream Blocks. They are the same as those owned by the Camera Block.
labels – The labels to use when sending data to downstream Blocks.
log_queue – A Queue for sending the log messages to the main Logger, only used in Windows.
log_level – The minimum logging level of the entire Crappy script, as an int.
display_freq – If True, the looping frequency of this class will be displayed while running.

run() → None[source]

This method is the core of the Process.

It starts by initializing the Logger, and then performs any additional action required before processing images. Once all the CameraProcesses are ready, it loops forever and processes images until told to stop or an exception is raised. And finally, it performs any action required for properly exiting.

This method is quite similar to the run() method of the Block, although it is much simpler.

init() → None[source]

This method should perform any action required for initializing the CameraProcess.

It is meant to be overwritten by children classes, at is otherwise does not perform any action.

It is called right after the Process starts, and when the images haven’t started to be acquired yet.

loop() → None[source]

This method is the main loop of the CameraProcess.

It is called repeatedly until the Process is told to stop. It should perform the desired action for handling the latest available frame, stored in the self.img attribute. The latest available metadata containing at least the timestamp and frame index of the latest image is stored in self.metadata.

This method is meant to be overwritten by children classes, at is otherwise does not perform any action.

finish() → None[source]

This method should perform any action required for properly exiting the CameraProcess.

It is meant to be overwritten by children classes, at is otherwise does not perform any action.

It is the last method called before the Process ends, and at that point no more images are being acquired.

send(data: Dict[str, Any] | Iterable[Any] | None) → None[source]

This method allows sending data to downstream Blocks.

It is similar to the send() method of the Block. It accepts data either as a dict, or as an iterable of values.

send_to_draw(to_draw: Iterable[Overlay]) → None[source]

This method sends a collection of Overlay objects to the Displayer CameraProcess.

The overlays are sent by the CameraProcess performing the image processing, so that the area(s) of interest can be displayed simultaneously.

log(level: int, msg: str) → None[source]

Sends a log message to the Logger.

Parameters:

level – The logging level, as an int.
msg – The message to log, as a str.

DIC VE Process

class crappy.blocks.camera_processes.DICVEProcess(patches: SpotsBoxes, method: str = 'Disflow', alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, patch_size: int = 8, patch_stride: int = 3, border: float = 0.2, safe: bool = True, follow: bool = True, raise_on_exit: bool = True)[source]

This CameraProcess can perform video-extensometry by tracking patches on images using various Digital Image Correlation techniques.

It is used by the DICVE Block to parallelize the image processing and the image acquisition. It delegates most of the computation to the DICVETool. It is from this class that the output values are sent to the downstream Blocks, and that the SpotsBoxes are sent to the Displayer CameraProcess for display.

New in version 2.0.0.

__init__(patches: SpotsBoxes, method: str = 'Disflow', alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, patch_size: int = 8, patch_stride: int = 3, border: float = 0.2, safe: bool = True, follow: bool = True, raise_on_exit: bool = True) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

patches – An instance of the SpotsBoxes class, containing the coordinates of the patches to track. This argument is passed to the DICVETool and not used in this class.
method – The method to use to calculate the displacement. Disflow uses opencv’s DISOpticalFlow and Lucas Kanade uses opencv’s calcOpticalFlowPyrLK, while all other methods are based on a basic cross-correlation in the Fourier domain. Pixel precision calculates the displacement by getting the position of the maximum of the cross-correlation, and has thus a 1-pixel resolution. It is mainly meant for debugging. Parabola refines the result of Pixel precision by interpolating the neighborhood of the maximum, and has thus a sub-pixel resolution. This argument is passed to the DICVETool and not used in this class.
alpha – Weight of the smoothness term in DISFlow, as a float. This argument is passed to the DICVETool and not used in this class.
delta – Weight of the color constancy term in DISFlow, as a float. This argument is passed to the DICVETool and not used in this class.
gamma – Weight of the gradient constancy term in DISFlow , as a float. This argument is passed to the DICVETool and not used in this class.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DISFlow (0 means full scale), as an int. This argument is passed to the DICVETool and not used in this class.
iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. This argument is passed to the DICVETool and not used in this class.
gradient_iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. This argument is passed to the DICVETool and not used in this class.
patch_size – Size of an image patch for matching in DISFlow (in pixels). This argument is passed to the DICVETool and not used in this class.
patch_stride – Stride between neighbor patches in DISFlow. Must be less than patch size. This argument is passed to the DICVETool and not used in this class.
border – Crop the patch on each side according to this value before calculating the displacements. 0 means no cropping, 1 means the entire patch is cropped. This argument is passed to the DICVETool and not used in this class.
safe – If True, checks whether the patches aren’t exiting the image, and raises an error if that’s the case. This argument is passed to the DICVETool and not used in this class.
follow – It True, the patches will move to follow the displacement of the image. This argument is passed to the DICVETool and not used in this class.
raise_on_exit – If True, raises an exception and stops the test when losing the patches. Otherwise, simply stops processing but lets the test go on.

init() → None[source]: Instantiates the DICVETool that will perform the image correlation.

loop() → None[source]

This method grabs the latest frame and gives it for processing to the DICVETool. Then sends the result of the correlation to the downstream Blocks.

If there’s no new frame grabbed, or if the patches were already lost, doesn’t do anything. On the first acquired frame, does not process it but initializes the DICVETool with it instead. Also sends the current patches for display to the Displayer CameraProcess.

DIS Correl Process

class crappy.blocks.camera_processes.DISCorrelProcess(patch: Box, fields: List[str | ndarray] | None = None, alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, init: bool = True, patch_size: int = 8, patch_stride: int = 3, residual: bool = False)[source]

This CameraProcess can perform Dense Inverse Search on a given ROI in the acquired images, and calculate various fields from it.

It is used by the DISCorrel Block to parallelize the image processing and the image acquisition. It delegates most of the computation to the DISCorrelTool. It is from this class that the output values are sent to the downstream Blocks, and that the SpotsBoxes are sent to the Displayer CameraProcess for display.

New in version 2.0.0.

__init__(patch: Box, fields: List[str | ndarray] | None = None, alpha: float = 3, delta: float = 1, gamma: float = 0, finest_scale: int = 1, iterations: int = 1, gradient_iterations: int = 10, init: bool = True, patch_size: int = 8, patch_stride: int = 3, residual: bool = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

patch – An instance of the Box class, containing the coordinates of the ROI to perform DIS on. This argument is passed to the DISCorrelTool and not used in this class.
fields –
fields: The base of fields to use for the projection, given as a list of str or numpy arrays (both types can be mixed). Strings are for using automatically-generated fields, the available ones are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
If users provide their own fields as arrays, they will be used as-is to run the correlation. The user-provided fields must be of shape:
```
(patch_height, patch_width, 2)
```
This argument is passed to the DISCorrelTool and not used in this class.

Changed in version 2.0.5: provided fields can now be numpy arrays
alpha – Weight of the smoothness term in DISFlow, as a float. This argument is passed to the DISCorrelTool and not used in this class.
delta – Weight of the color constancy term in DISFlow, as a float. This argument is passed to the DISCorrelTool and not used in this class.
gamma – Weight of the gradient constancy term in DISFlow , as a float. This argument is passed to the DISCorrelTool and not used in this class.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DISFlow (0 means full scale), as an int. This argument is passed to the DISCorrelTool and not used in this class.
init – If True, the last field is used to initialize the calculation for the next one. This argument is passed to the DISCorrelTool and not used in this class.
iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. This argument is passed to the DISCorrelTool and not used in this class.
gradient_iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DISFlow, as an int. This argument is passed to the DISCorrelTool and not used in this class.
patch_size – Size of an image patch for matching in DISFlow (in pixels). This argument is passed to the DISCorrelTool and not used in this class.
patch_stride – Stride between neighbor patches in DISFlow. Must be less than patch size. This argument is passed to the DISCorrelTool and not used in this class.
residual – If True, the residuals will be computed at each new frame and sent to downstream Blocks, by default under the 'res' label.

init() → None[source]: Instantiates the DISCorrelTool that will perform the Dense Inverse Search.

loop() → None[source]

This method grabs the latest frame and gives it for processing to the DISCorrelTool. Then sends the result of the dense inverse search to the downstream Blocks.

If there’s no new frame grabbed, doesn’t do anything. On the first acquired frame, does not process it but initializes the DISCorrelTool with it instead. Also sends the ROI for display to the Displayer CameraProcess.

Displayer Process

class crappy.blocks.camera_processes.Displayer(title: str, framerate: float, backend: str | None = None)[source]

This CameraProcess can display images acquired by a Camera Block in a dedicated window.

It is meant to serve as a control or validation feature, its resolution is thus limited to 640x480 and it should not be used at high framerates. On top of the displayed image, it can also draw Overlay objects sent by other CameraProcess. This way, the user can for example visualize the spots being tracked by the VideoExtenso Block in real time.

The images can be displayed using two different backends : either using cv2 (OpenCV), or using matplotlib. OpenCV is by far the fastest and most convenient.

New in version 2.0.0.

__init__(title: str, framerate: float, backend: str | None = None) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

title – The name of the Displayer window, that will be displayed on the window border.
framerate – The target framerate for the display. The actual achieved framerate might be lower, but never greater than this value.
backend – The module to use for displaying the images. Can be either 'cv2' or 'mpl', to use respectively cv2 or matplotlib.

__del__() → None[source]: On exit, ensuring that the Thread in charge of grabbing the Overlay to display has stopped, otherwise stopping it.

init() → None[source]: Starts the Thread for grabbing the Overlay to display, and initializes the Displayer window.

loop() → None[source]

This method grabs the latest frame, casts it to 8 bits if necessary, and updates the Displayer window to draw it.

It also draws the latest received Overlay on top of the displayed frame.

finish() → None[source]: Closes the Displayer window and stops the Thread grabbing the Overlay

GPU Correl Process

class crappy.blocks.camera_processes.GPUCorrelProcess(discard_limit: float = 3, discard_ref: int = 5, calc_res: bool = False, img_ref: ndarray | None = None, verbose: int = 0, levels: int = 5, resampling_factor: float = 2, kernel_file: str | Path | None = None, iterations: int = 4, fields: List[str | ndarray] | None = None, mask: ndarray | None = None, mul: float = 3)[source]

This CameraProcess can perform GPU-accelerated digital image correlation on a given mask of the acquired images, and calculate various fields from it.

It is used by the GPUCorrel Block to parallelize the image processing and the image acquisition. It delegates most of the computation to the GPUCorrelTool. It is from this class that the output values are sent to the downstream Blocks.

It is also this class that takes the decision to send or not the results to downstream Blocks based on the value of the calculated residuals, if this option is enabled by the user.

New in version 2.0.0.

__init__(discard_limit: float = 3, discard_ref: int = 5, calc_res: bool = False, img_ref: ndarray | None = None, verbose: int = 0, levels: int = 5, resampling_factor: float = 2, kernel_file: str | Path | None = None, iterations: int = 4, fields: List[str | ndarray] | None = None, mask: ndarray | None = None, mul: float = 3) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

discard_limit – If calc_res is True, the result of the correlation is not sent to the downstream Blocks if the residuals for the current image are greater than discard_limit times the average residual for the last discard_ref images.
discard_ref – If calc_res is True, the result of the correlation is not sent to the downstream Blocks if the residuals for the current image are greater than discard_limit times the average residual for the last discard_ref images.
calc_res – If True, calculates the residuals after performing the correlation and returns the residuals along with the correlation data. By default, the residuals are returned under the label 'res'.
img_ref – A reference image for the correlation, given as a 2D numpy.array with dtype numpy.float32. If not given, the first acquired frame will be used as the reference image instead. This argument is passed to the GPUCorrelTool and not used in this class.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is displayed, and at level 3 so much information is displayed that it slows the code down. This argument is passed to the GPUCorrelTool and not used in this class.
levels – Number of levels of the pyramid. More levels may help converging on images with large strain, but may fail on images that don’t contain low spatial frequency. Fewer levels mean that the program runs faster. This argument is passed to the GPUCorrelTool and not used in this class.
resampling_factor – The factor by which the resolution is divided between each stage of the pyramid. A low factor ensures coherence between the stages, but is more computationally intensive. A high factor allows reaching a finer detail level, but may lead to a coherence loss between the stages. This argument is passed to the GPUCorrelTool and not used in this class.
kernel_file – The path to the file containing the kernels to use for the correlation. Can be a pathlib.Path object or a str. If not provided, the default GPU Kernels are used. This argument is passed to the GPUCorrelTool and not used in this class.
iterations – The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing. This argument is passed to the GPUCorrelTool and not used in this class.
fields –
The base of fields to use for the projection, given as a list of str or numpy arrays (both types can be mixed). Strings are for using automatically-generated fields, the available ones are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
If users provide their own fields as arrays, they will be used as-is to run the correlation. The user-provided fields must be of shape:
```
(patch_height, patch_width, 2)
```
This argument is passed to the GPUCorrelTool and not used in this class.

Changed in version 2.0.5: provided fields can now be numpy arrays
mask – The mask used for weighting the region of interest on the image. It is generally used to prevent unexpected behavior on the border of the image. This argument is passed to the GPUCorrelTool and not used in this class.
mul – The scalar by which the direction will be multiplied before being added to the solution. If it’s too high, the convergence will be fast but there’s a risk to go past the solution and to diverge. If it’s too low, the convergence will be slower and require more iterations. 3 was found to be an acceptable value in most cases, but it is recommended to tune this value for each application so that the convergence is neither too slow nor too fast. This argument is passed to the GPUCorrelTool and not used in this class.

init() → None[source]: Initializes the GPUCorrelTool, and set its reference image if a img_ref argument was provided.

loop() → None[source]

This method grabs the latest frame and gives it for processing to the GPUCorrelTool. Then sends the result of the correlation to the downstream Blocks.

If there’s no new frame grabbed, doesn’t do anything. On the first acquired frame, does not process it but initializes the GPUCorrelTool with it instead if no reference image was given as argument. If requested by the user, also calculates the residuals and checks that they are within the provided limit. Otherwise, just drops the calculated data.

finish() → None[source]: Performs cleanup on the GPUCorrelTool.

GPU VE Process

class crappy.blocks.camera_processes.GPUVEProcess(patches: List[Tuple[int, int, int, int]], verbose: int = 0, kernel_file: str | Path | None = None, iterations: int = 4, img_ref: ndarray | None = None, mul: float = 3)[source]

This CameraProcess can perform GPU-accelerated video-extensometry by tracking patches using digital image correlation. It then returns the position of the tracked patches.

It is used by the GPUVE Block to parallelize the image processing and the image acquisition. It delegates most of the computation to the GPUCorrelTool. It is from this class that the output values are sent to the downstream Blocks, and that the SpotsBoxes are sent to the Displayer CameraProcess for display.

New in version 2.0.0.

__init__(patches: List[Tuple[int, int, int, int]], verbose: int = 0, kernel_file: str | Path | None = None, iterations: int = 4, img_ref: ndarray | None = None, mul: float = 3) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

patches – A list containing the coordinates of the patches to track, as a tuple for each patch. Each tuple should contain exactly 4 elements, giving in pixels the y origin, x origin, height and width of the patch.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is displayed, and at level 3 so much information is displayed that it slows the code down. This argument is passed to the GPUCorrelTool and not used in this class.
kernel_file – The path to the file containing the kernels to use for the correlation. Can be a pathlib.Path object or a str. If not provided, the default GPU Kernels are used. This argument is passed to the GPUCorrelTool and not used in this class.
iterations – The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing. This argument is passed to the GPUCorrelTool and not used in this class.
img_ref – A reference image for the correlation, given as a 2D numpy.array with dtype numpy.float32. If not given, the first acquired frame will be used as the reference image instead. This argument is passed to the GPUCorrelTool and not used in this class.
mul – The scalar by which the direction will be multiplied before being added to the solution. If it’s too high, the convergence will be fast but there’s a risk to go past the solution and to diverge. If it’s too low, the convergence will be slower and require more iterations. 3 was found to be an acceptable value in most cases, but it is recommended to tune this value for each application so that the convergence is neither too slow nor too fast. This argument is passed to the GPUCorrelTool and not used in this class.

init() → None[source]: Initializes the GPUCorrelTool instances, and set their reference image if a img_ref argument was provided.

loop() → None[source]

This method grabs the latest frame and gives it for processing to the several instances of GPUCorrelTool. Then sends the displacement data to the downstream Blocks.

If there’s no new frame grabbed, doesn’t do anything. On the first acquired frame, does not process it but initializes the instances of GPUCorrelTool with it instead if no reference image was given as argument. Also sends the patches for display to the Displayer CameraProcess.

finish() → None[source]: Performs cleanup on the several GPUCorrelTool used.

Recorder Process

class crappy.blocks.camera_processes.ImageSaver(img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, send_msg: bool = False)[source]

This CameraProcess can record images acquired by a Camera Block to the desired location and in the desired format.

Various backends can be used for recording the images, some may be faster or slower depending on the machine. It is possible to only save one out of a given number of images, if not all frames are needed.

New in version 2.0.0.

__init__(img_extension: str = 'tiff', save_folder: str | Path | None = None, save_period: int = 1, save_backend: str | None = None, send_msg: bool = False) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

img_extension – The file extension for the recorded images, as a str and without the dot. Common file extensions include tiff, png, jpg, etc.
save_folder – Path to the folder where to save the images. Can be an absolute or a relative path. The folder does not need to already exit, in which case it is created.
save_period – Only one out of that number images at most will be saved. Allows to have a known periodicity in case the framerate is too high to record all the images. Or simply to reduce the number of saved images if saving them all is not needed.
save_backend –
The backend to use for saving the images. Should be one of:
```
'sitk', 'pil', 'cv2', 'npy'
```
They correspond to the modules SimpleITK, PIL (Pillow Fork), cv2 (OpenCV), and numpy. Depending on the machine, some may be faster or slower. The img_extension is ignored for the backend 'npy', that saves the images as raw numpy arrays.
send_msg –
In case no processing is performed, and if output Links are present, this argument is set to True. In that case, a message containing the timestamp, the index, and the metadata of the image is sent to downstream Blocks each time an image is saved.

New in version 2.0.5.

init() → None[source]

Creates the folder for saving the images.

If a folder is already present at the indicated path and contains images, saving to a new folder with the same name but ending with a suffix.

loop() → None[source]

This method grabs the latest frame, writes its metadata to a .csv file and saves the image at the chosen location using the chosen backend.

On the first frame, the metadata file is created and its header is populated using the metadata of the frame.

Video Extenso Process

class crappy.blocks.camera_processes.VideoExtensoProcess(detector: SpotsDetector, raise_on_lost_spot: bool = True)[source]

This CameraProcess can perform video-extensometry by tracking spots on images. It returns the strain and the position of the detected spots on the image.

It is used by the VideoExtenso Block to parallelize the image processing and the image acquisition. It delegates most of the computation to the VideoExtensoTool. It is from this class that the output values are sent to the downstream Blocks, and that the SpotsBoxes are sent to the Displayer CameraProcess for display.

New in version 2.0.0.

__init__(detector: SpotsDetector, raise_on_lost_spot: bool = True) → None[source]

Sets the arguments and initializes the parent class.

Parameters:

detector – An instance of the SpotsDetector Tool, containing the coordinates of the detected spots to track. This argument is passed to the VideoExtensoTool and not used in this class.
raise_on_lost_spot – If True, raises an exception when losing the spots to track, which stops the test. Otherwise, stops the tracking but lets the test go on and silently sleeps.

init() → None[source]: Instantiates the VideoExtensoTool and starts tracking the spots.

loop() → None[source]

This method grabs the latest frame and gives it for processing to the VideoExtensoTool. Then sends the strain and displacement data to the downstream Blocks.

If there’s no new frame grabbed or if the spots were already lost, doesn’t do anything. When losing the spots, decides whether to raise an exception or not based on the user’s choice. Also sends the patches for display to the Displayer CameraProcess.

finish() → None[source]: Indicates the VideoExtensoTool to stop tracking the spots.

Parent Block

Block

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

This class constitutes the base object in Crappy.

It is extremely versatile, and can perform a wide variety of actions during a test. Many Blocks are already defined in Crappy, but it is possible to define custom ones for specific purposes.

It is a subclass of multiprocessing.Process, and is thus an independent process in Python. It communicates with other Blocks via multiprocessing objects.

This class also contains the class methods that allow driving a script with Crappy. They are always called in the __main__ Process, and drive the execution of all the children Blocks.

New in version 1.4.0.

__init__() → None[source]: Sets the attributes and initializes the parent class.

classmethod get_name(name: str) → str[source]: Method attributing to each new Block a unique name, based on the name of the class and the number of existing instances for this class.

New in version 2.0.0.

classmethod start_all(allow_root: bool = False, log_level: int | None = 10, no_raise: bool = False) → None[source]

Method for starting a script with Crappy.

It sets the synchronization objects for all the Blocks, renices the corresponding Process and starts the Blocks.

The call to this method is blocking until Crappy finishes.

Note

It is possible to have a finer grained control of the start of a Crappy script with the methods prepare_all(), renice_all() and launch_all().

Parameters:

allow_root –
If set to True, tries to renice the Processes with sudo privilege in Linux. It requires the Python script to be run with sudo privilege, otherwise it has no effect.

Changed in version 2.0.0: renamed from high_prio to allow_root
log_level –
The maximum logging level that will be handled by Crappy. By default, it is set to the lowest level (DEBUG) so that all messages are handled. If set to a higher level, the levels specified for each Block with the debug argument may be ignored. If set to None, logging is totally disabled. Refer to the documentation of the logging module for information on the possible levels.

New in version 2.0.0.
no_raise –
When set to False, the Exceptions encountered during Crappy’s execution, as well as the KeyboardInterrupt, will raise an Exception right before Crappy returns. This is meant to prevent the execution of code that would come after Crappy, in case Crappy does not terminate as expected. This behavior can be disabled by setting this argument to True.

New in version 2.0.0.

Removed in version 2.0.0: t0, verbose, bg arguments

classmethod prepare_all(log_level: int | None = 10) → None[source]

Creates the synchronization objects, shares them with the Blocks, and starts the Process associated to the Blocks.

Also initializes the Logger for the Crappy script.

Once started with this method, the Blocks will call their prepare() method and then be blocked by a multiprocessing.Barrier.

If an error is caught at a moment when the Blocks might already be running, performs an extensive cleanup to ensure everything stops as expected.

Parameters:

log_level –

The maximum logging level that will be handled by Crappy. By default, it is set to the lowest level (DEBUG) so that all messages are handled. If set to a higher level, the levels specified for each Block with the debug argument may be ignored. If set to None, logging is totally disabled. Refer to the documentation of the logging module for information on the possible levels.

New in version 2.0.0.

Removed in version 2.0.0: verbose argument

classmethod renice_all(allow_root: bool) → None[source]

On Linux and macOS, renices the Process associated with the Blocks.

On Windows, does nothing.

If an error is caught, performs an extensive cleanup to ensure everything stops as expected.

Parameters:

allow_root –

If set to True, tries to renice the Processes with sudo privilege in Linux. It requires the Python script to be run with sudo privilege, otherwise it has no effect.

Changed in version 2.0.0: renamed from high_prio to allow_root

classmethod launch_all(no_raise: bool = False) → None[source]

The final method being called by the main Process running a script with Crappy.

It unlocks all the Blocks by releasing the synchronization Barrier, sets the shared t0 Value, and then waits for all the Blocks to finish.

In case an exception is raised, sets the stop Event for warning the Blocks, waits for the Blocks to finish, and if they don’t, terminates them.

Parameters:: no_raise – When set to False, the Exceptions encountered during Crappy’s execution, as well as the KeyboardInterrupt, will raise an Exception right before Crappy returns. This is meant to prevent the execution of code that would come after Crappy, in case Crappy does not terminate as expected. This behavior can be disabled by setting this argument to True.

Removed in version 2.0.0: t0, verbose and bg arguments

classmethod stop_all() → None[source]: Method for stopping all the Blocks by setting the stop Event.

Removed in version 2.0.0: verbose argument

classmethod reset() → None[source]

Resets Crappy by emptying the WeakSet containing references to all the Blocks and resetting the synchronization objects.

This method is called at the very end of the _cleanup() method, but can also be called to “revert” the instantiation of Blocks while Crappy isn’t started yet.

run() → None[source]

The method run by the Blocks when their Process is started.

It first calls prepare(), then waits at the Barrier for all Blocks to be ready, then calls begin(), then main(), and finally finish().

If an exception is raised, sets the shared stop Event to warn all the other Blocks.

main() → None[source]: The main loop of the run() method. Repeatedly calls the loop() method and manages the looping frequency.

prepare() → None[source]

This method should perform any action required for initializing the Block before the test starts.

For example, it can open a network connection, create a file, etc. It is also fine for this method not to be overriden if there’s no particular action to perform.

Note that this method is called once the Process associated to the Block has been started.

begin() → None[source]

This method can be considered as the first loop of the test, and is called before the loop() method.

It allows to perform initialization actions that cannot be achieved in the prepare() method.

loop() → None[source]

This method is the core of the Block. It is called repeatedly during the test, until the test stops or an error occurs.

Only in this method should data be sent to downstream Blocks, or received from upstream Blocks.

Although it is possible not to override this method, that has no practical interest and this method should always be rewritten.

finish() → None[source]

This method should perform any action required for properly ending the test.

For example, it can close a file or disconnect from a network. It is also fine for this method not to be overriden if no particular action needs to be performed.

Note that this method will normally be called even in case an error occurs, although that cannot be guaranteed.

property debug: bool | None

Indicates whether the debug information should be displayed or not.

If False (the default), only displays the INFO logging level. If True, displays the DEBUG logging level for the Block. And if None, displays only the CRITICAL logging level, which is equivalent to no information at all.

New in version 2.0.0.

property t0: float: Returns the value of t0, the exact moment when the test started that is shared between all the Blocks.

New in version 2.0.0.

add_output(link: Link) → None[source]: Adds an output Link to the list of output Links of the Block.

add_input(link: Link) → None[source]: Adds an input Link to the list of input Links of the Block.

log(log_level: int, msg: str) → None[source]

Method for recording log messages from the Block. This option should be preferred to calling print().

Parameters:

log_level – An int indicating the logging level of the message.
msg – The message to log, as a str.

New in version 2.0.0.

send(data: Dict[str, Any] | Iterable[Any] | None) → None[source]

Method for sending data to downstream Blocks.

The exact same dict is sent to every downstream Block.

This method accepts the data to send either as a dict or as another type of iterable (like a list or a tuple). If data is provided as a dict, it is sent as is. The keys of the dict then correspond to the labels. Otherwise, the values given as an iterable are first converted to a dict using the self.labels attribute containing the labels to use.

It is up to the user to match the order of the values in the iterable with the order of the labels in self.labels. If the number of labels and the number of values to send do not match, no error is raised but some data might not get sent.

data_available() → bool[source]: Returns True if there’s data available for reading in at least one of the input Link.

Changed in version 2.0.0: renamed from poll to data_available

recv_data() → Dict[str, Any][source]

Reads the first available values from each incoming Link and returns them all in a single dict.

The returned dict might not always have a fixed number of keys, depending on the availability of incoming data.

Also, the returned values are the oldest available in the Links. See recv_last_data() for getting the newest available values.

Important

If data is received over a same label from different Links, part of it will be lost ! Always avoid using a same label twice in a Crappy script.

Returns:: A dict whose keys are the received labels and with a single value for each key (usually a float or a str).

Changed in version 2.0.0: renamed from recv_all to recv_data

recv_last_data(fill_missing: bool = True) → Dict[str, Any][source]

Reads all the available values from each incoming Link, and returns the newest ones in a single dict.

The returned dict might not always have a fixed number of keys, depending on the availability of incoming data.

Important

If data is received over a same label from different Links, part of it will be lost ! Always avoid using a same label twice in a Crappy script.

Parameters:: fill_missing – If True, fills up the missing data for the known labels. This way, the last value received from all known labels is always returned. It can of course not fill up missing data for labels that haven’t been received yet.
Returns:: A dict whose keys are the received labels and with a single value for each key (usually a float or a str).

Removed in version 1.5.10: num argument

New in version 1.5.10: blocking argument

Removed in version 2.0.0: blocking argument

Changed in version 2.0.0: renamed from get_last to recv_last_data

recv_all_data(delay: float | None = None, poll_delay: float = 0.1) → Dict[str, List[Any]][source]

Reads all the available values from each incoming Link, and returns them all in a single dict.

The returned dict might not always have a fixed number of keys, depending on the availability of incoming data.

Important

If data is received over a same label from different Links, part of it will be lost ! Always avoid using a same label twice in a Crappy script. See the recv_all_data_raw() method for receiving data with no loss.

Warning

As the time label is (normally) shared between all Blocks, the values returned for this label will be inconsistent and shouldn’t be used !

Parameters:

delay – If given specifies a delay, as a float, during which the method acquired data before returning. All the data received during this delay is saved and returned. Otherwise, just reads all the available data and returns as soon as it is exhausted.
poll_delay – If the delay argument is given, the Links will be polled once every this value seconds. It ensures that the method doesn’t spam the CPU in vain.

Returns:

A dict whose keys are the received labels and with a list of received values for each key. The first item in the list is the oldest one available in the Link, the last item is the newest available.

Removed in version 1.5.10: num argument

New in version 1.5.10: blocking argument

Removed in version 2.0.0: blocking argument

Changed in version 2.0.0: renamed from get_all_last to recv_all_data

recv_all_data_raw(delay: float | None = None, poll_delay: float = 0.1) → List[Dict[str, List[Any]]][source]

Reads all the available values from each incoming Link, and returns them separately in a list of dicts.

Unlike recv_all_data() this method does not fuse the received data into a single dict, so it is guaranteed to return all the available data with no loss.

Parameters:

delay – If given specifies a delay, as a float, during which the method acquired data before returning. All the data received during this delay is saved and returned. Otherwise, just reads all the available data and returns as soon as it is exhausted.
poll_delay – If the delay argument is given, the Links will be polled once every this value seconds. It ensures that the method doesn’t spam the CPU in vain.

Returns:

A list containing dict, whose keys are the received labels and with a list of received value for each key.

New in version 2.0.0.

Meta Block

class crappy.blocks.meta_block.MetaBlock(name: str, bases: tuple, dct: dict)[source]

Metaclass ensuring that two Blocks don’t have the same name, and that all Blocks define the required methods. Also keeps track of all the Block classes, including the custom user-defined ones.

New in version 2.0.0.

__init__(name: str, bases: tuple, dct: dict) → None[source]