Blocks

Autodrive

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

This block is meant to drive an actuator on which a camera performing videoextensometry 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.

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.

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

Sets the args and initializes the parent class.

Parameters:

actuator – A dict for initializing the actuator to drive. It should contain the name of the actuator under the key 'name', and all the arguments to pass to the actuator as key/value pairs. The default actuator if this argument is not set is the CM Drive with its default arguments.
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.
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 make bring the spots closer or farther, you have to try !
pixel_range – The size of the image (in pixels) along the chosen axis.
max_speed – The absolute maximum speed value that can be sent to the actuator.
freq – The block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

finish() → None[source]: Simply sets the device speed to 0.

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.

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

Camera

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

This block simply acquires images from a camera.

It can then save the images, and / or display them. The image acquisition can be triggered via incoming links. Optionally, a configuration window can be displayed for interactively tuning the camera settings before the test starts.

This class also serves as a base class for other blocks that perform image processing on the acquired frames.

__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, verbose: bool = False, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, **kwargs) → None[source]

Sets the args and initializes the parent class.

Parameters:

camera – The name of the camera to control. See Cameras for an exhaustive list of available cameras.
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
config – If True, a config window is shown before the test starts for interactively tuning the camera settings. It also allows selecting the spots to track.
display_images – If True, a window displays the acquired images in low resolution during the test. This display is mainly intended for debugging and visual follow-up, but not for displaying high-quality images.
displayer_backend –
If display_images is True, the backend to use for the display window. Should be one of :
```
'cv2', 'mpl'
```
If not given, OpenCV will be used if available.
displayer_framerate – If display_images is True, sets the maximum framerate for updating the display window. This setting allows limiting the resources used by the displayer. Note that the actual achieved framerate might differ, this is just the maximum limit.
software_trig_label – If given, the block will only acquire images when receiving data on this label. The received data can be anything, even empty. This label will thus de facto act as a software trigger for the camera.
verbose – If True, the achieved framerate will be displayed in the console during the test.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
**kwargs – Any additional argument to pass to the camera.

finish() → None[source]: Closes the camera and the displayer.

loop() → None[source]: Receives the incoming data, acquires an image, displays it, saves it, and finally processes it if needed.

prepare() → None[source]: Preparing the save folder, opening the camera and displaying the configuration GUI.

Client Server

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

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

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

__init__(broker: bool = False, address: Any = 'localhost', port: int = 1883, init_output: Dict[str, Any] | None = None, topics: List[str | Tuple[str, ...]] | None = None, cmd_labels: List[str | Tuple[str, ...]] | None = None, labels_to_send: List[str | Tuple[str, ...]] | None = None, verbose: bool = False, freq: float = 200, spam: bool = False) → None[source]

Checks arguments validity 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 (optional) – The network address on which the MQTT broker is running.
port (int, optional) – A network port on which the MQTT broker is listening.
init_output (dict, optional) – A dict containing for labels 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.
topics (list, optional) – A list of str and/or tuple of str. Each string corresponds to the name of a crappy label to be received from the broker. Each element of the list is considered to be the name of an MQTT topic, to which the client subscribes. After a message has been received on that topic, the block returns for each label in the topic (i.e. each string in the tuple) the corresponding data from the message. It also returns the current timestamp in the label ‘t(s)’.
cmd_labels (list, optional) – A list of str and/or tuple of str. Each string corresponds to the name of a crappy label to send to the broker. Each element of the list is considered to be the name of an MQTT topic, in which the client publishes. Grouping labels in a same topic (i.e. strings in a same tuple) allows to keep the synchronization between signals coming from a same block, as they will be published together in a same message. This is mostly useful for sending a signal along with its timeframe.
labels_to_send (list, optional) – A list of str and/or tuple of str. Allows to rename the labels before publishing data. The structure of labels_to_send should be the exact same as cmd_labels, with each label in labels_to_send replacing the corresponding one in cmd_labels. This is especially useful for transferring several signals along with their timestamps, as the label 't(s)' should not appear more than once in the topics list of the receiving block.
verbose – If True, displays the looping frequency of the block.
freq – The block will try to loop at this frequency.
spam – If True, sends the last received values at each loop even if no new values were received from the broker.

Note

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

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

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

is equivalent to

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

Examples

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

cmd_labels: If

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

the client will publish data in the form of

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

in the topics

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

labels_to_send: If

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

the data from labels

't(s)', 'cmd'

will be published in the topic

('t1', 'cmd1')

and the data from label

'sign'

in the topic

('sign',)

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

loop() → None[source]

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

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

prepare() → None[source]: Starts the broker and connects to it.

Dashboard

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

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

It can only display data coming from one block.

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

Sets the args and initializes parent class.

Parameters:

labels – Only the data from these labels will be printed on the window.
nb_digits – Number of decimals to show.
verbose – If True, prints the looping frequency of the block.
freq – If set, the block will try to loop at this frequency.

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

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

class crappy.blocks.dashboard.Dashboard_window(labels: List[str])[source]

The GUI for displaying the label values.

__init__(labels: List[str]) → None[source]: Initializes the GUI and sets the layout.

Discorrel

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

This block performs Dense Inverse Search on an image using OpenCV’s DISOpticalFlow.

First, a region of interest has to be selected in a GUI. At each new frame, the displacement field between the frame and the reference image is projected on the desired base of fields, and the returned results are then the average values of each field in the selected region of interest.

This block is mainly intended for calculating the average displacement and/or the average strain in the region of interest, but other fields can also be computed.

__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, verbose: bool = False, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, fields: List[str] | None = None, labels: List[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 args and initializes the camera.

Parameters:

camera – The name of the camera to control. See Cameras for an exhaustive list of available cameras.
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
config – If True, a config window is shown before the test starts for interactively tuning the camera settings. It also allows selecting the spots to track.
display_images – If True, a window displays the acquired images in low resolution during the test. This display is mainly intended for debugging and visual follow-up, but not for displaying high-quality images.
displayer_backend –
If display_images is True, the backend to use for the display window. Should be one of :
```
'cv2', 'mpl'
```
If not given, OpenCV will be used if available.
displayer_framerate – If display_images is True, sets the maximum framerate for updating the display window. This setting allows limiting the resources used by the displayer. Note that the actual achieved framerate might differ, this is just the maximum limit.
verbose – If True, the achieved framerate will be displayed in the console during the test.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
fields –
A list of str representing the base of fields on which the image will be projected during correlation. The possible fields are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
If not given, the default fields are :
```
["x", "y", "exx", "eyy"]
```
labels –
A list containing the labels to send to downstream blocks. The first label should be time, the second the metadata, and there should then be one label per field. If not given, the default labels are :
```
['t(s)', 'meta', 'x(pix)', 'y(pix)', 'Exx(%)', 'Eyy(%)']
```
alpha – Weight of the smoothness term in DisFlow.
delta – Weight of the color constancy term in DisFlow.
gamma – Weight of the gradient constancy term in DisFlow.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DisFlow (0 means full scale).
iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DisFlow.
gradient_iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DisFlow.
init – If True, the new optical flow is at each loop initialized using the previous optical flow.
patch_size – Size of an image patch for matching in DisFlow (in pixels).
patch_stride – Stride between neighbor patches in DisFlow. Must be less than patch size.
residual – If True, the residuals are computed and sent under the label 'res'. This label shouldn’t be included if custom labels are given.
**kwargs – Any additional argument to pass to the camera.

begin() → None[source]: Capturing a first image and setting it as a reference for the correlation.

finish() → None[source]: Closing the displayer and the camera.

prepare() → None[source]: Opening the camera, displaying the config window and setting the region of interest.

Displayer

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

This class manages the display of images captured by a camera object.

It can either work with Matplotlib or OpenCV as a backend, OpenCV being the fastest. It tries to display images at a given framerate, dropping part of the received frames if necessary.

The images are displayed with a maximum resolution of 640x480, and are resized to match that resolution if necessary. Similarly, the maximum bit depth is 8 bits, and the images are cast if necessary. Resizing and casting are anyway less demanding on the CPU than displaying big images.

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

Sets the args and the other instance attributes, and looks for an available display backend if none was specified.

Parameters:

title – The title to display on the display window.
framerate – The maximum framerate for displaying the images. To achieve this framerate, part of the received frames are simply dropped.
backend –
The backend to use for displaying the images. Should be one of:
```
'cv2', 'mpl'
```

finish() → None[source]: Calls the right finish method depending on the chosen backend.

prepare() → None[source]: Calls the right prepare method depending on the chosen backend.

update(img: ndarray) → None[source]

Ensures the target framerate is respected, and calls the right update method depending on the chosen backend.

Parameters:: img – The image to display on the displayer.

Disve

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

This block tracks the motion of regions of an image (patches), taking the first image as a reference.

It relies on cross-correlation, and is thus well-suited for tracking speckled patches. The displacement output may then be used to compute strains, and so to perform video-extensometry.

The images may be acquired by a camera, or be sent from another block. Several algorithms are available for tracking the patches, with different characteristics. All the computations are done by the Disve tool.

__init__(camera: str, patches: List[Tuple[int, int, int, int]], transform: Callable[[ndarray], ndarray] | None = None, config: bool = True, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, verbose: bool = False, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: List[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, **kwargs) → None[source]

Sets a few attributes.

Parameters:

camera – The camera to use for acquiring the images. It should be one of the Supported cameras.
patches – A list containing the different patches to track. Each patch should be given as follows : (y min, x_min, height, width).
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
config – If True, a config window is shown before the test starts for interactively tuning the camera settings. It also allows selecting the spots to track.
display_images – If True, a window displays the acquired images in low resolution during the test. This display is mainly intended for debugging and visual follow-up, but not for displaying high-quality images.
displayer_backend –
If display_images is True, the backend to use for the display window. Should be one of :
```
'cv2', 'mpl'
```
If not given, OpenCV will be used if available.
displayer_framerate – If display_images is True, sets the maximum framerate for updating the display window. This setting allows limiting the resources used by the displayer. Note that the actual achieved framerate might differ, this is just the maximum limit.
verbose – If True, the achieved framerate will be displayed in the console during the test.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
labels – The labels associated with the timestamp and the displacement of the patches. If not given, the time label is t(s), the metadata label is 'meta', the first patch displacement labels are p0x and p0y, the second p1x and p1y, etc.
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 have thus sub-pixel resolutions.
alpha – Weight of the smoothness term in DisFlow.
delta – Weight of the color constancy term in DisFlow.
gamma – Weight of the gradient constancy term in DisFlow.
finest_scale – Finest level of the Gaussian pyramid on which the flow is computed in DisFlow (0 means full scale).
iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DisFlow.
gradient_iterations – Maximum number of gradient descent iterations in the patch inverse search stage in DisFlow.
patch_size – Size of an image patch for matching in DisFlow (in pixels).
patch_stride – Stride between neighbor patches in DisFlow. Must be less than patch size.
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.
safe – If True, checks whether the patches aren’t exiting the image, and raises an error if that’s the case.
follow – It True, the patches will move to follow the displacement of the image.
**kwargs – Any additional argument to pass to the camera.

begin() → None[source]: Takes a first image from the camera and uses it to initialize the Disve tool.

finish() → None[source]: Closes the Disve tool and the camera acquiring the images.

prepare() → None[source]: Opens the camera for acquiring images and displays the corresponding settings window.

Drawing

class crappy.blocks.drawing.Dot_text(drawing, coord: Tuple[int, int], text: str, label: str, **__: str)[source]

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

__init__(drawing, coord: Tuple[int, int], text: str, label: str, **__: str) → None[source]

Simply sets the args.

Parameters:

drawing – The parent drawing block.
coord – The coordinates of the text and the color dot on the drawing.
text – The text to display.
label – The label carrying the information for updating the text and the color of the dot.
**__ –
Other unused arguments.

Important

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

update(data: Dict[str, float]) → None[source]: Updates the text and the color dot according to the received values.

class crappy.blocks.drawing.Drawing(*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 links.

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

__init__(image: str, draw: List[Dict[str, Any]] | None = None, color_range: Tuple[float, float] = (20, 300), title: str = 'Drawing', window_size: Tuple[int, int] = (7, 5), backend: str = 'TkAgg', freq: float = 2, verbose: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters:

image – Path to the image that will be the background of the canvas, as a str.
draw – A list 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.
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 block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

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

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

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

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

class crappy.blocks.drawing.Text(_, coord: Tuple[int, int], text: str, label: str, **__: str)[source]

Displays a simple text line on the drawing.

__init__(_, coord: Tuple[int, int], text: str, label: str, **__: str) → None[source]

Simply sets the args.

Parameters:

_ – The parent drawing block.
coord – The coordinates of the text on the drawing.
text – The text to display.
label – The label carrying the information for updating the text.
**__ – Other unused arguments.

update(data: Dict[str, float]) → None[source]: Updates the text according to the received values.

class crappy.blocks.drawing.Time(drawing, coord: Tuple[int, int], **__)[source]

Displays a time counter on the drawing, starting at the beginning of the test.

__init__(drawing, coord: Tuple[int, int], **__) → None[source]

Simply sets the args.

Parameters:

drawing – The parent drawing block.
coord – The coordinates of the time counter on the drawing.
**__ – Other unused arguments.

update(_: Dict[str, float]) → None[source]: Updates the time counter, independently of the received values.

Fake machine

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

This block simulates the behavior of a tensile test machine.

It should be used to simulate tensile teste, not compression tests. By default, it assumes a plastic behavior of the material. The main mechanical parameters of the material are tunable.

This block is meant to be driven like the Machine block. However, its outputs are different and are : t(s), F(N), x(mm), Exx(%), Eyy(%).

__init__(k: 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 = 100, verbose: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters:

k – The rigidity of the material, in N, so that force = k x strain.
l0 – The initial length of the fake sample to test, in mm.
max_strain – The maximum strain the material can withstand before breaking.
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 block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

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[source]: Sends a value so that the camera gets an image during prepare for the fake test examples. To be removed at some point.

crappy.blocks.fake_machine.plastic(v: float, yield_strain: float = 0.005, rate: float = 0.02) → float[source]: A basic plastic law given as an example.

Generator

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

This block generates a signal following a user-defined path.

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

It should be used as an input for other blocks, for example for driving a Machine or an IOBlock. It can also accept inputs, to make the path dependent on data coming from other blocks.

__init__(path: List[Dict[str, Any]], freq: float = 200, cmd_label: str = 'cmd', cycle_label: str = 'cycle', repeat: bool = False, spam: bool = False, verbose: bool = False, end_delay: float | None = 2, safe_start: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters:

path – It must be a list of dict, each dict providing the parameters to generate the path. Refer to the Note below for more information.
freq – The looping frequency this block will try to achieve. Note that the higher this value, the more accurate the path will be. It will also consume more resources.
cmd_label – The label of the signal sent to the downstream blocks.
cycle_label – In addition to the cmd_label, this label holds the index of the current dict in the path list. Useful to trig a block upon change in the current dict.
repeat – If True, the path will loop forever instead of stopping when the list of dicts is exhausted.
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 path switched to a new dict.
verbose – if True, displays the loop frequency of the block and a message when switching to the next dict of path.
end_delay – When all the dicts in path are exhausted, waits this many seconds before stopping the entire program. Can be set to None, in which case the Generator won’t stop the program when finishing.
safe_start – Ensures the first dict in 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.

Note

The different types of signals that can be generated by the Generator can be found at generator path. The path list contains one dict per signal shape to generate. They are generated in the order in which they appear in the list.

Each dict contains information on the signal shape to generate, like its type, any applicable parameter(s), and the stop condition(s). Refer to the documentation of each signal shape to which information to give.

begin() → None[source]: Initializes the first path and runs a loop(), that may be blocking.

loop(blocking: bool = False) → 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 paths.

Parameters:: blocking – It True, waits blocks until there’s data available from the upstream blocks before getting the next command to send.

generator path

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

class crappy.blocks.generator_path.path.Path(_last_time: float, _last_cmd: float | None = None)[source]

Parent class for all the generator paths.

Allows them to have access to the:meth:parse_condition method.

__init__(_last_time: float, _last_cmd: float | None = None) → None[source]: Simply sets the arguments.

get_cmd(_: Dict[str, list]) → float[source]: If not overridden, simply returns the last_cmd attribute.

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 the stop condition is met or not.

Its main use is to parse the conditions given as strings, but it can also accept None or a callable as arguments.

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

constant

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

The simplest path, simply sends the same value until the condition is met.

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

Sets the args and initializes the parent class.

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition – The condition for switching to the next path. Refer to generator path for more info.
value – The value to send.

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

custom

class crappy.blocks.generator_path.custom.Custom(_last_time: float, _last_cmd: float, filename: 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.

__init__(_last_time: float, _last_cmd: float, filename: str | Path, delimiter: str = ',') → None[source]

Loads the file and sets the args.

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

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
filename – Path to the file to read the path from. 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.
delimiter – The delimiter between columns in the file, usually a coma.

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 ramp

class crappy.blocks.generator_path.cyclic_ramp.Cyclic_ramp(_last_time: float, _last_cmd: float, 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]

The path cyclically alternates between two ramps with different slopes, based on two different conditions.

It is equivalent to a succession of ramp paths.

__init__(_last_time: float, _last_cmd: float, 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 args and initializes the parent class.

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

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition1 – The condition for switching to speed2. Refer to generator path for more info.
condition2 – The condition for switching to speed1. Refer to generator path for more info.
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 list of dicts, this argument must be given !

Note

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

is equivalent to

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

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.

cyclic

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

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

__init__(_last_time: float, _last_cmd: float, 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 args and initializes the parent class.

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

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition1 – The condition for switching to value2. Refer to generator path for more info.
condition2 – The condition for switching to value1. Refer to generator path for more info.
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

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.

inertia

class crappy.blocks.generator_path.inertia.Inertia(_last_time: float, _last_cmd: float, 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

__init__(_last_time: float, _last_cmd: float, 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 args and initializes the parent class.

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition – The condition for switching to the next path. Refer to generator path for more info.
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.
time_label – The name of the time label for the integration.
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 list of dicts, this argument must be given !

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.

protection

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

Depending on two different conditions checked at each loop, this path can output one between 3 constant values.

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

__init__(_last_time: float, _last_cmd: float, 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:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition1 – The first condition checked by the path. Refer to generator path for more info.
condition2 – The second condition checked by the path. Refer to generator path for more info.
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.

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.

ramp

class crappy.blocks.generator_path.ramp.Ramp(_last_time: float, _last_cmd: float, 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.

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

Sets the args and initializes the parent class.

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition – The condition for switching to the next path. Refer to generator path for more info.
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 list of dicts, this argument must be given !

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.Sine(_last_time: float, _last_cmd: float, 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.

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

Sets the args and initializes the parent class.

Parameters:

_last_time – The last timestamp when a command was generated. For internal use only, do not overwrite.
_last_cmd – The last sent command. For internal use only, do not overwrite.
condition – The condition for switching to the next path. Refer to generator path for more info.
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).

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

GPUCorrel

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

This blocks projects a displacement field on a given base of fields, and sends the decomposition to downstream blocks.

It relies on the GPU Correl class. The displacement is calculated for the entire image, and it is not possible to select a region of interest.

__init__(camera: str, fields: List[str], transform: Callable[[ndarray], ndarray] | None = None, display_images: bool = False, displayer_backend: str | None = None, displayer_framerate: float = 5, verbose: int = 0, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: List[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: Path | str | None = None, iterations: int = 4, mask: ndarray | None = None, mul: float = 3, res: bool = False, **kwargs) → None[source]

Sets the args and initializes the camera object.

Parameters:

camera – The name of the camera to control. See Cameras for an exhaustive list of available cameras.
fields –
A list of str representing the base of fields on which the image will be projected during correlation. The possible fields are :
```
'x', 'y', 'r', 'exx', 'eyy', 'exy', 'eyx', 'exy2', 'z'
```
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
display_images – If True, the difference between the original and the displaced image after correlation will be displayed. 128 means no difference, lighter means positive and darker negative.
displayer_backend –
If display_images is True, the backend to use for the display window. Should be one of :
```
'cv2', 'mpl'
```
If not given, OpenCV will be used if available.
displayer_framerate – If display_images is True, sets the maximum framerate for updating the display window. This setting allows limiting the resources used by the displayer. Note that the actual achieved framerate might differ, this is just the maximum limit.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is printed, and at level 3 so much information is printed that is slows the code down.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
labels – A list containing the labels to send to downstream blocks, carrying the displacement projected on the given basis of fields. If not given, the labels list is just a copy of the fields list, with 't(s)' added in position 0, 'meta' in position 1, and 'res' added in last position if the res argument is True.
discard_limit – If given, the data is sent to downstream blocks only if the residuals are lower than the average of the last few residuals multiplied by this value.
discard_ref – When checking whether the data should be sent based on the discard_limit argument, only that many previous values will be considered when calculating the average of the last residuals.
img_ref – The reference image to which all the acquired images will be compared for performing the correlation. If not given, the first acquired images will be used as the reference image.
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.
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.
kernel_file – The path to the file containing the kernels to use for the correlation. Can be a pathlib.Path object or a str.
iterations – The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing.
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.
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 that 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.
res – If True, the residuals will be sent to downstream blocks along with the other information under the label 'res'.
**kwargs – Any additional argument to pass to the camera.

begin() → None[source]: Acquires an image and sets it as the reference image if no image was given as reference in the arguments.

finish() → None[source]: Closes the displayer and the camera object.

prepare() → None[source]: Opens the camera, prepares the displayer and sets the reference image if one was given.

GPUve

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

This block tracks patches on an image using GPU-accelerated Dense Inverse Search.

The patches must be given explicitly as arguments, they cannot be selected in a GUI. The block sends the updated positions of the tracked patches to the downstream blocks, as well as the timestamp. It is roughly equivalent to the Disve block, but with GPU acceleration.

__init__(camera: str, patches: List[Tuple[int, int, int, int]], transform: Callable[[ndarray], ndarray] | None = None, verbose: int = 0, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: List[str] | None = None, img_ref: ndarray | None = None, kernel_file: Path | str | None = None, iterations: int = 4, mul: float = 3, **kwargs) → None[source]

Sets the args and initializes the camera.

Parameters:

camera – The name of the camera to control. See Cameras for an exhaustive list of available cameras.
patches – A list containing the patches to track given as tuple. Each patch should contain in that order: the y coord of its origin, the x coord of its origin, its size along the y axis and its size along the x axis. Any number of patches can be given.
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
verbose – The verbose level as an integer, between 0 and 3. At level 0 no information is printed, and at level 3 so much information is printed that is slows the code down.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
labels –
A list containing the labels to send to downstream blocks, carrying the information on the latest position of the patches. If not given, the default labels are :
```
['t(s)', 'meta', 'p0x', 'p0y', ..., pix, piy]
```
with i the number of given patches.
img_ref – The reference image to which all the acquired images will be compared for performing the correlation. If not given, the first acquired images will be used as the reference image.
kernel_file – The path to a file containing the kernel modules for pycuda to use. If not given ,the default kernels of Crappy are used.
iterations – The maximum number of iterations to run before returning the results. The results may be returned before if the residuals start increasing.
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 that 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.
**kwargs – Any additional argument to pass to the camera.

begin() → None[source]: Acquires a first image and sets it as the reference image if no reference image was previously given.

finish() → None[source]: Closes the correlation objects as well as the camera.

prepare() → None[source]: Opens the camera and sets the reference sub-image for each patch if a reference image was given.

Grapher

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

The grapher receive data from a block and plots it.

Multiple curves can be plotted on a same graph, and the data can come from different blocks.

Note

To reduce the memory and CPU usage of graphs, try lowering the maxpt parameter (2-3000 is already enough to follow a short test), or set the length parameter to a non-zero value (again, 2-3000 is fine). Lowering the freq is also a good option to limit the CPU use.

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

Sets the args and initializes the parent class.

Parameters:

*labels (tuple) – 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 (int, optional) – If 0 the graph is static and displays all data from the start of the assay. Else only displays the last length received chunks, and drops the previous ones.
freq (float, optional) – The refresh rate of the graph. May cause high CPU use if set too high.
maxpt (int, optional) – The maximum number of points displayed on the graph. When reaching this limit, the block deletes one point out of two to avoid using too much memory and CPU.
window_size (tuple, optional) – The size of the graph, in inches.
window_pos (tuple, optional) – The position of the graph in pixels. The first value is for the x direction, the second for the y direction. The origin is the top-left corner. Works with multiple screens.
interp (bool, optional) – If True, the data points are linked together by straight lines. Else, only the points are displayed.
backend (int, optional) – The matplotlib backend to use. Performance may vary according to the chosen backend. Also, every backend may not be available depending on your machine.
verbose (bool, optional) – To display the loop frequency of the block.

Example

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

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

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

will plot a static graph.

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

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

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

prepare() → None[source]

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

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

It can remain empty and do nothing.

GUI

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

This block allows the user to send a signal 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.

__init__(send_0: bool = False, label: str = 'step', time_label: str = 't(s)', freq: float = 50, spam: bool = False) → None[source]: Sets the args and initializes the parent class.

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

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

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

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

HDF Recorder

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

This block saves data efficiently into a hdf5 file.

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

Important

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

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

Sets the args 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.
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 block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

begin() → None[source]: Receives the first chunk of data, makes sure that it contains the label to save, and initializes the HDF array with it.

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

loop() → None[source]: Simply receives data from the upstream block and saves it.

prepare() → None[source]: Checking 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.

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

IOBlock

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

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

If it has incoming links, it will set the commands received over the labels given in cmd_labels. Additional commands to set at the very beginning or the very end of the test can also be specified.

If it has outgoing links, it will acquire data and send it downstream over the labels given in labels. It is possible to trigger the acquisition using a predefined label.

__init__(name: str, labels: List[str] | None = None, cmd_labels: List[str] | None = None, trigger_label: str | None = None, streamer: bool = False, initial_cmd: list | None = None, exit_cmd: list | None = None, make_zero_delay: float | None = None, spam: bool = False, freq: float | None = None, verbose: bool = False, **kwargs) → None[source]

Sets the args and initializes the parent class.

Parameters:

name – The name of the In / Out class to instantiate.
labels – A list 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 values returned, and given in the appropriate order. The first label must always be the timestamp, preferably called 't(s)'. This argument can be omitted if get_data() returns a dict instead of a list. Ignored if the block has no output link.
cmd_labels – A list of the labels considered as inputs for 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 (can be any value) is received on this label. Ignored if the block has no output link. A trigger label can also be a cmd label.
streamer – If False, the get_data() method of the InOut object is called for acquiring data, else it’s the get_stream() method.
initial_cmd – An initial command for the InOut, set during prepare(). If given, there must be as many values as in cmd_labels.
exit_cmd – A final command for the InOut, set during finish. If given, there must be as many values as in cmd_labels.
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.
spam – If False, the block will call set_cmd() on the InOut object only if the current command is different from the previous.
freq – The block will try to loop as this frequency, or as fast as possible if no value is given.
verbose – If True, prints the looping frequency of the block.
**kwargs – The arguments to be passed to the In / Out class.

finish() → None[source]: Stops the stream, sets the exit command if necessary, and closes the device.

loop() → None[source]

Gets the latest command, reads data from the device and sets the command.

Also handles the trig label if one was given, and manages the buffer for the previously received commands.

prepare() → None[source]: Checks the consistency of the link layout, opens the device and sets the initial command if required.

Machine

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

This block is meant to drive one or several 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.

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

Sets the args and initializes the parent class.

Parameters:

actuators – The list of all the Actuators 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 key will be passed to the Actuator object as argument when instantiating it.
common – The keys of this dict will be common to all the Actuators. If it 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.
spam – If True, a command is sent to the Actuators on each loop of the block, else it is sent every time a command is received.
freq – The block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

Note

actuators keys:
- type: The name of the Actuator class to instantiate.
- cmd: The label carrying the command for driving the Actuator.
- mode: Can be either ‘speed’ or ‘position’. Will either call set_speed() or set_position() to drive the actuator, and get_speed() or get_position() for acquiring the current speed or position.
- speed: If mode is ‘position’, the speed at which the Actuator should move. This key is not mandatory, even in the ‘position’ mode.
- pos_label: If given and the mode is ‘position’, the block will return the value of get_position() under this label. This key is not mandatory.
- speed_label: If given and the mode is ‘speed’, the block will return the value of get_speed() under this label. This key is not mandatory.

finish() → None[source]: Stops and closes all the actuators to drive.

loop() → None[source]: Receives the commands from upstream blocks, sets them on the actuators to drive, and sends the read positions and speed to the downstream blocks.

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

Mean_block

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

This block computes the average values over a given delay of each label received, and returns them.

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.

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.

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

Sets the args and initializes the parent class.

Parameters:

delay – The averaged data will be sent each delay seconds.
time_label – The label containing the time information.
out_labels – If given, only the listed labels and the time will be returned. Otherwise, all of them are returned.
verbose – If True, prints the looping frequency of the block.
freq – The block will try to loop at this frequency.

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

prepare() → None[source]: Initializes the buffer and the time counters.

Multiplex

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

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

It is useful for synchronizing data acquired from different sensors, e.g. to plot a real-time stress-strain curve. This block is however quite resource-consuming, so it is preferable to perform interpolation in 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. So it should only be used with care when it is an input of a decision-making block. This is especially true when the upstream blocks have very different sample rates.

__init__(time_label: str = 't(s)', freq: float = 200, verbose: bool = False) → None[source]

Sets the args and initializes the parent class.

Parameters:

time_label – The label carrying the time information.
freq – The sample rate for the interpolation, and the target looping frequency for the block. If this value is set too high and your machine cannot keep up, the block will most likely lag.
verbose – If True, prints information about the looping frequency of the block.

begin() → None[source]

Receiving the first data from the upstream blocks, and checking that it is valid.

If part of the data is not valid, warning the user. For the valid data, initializing the different dicts with it.

finish() → None[source]: Just sending any remaining data.

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

PID

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

A basic implementation of a PID corrector.

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

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

Sets the args and initializes the parent class.

Parameters:

kp – The P gain.
ki – The I gain.
kd – The D gain.
out_max – Ensures the output is always inferior to this value.
out_min – Ensures the output is always superior to this value.
target_label – The label carrying the setpoint value.
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.
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 block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

begin() → None[source]: Receives the first target and input values and makes sure the given labels are correct.

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.

Reader

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

Reads and prints the data flowing through the input Link.

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

Sets the arg and initializes the parent class.

Parameters:

name – If set, will be printed to identify the reader.
freq – The block will try to loop at this frequency.
verbose – If True, the looping frequency will be printed every 2s.

loop() → None[source]: Simply flushes the link and prints its data.

Recorder

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

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 one upstream block. To save data from multiple blocks, use several instances of Recorder (recommended) or a Multiplex block.

__init__(filename: str | Path, delay: float = 2, labels: List[str] | None = None, freq: float = 200, verbose: bool = True) → None[source]

Sets the args 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.
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 block will try to loop at this frequency.
verbose – If True, prints the looping frequency of the block.

begin() → None[source]: Receives the first chunk of data, writes the labels names in the first row of the file and starts saving the actual data.

finish() → None[source]: Gathers any data left in the links, and saves it.

loop() → None[source]: Simply receives data from the upstream block and saves it.

prepare() → None[source]: Checking 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.

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

Sink

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

Test block used to get data and do nothing.

__init__(verbose: bool = False, freq: float = 10) → None[source]: Sets the args and initializes the parent class.

loop() → None[source]: Simply drops all received data.

UController

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

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

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

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

Checks the validity of the arguments.

Parameters:

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

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

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

Important

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

Note

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

prepare() → None[source]

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

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

Note

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

VideoExtenso

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

This block can detect and track spots on images, and compute a strain value based on the displacement of the spots.

It is meant to be used for following spots drawn on a sample during a tensile test, so that the local strain values can be deduced from the displacement of the spots. The spots are selected interactively in a GUI before the test starts.

The timestamp, strain values, as well as the position of the detected spots are sent to downstream blocks for each received image.

__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, verbose: bool = False, freq: float = 200, save_images: bool = False, img_name: str = '{self._n_loops:6d}_{t-self.t0:.6f}.tiff', save_folder: Path | str | None = None, save_period: int = 1, save_backend: str | None = None, image_generator: Callable[[float, float], ndarray] | None = None, labels: List[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 = 5, **kwargs) → None[source]

Sets the args and initializes the camera.

Parameters:

camera – The name of the camera to control. See Cameras for an exhaustive list of available cameras.
transform – A function taking an image as an argument and returning a transformed image. The original image is discarded and only the transformed one is kept for processing, display and saving.
config – If True, a config window is shown before the test starts for interactively tuning the camera settings. It also allows selecting the spots to track.
display_images – If True, a window displays the acquired images in low resolution during the test. This display is mainly intended for debugging and visual follow-up, but not for displaying high-quality images.
displayer_backend –
If display_images is True, the backend to use for the display window. Should be one of :
```
'cv2', 'mpl'
```
If not given, OpenCV will be used if available.
displayer_framerate – If display_images is True, sets the maximum framerate for updating the display window. This setting allows limiting the resources used by the displayer. Note that the actual achieved framerate might differ, this is just the maximum limit.
verbose – If True, the achieved framerate will be displayed in the console during the test.
freq – If given, the block will try to loop at this frequency. If it is lower than the framerate of the camera, frames will be dropped. This argument can be used for limiting the achieved framerate when the camera doesn’t support framerate control.
save_images – If True, the acquired images are saved on the computer during the test. Note that saving images uses CPU, so the achieved performance might drop when this feature is in use.
img_name – If save_images is True, the template for naming the recorded images. It is evaluated as an f-string, and must contain the file extension at the end. For building the f-string, the self._n_loops attribute holds the loop number, and t-self.t0 holds the current timestamp.
save_folder – If save_images is True, the directory to save images to. If it doesn’t exist, it will be created. If not given, the images are saved in a folder named Crappy_images and created next to the file being run.
save_period – If save_images is True, only one every this number of images will be saved.
save_backend –
The backend to use for saving the images. Should be one of :
```
'sitk', 'pil', 'cv2'
```
If not specified, SimpleITK will be used if available, then OpenCV as a second choice, and finally Pillow if none of the others was available.
image_generator – A function taking two floats as arguments, and returning an image. It is only used for demonstration without camera in the examples, and isn’t meant to be used in an actual test.
labels –
A list containing the labels to send to downstream blocks, carrying the information about the position of the tracked spots and the strain values. If not given, the default labels are :
```
['t(s)', 'meta', 'Coord(px)', 'Eyy(%)', 'Exx(%)']
```
raise_on_lost_spot – If True, an exception is raised as soon as the block is losing track of a spot, what causes the test to stop. Otherwise, the block simply stops processing incoming images but doesn’t raise any exception.
white_spots – If True, detects white objects on a black background, else black objects on a white background.
update_thresh – If True, the threshold for detecting the spots is re-calculated for 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, between 1 and 4. The class will then try to detect this exact number of spots, and won’t work if not enough spots can be found. If this argument is not given, at most 4 spots can be detected but the class will work with any number of detected spots between 1 and 4.
safe_mode – If True, the class will stop and raise an exception as soon as overlapping is 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 class 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 too big, 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 camera and the size of the spots to detect.
blur – The size in pixels of the kernel to use for applying a median blur 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 to pass to the camera.

finish() → None[source]: Stopping the tracker processes, the displayer and the camera.

prepare() → None[source]: Opening the camera, starting the Video Extenso config and the tracker processes.