import numpy as np
import yaml
import cv2
[docs]
class Parameters:
"""
Container for all the parameter options used to perform PIV.
Default values are defined here. There are two ways to change the
values for a specific DPIV run:
- Change the specific value manually on python shell or on the running
script.
Example: Parameters.box_size_1_x = 128
- Load all the parameters included in a yaml file by using readParameters
classmethod.
Example: Parameters.readParameters('folder/filename')
"""
# Default first step parameters
box_size_1_x = 64 # Cross-correlation box1
box_size_1_y = 64 # Cross-correlation box1
no_boxes_1_x = 64 # Number of x-windows
no_boxes_1_y = 32 # Number of y-windows
window_1_x = 48
window_1_y = 48
# Default second step parameters
box_size_2_x = 32 # Cross-correlation box 2
box_size_2_y = 32 # Cross-correlation box 2
no_boxes_2_x = 128 # Number of x-windows
no_boxes_2_y = 64 # Number of y-windows
window_2_x = 32
window_2_y = 32
# Number of passes of first step
no_iter_1 = 1
# Number of passes of second step
no_iter_2 = 2
# Direct calculation or FFT
direct_calc = False # True=direct; False=FFT
# Default general parameters
mask = False
stereo = False
peak_ratio = 1
weighting = False
gaussian_size = 0
median_limit = 0.5
calibration = 1
delta_t = 1
# Extra data needed in some cases (mask array, stereo calibration, etc.)
[docs]
class Data:
"""Holds run-specific data such as the mask and its file paths."""
# Path of mask images
path_mask = "none"
path_stereo = "none"
[docs]
class Stereo_Calibration:
"""
Loads a stereo calibration from an .npz file: the pixel->world
forward fits (fwd_*) and the Soloff world->pixel maps (map_*)
for both cameras, plus the camera configuration.
"""
def __init__(self, npz_file):
"""Load the calibration from npz_file (see load_variables)."""
self.load_variables(npz_file)
[docs]
def load_variables(self, npz_file):
"""Read the calibration arrays from npz_file into attributes."""
temp = np.load(npz_file)
if 'fwd_a_l' not in temp:
raise ValueError(
"Calibration file uses the old DLT format. "
"Please re-run stereo calibration to generate a new file.")
self.fwd_a_l = temp['fwd_a_l']
self.fwd_b_l = temp['fwd_b_l']
self.fwd_a_r = temp['fwd_a_r']
self.fwd_b_r = temp['fwd_b_r']
self.camera_config = str(temp['camera_config']) if 'camera_config' in temp else 'Front-Front'
# Soloff world->pixel maps (9-term, with Z cross-terms) for
# soloff_velocity. Absent in old calibration files -> None, and
# soloff_velocity will ask the user to re-run calibration.
self.map_a_l = temp['map_a_l'] if 'map_a_l' in temp else None
self.map_b_l = temp['map_b_l'] if 'map_b_l' in temp else None
self.map_a_r = temp['map_a_r'] if 'map_a_r' in temp else None
self.map_b_r = temp['map_b_r'] if 'map_b_r' in temp else None
[docs]
@classmethod
def readParameters(cls, fileName):
"""
Load all PIV parameters from a yaml file and set them on the
class (overwriting the defaults).
Parameters
----------
fileName : str
Path to the yaml parameter file.
"""
with open(fileName) as f:
data = yaml.load(f, Loader=yaml.FullLoader)
print("starting")
# Default first step parameters
cls.box_size_1_x = data['box_size_1_x']
cls.box_size_1_y = data['box_size_1_y']
cls.no_boxes_1_x = data['no_boxes_1_x']
cls.no_boxes_1_y = data['no_boxes_1_y']
cls.window_1_x = data['window_1_x']
cls.window_1_y = data['window_1_y']
# Number of passes
cls.no_iter_1 = data['no_iter_1']
cls.no_iter_2 = data['no_iter_2']
# Direct calculation or FFT
cls.direct_calc = data['direct_calc']
# Default second step parameters
cls.box_size_2_x = data['box_size_2_x']
cls.box_size_2_y = data['box_size_2_y']
cls.no_boxes_2_x = data['no_boxes_2_x']
cls.no_boxes_2_y = data['no_boxes_2_y']
cls.window_2_x = data['window_2_x']
cls.window_2_y = data['window_2_y']
# Default general parameters
try:
cls.mask = data['mask']
except:
cls.mask = False
try:
cls.stereo = data['stereo']
except:
cls.stereo = False
cls.peak_ratio = data['peak_ratio']
cls.weighting = data['weighting']
cls.gaussian_size = data['gaussian_size']
cls.median_limit = data['median_limit']
cls.calibration = data['calibration']
cls.delta_t = data['delta_t']
# Extra data
if cls.mask:
if data['path_mask'].endswith('.np'):
cls.Data.mask = bool(np.load(data['path_mask']))
else:
cls.Data.mask = np.asarray(cv2.cvtColor(cv2.imread(
data['path_mask']), cv2.COLOR_BGR2GRAY)).astype(bool)
if cls.stereo:
cls.stereo_calibration = cls.Stereo_Calibration(
data['path_stereo'])
[docs]
def introParameters():
"""
Introduce a parameter manually, not implemented yet (probably
needed for a GUI)
"""
pass
[docs]
class grid:
"""
Holds the PIV grids (window centers and box origins) for both
passes, built from the image size and the current Parameters.
"""
[docs]
@classmethod
def generate_mesh(cls, width, height):
"""
Build the meshgrid of x and y positions of the correlation
windows for the two passes, according to the selected PIV
parameters, and store them as class attributes.
Parameters
----------
width, height : int
Image size in pixels.
"""
pixels = width * height
no_boxes_1_x = Parameters.no_boxes_1_x
no_boxes_1_y = Parameters.no_boxes_1_y
box_size_1_x = Parameters.box_size_1_x
box_size_1_y = Parameters.box_size_1_y
no_boxes_2_x = Parameters.no_boxes_2_x
no_boxes_2_y = Parameters.no_boxes_2_y
box_size_2_x = Parameters.box_size_2_x
box_size_2_y = Parameters.box_size_2_y
# Obtain PIV mesh for calculations
box_origin_x_1 = (1 + np.round((np.arange(0, no_boxes_1_x)
* (width - box_size_1_x - 2)) / (no_boxes_1_x - 1)).astype(np.int32))
box_origin_y_1 = (1 + np.round((np.arange(0, no_boxes_1_y)
* (height - box_size_1_y - 2)) / (no_boxes_1_y - 1)).astype(np.int32))
x_1 = (box_origin_x_1 - 1 + box_size_1_x / 2).astype(np.int32)
y_1 = (box_origin_y_1 - 1 + box_size_1_y / 2).astype(np.int32)
x_1, y_1 = np.meshgrid(x_1, y_1)
box_origin_x_1, box_origin_y_1 = np.meshgrid(
box_origin_x_1, box_origin_y_1)
# Obtain special mesh for direct calculation if needed
if Parameters.direct_calc:
window_x = Parameters.window_1_x
window_y = Parameters.window_1_y
x_1 = (1 + np.round((window_x / 2 + box_size_1_x) / 2) +
np.round(np.arange(0, no_boxes_1_x) *
(width - box_size_1_x - window_x / 2 - 4) / (no_boxes_1_x - 1)))
y_1 = (1 + np.round((window_y / 2 + box_size_1_y) / 2) +
np.round(np.arange(0, no_boxes_1_y) *
(height - box_size_1_y - window_y / 2 - 4) / (no_boxes_1_y - 1)))
x_1, y_1 = np.meshgrid(x_1, y_1)
box_origin_x_d = x_1 + 1 - round(box_size_1_x / 2)
box_origin_y_d = y_1 + 1 - round(box_size_1_y / 2)
cls.box_origin_x_d = box_origin_x_d.astype(np.int32)
cls.box_origin_y_d = box_origin_y_d.astype(np.int32)
x_margin = 3 / 2 * np.amax([box_size_1_x, box_size_2_x])
y_margin = 3 / 2 * np.amax([box_size_1_y, box_size_2_y])
# Second grid is placed completely inside first one
x_2 = np.int32(2 + np.round(np.arange(0, no_boxes_2_x) *
(width - x_margin - 6) / (no_boxes_2_x - 1) + x_margin / 2))
y_2 = np.int32(2 + np.round(np.arange(0, no_boxes_2_y) *
(height - y_margin - 6) / (no_boxes_2_y - 1) + y_margin / 2))
x_2, y_2 = np.meshgrid(x_2, y_2)
box_origin_x_2 = (x_2 - box_size_2_x / 2).astype(np.int32)
box_origin_y_2 = (y_2 - box_size_2_y / 2).astype(np.int32)
cls.x_1 = x_1.astype(np.int32)
cls.y_1 = y_1.astype(np.int32)
cls.x_2 = x_2.astype(np.int32)
cls.y_2 = y_2.astype(np.int32)
cls.box_origin_x_1 = box_origin_x_1.astype(np.int32)
cls.box_origin_y_1 = box_origin_y_1.astype(np.int32)
cls.box_origin_x_2 = box_origin_x_2.astype(np.int32)
cls.box_origin_y_2 = box_origin_y_2.astype(np.int32)
# Create mask mesh if needed
cls.mask_1 = np.full(x_1.shape, False)
cls.mask_2 = np.full(x_2.shape, False)
if Parameters.mask:
for j in range(0, len(x_1[:, 0])):
for i in range(0, len(x_1[0, :])):
if Parameters.Data.mask[int(y_1[j, i]), int(x_1[j, i])]:
cls.mask_1[j, i] = True
for j in range(0, len(x_2[:, 0])):
for i in range(0, len(x_2[0, :])):
if Parameters.Data.mask[int(y_2[j, i]), int(x_2[j, i])]:
cls.mask_2[j, i] = True
[docs]
def read_mesh(self, height, width):
"""
Read the positions of a custom mesh from a file (not implemented yet)
"""
pass
[docs]
class Synt_Img():
"""
Contains the parameters to generate a pair of synthetic images
following a given flow velocity field. The default values can be
changed manually.
Attributes
----------
width : int
Number of pixels in the x-direction of the generated image.
height : int
Number of pixels in the y-direction of the generated image.
trazers_density : float
Number of particles generated per pixel. Must be a float
between 0 and 1.
vel : float
Characteristic velocity of the canonical flow used to generate
the particle displacement.
Note: it has a different definition for each flow.
Shine_m : int
Mean brightness of the tracer core (from 0 to 255).
d_Shine : int
Triangular variability of the tracer brightness.
D_m : int or float
Mean diameter of the tracers.
d_D : int or float
Triangular variability of the tracer diameter.
noise_m : int
Mean of the random noise.
d_noise : int
Triangular variability of the image noise.
vel_profile : str
Name of the flow velocity field used. The following options are
available:
Constant: Flow with a constant displacement of "vel" pixels on
the x-direction
Couette: Couette flow in x-direction using a moving condition
of "vel" pixels between images on top wall. Bottom
limit of the images is not moving.
Poiseuille: Poiseuille flow in x direction with no-slip condition
on top and bottom of the images. The parameter "vel"
indicates the maximum velocity of the flow in pixels.
Vortex: Flow generated by a Scully vortex (see Scully 1975). Vel
is the maximum velocity of the vortex in pixel/frame.
Frequency: Spatial frequency wave along the y-direction. This
flow allows testing the PIV frequency response (see
Scarano & Riethmuller, 2000). In this case the
maximum pixel displacement on the wave is always
2 pixels, and the parameter "vel" indicates the
wavelength.
ext : str
Extension used to save the images.
"""
width = 1024 # Width of generated image
height = 1024 # Height of generated image
vel_profile = 'Vortex' # Constant, Couette, Poiseuille, Vortex
vel = 8 # Velocity in pixels/frame
amp = 2*np.pi*20 # Amplitude for the Frequency case
n_steps = 10 # Runge-Kutta 4 subsampling steps
trazers_density = 0.05 # Number of tracers/pixel
Shine_m = 230 # Mean brightness of tracers
d_Shine = 80 # Variability for random tracer brightness
D_m = 4 # Mean diameter of tracers (in pixels)
d_D = 3 # Variability for random tracer diameter
noise_m = 1 # Mean white noise of the image
d_noise = 1 # Variability of the image noise
ext = '.png' # Image save format
[docs]
class GPU():
"""
Namespace for the GPU-side data. The OpenCL buffers and compiled
kernels are attached as attributes at runtime by
Cl_DPIV.compile_Kernels / initialization / processing.
"""
[docs]
def gpu_data(self, thr):
"""Placeholder hook for attaching GPU data to the reikna Thread
``thr``; the buffers/kernels are populated by Cl_DPIV instead."""
pass