pyvalib.py

import numpy as np
import ctypes
import os
import sys
 
#--------------------------------------------------------------------------------------------#
# A Python wrapper for the VALIB library
#--------------------------------------------------------------------------------------------#
 
 
# Set this flag to switch between 64-bit and 32-bit floating point precision. Note that the precision
# has to be set consistently both here in the plugin and in the VALIB library. 
useDoublePrecision = True
c_int_p = ctypes.POINTER(ctypes.c_int)
if useDoublePrecision:
    c_real = ctypes.c_double
    c_real_p = ctypes.POINTER(ctypes.c_double)
    np_real = np.float64
else:
    c_real = ctypes.c_float      
    c_real_p = ctypes.POINTER(ctypes.c_float)
    np_real = np.float32
 
def get_valib():
    # Load the VALIB library using ctypes
    wrapperpath = os.path.dirname(__file__)
    if sys.platform.startswith("win"):
        cLib = ctypes.CDLL(wrapperpath + "/../lib/libvalib.dll")
    else:
        cLib = ctypes.CDLL(wrapperpath + "/../lib/libvalib.so")
    return cLib
 
def valib_real_scalar_criterion(gradU, method_name):
    # Get the VALIB function
    func = getattr(get_valib(), method_name)
    func.argtypes = [c_real_p, c_real_p]
    func.restype = None
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    result_ct = c_real()
 
    # Call VALIB
    func(gradU_ct, ctypes.byref(result_ct))
 
    return result_ct.value
 
def valib_real_scalar_criterion_batch(gradU_arr, method_name):
    # Get the VALIB function
    func = getattr(get_valib(), method_name)
    func.argtypes = [c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output array
    n_tuples = gradU_arr.shape[0]
    result_arr = np.zeros(n_tuples)
 
    # Call VALIB for each tuple
    for i in range(n_tuples):
        # Get ctypes variables
        gradU_ct = gradU_arr[i].astype(np_real, copy=False).ctypes.data_as(c_real_p)
        result_ct = c_real()
 
        func(gradU_ct, ctypes.byref(result_ct))
        result_arr[i] = result_ct.value
 
    return result_arr
 
def valib_delta(gradU):
    return valib_real_scalar_criterion(gradU, "valib_delta")
 
def valib_delta_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_delta")
 
def valib_lambda2(gradU):
    return valib_real_scalar_criterion(gradU, "valib_lambda2")
 
def valib_lambda2_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_lambda2")
 
def valib_qcriterion(gradU):
    return valib_real_scalar_criterion(gradU, "valib_qcriterion")
 
def valib_qcriterion_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_qcriterion")
 
def valib_qdcriterion(gradU):
    return valib_real_scalar_criterion(gradU, "valib_qdcriterion")
 
def valib_qdcriterion_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_qdcriterion")
 
def valib_qmcriterion(gradU):
    return valib_real_scalar_criterion(gradU, "valib_qmcriterion")
 
def valib_qmcriterion_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_qmcriterion")
 
def valib_qwcriterion(gradU):
    return valib_real_scalar_criterion(gradU, "valib_qwcriterion")
 
def valib_qwcriterion_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_qwcriterion")
 
def valib_compactness(gradU):
    return valib_real_scalar_criterion(gradU, "valib_compactness")
 
def valib_compactness_batch(gradU_arr):
    return valib_real_scalar_criterion_batch(gradU_arr, "valib_compactness")
 
def valib_swirling_strength(gradU):
    # Get the VALIB function
    func = get_valib().valib_swirling_strength
    func.argtypes = [c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    lambdaCr = c_real()
    lambdaCi = c_real()
 
    # Call VALIB
    func(gradU_ct, ctypes.byref(lambdaCi), ctypes.byref(lambdaCr))
 
    return lambdaCr.value, lambdaCi.value
 
def valib_swirling_strength_batch(gradU_arr):
    # Get the VALIB function
    func = get_valib().valib_swirling_strength
    func.argtypes = [c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    lambda_cr_arr = np.zeros(n_tuples)
    lambda_ci_arr = np.zeros(n_tuples)
 
    # Call VALIB for each tuple
    for i in range(n_tuples):
        # Get ctypes variables
        gradU_ct = gradU_arr[i].astype(np_real, copy=False).ctypes.data_as(c_real_p)
        lambda_cr = c_real()
        lambda_ci = c_real()
 
        func(gradU_ct, ctypes.byref(lambda_ci), ctypes.byref(lambda_cr))
 
        lambda_cr_arr[i] = lambda_cr.value
        lambda_ci_arr[i] = lambda_ci.value
 
    return lambda_cr_arr, lambda_ci_arr
 
def valib_vector_criterion(gradU, method_name):
    # Get the VALIB function
    func = getattr(get_valib(), method_name)
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_real_p]
    func.restype = None
 
    # Allocate output variable
    result = np.zeros(3, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    result_ct = result.ctypes.data_as(c_real_p)
 
    # Call VALIB
    func(gradU_ct, result_ct)
 
    return result
 
def valib_vector_criterion_batch(gradU_arr, method_name, accelerator):
    # Get the VALIB function
    func = getattr(get_valib(), method_name)
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    result_arr = np.zeros((n_tuples, 3), dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU_arr.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    result_arr_ct = result_arr.ctypes.data_as(c_real_p)
    accelerator_ct = ctypes.c_int(accelerator)
    n_tuples_ct = ctypes.c_int(n_tuples)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), gradU_ct, ctypes.byref(n_tuples_ct), result_arr_ct)
 
    return result_arr
 
def valib_corotation(gradU):
    return valib_vector_criterion(gradU, "valib_corotation")
 
def valib_corotation_batch(gradU_arr, accelerator):
    return valib_vector_criterion_batch(gradU_arr, "valib_corotation", accelerator)
 
def valib_contrarotation(gradU):
    return valib_vector_criterion(gradU, "valib_contrarotation")
 
def valib_contrarotation_batch(gradU_arr, accelerator):
    return valib_vector_criterion_batch(gradU_arr, "valib_contrarotation", accelerator)
 
def valib_rortex(gradU):
    return valib_vector_criterion(gradU, "valib_rortex")
 
def valib_rortex_batch(gradU_arr, accelerator):
    return valib_vector_criterion_batch(gradU_arr, "valib_rortex", accelerator)
 
def valib_triple_decomposition(gradU):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition
    func.argtypes = [c_real_p, c_int_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = c_real()
    res_strain_ct = c_real()
    shear_ct = c_real()
 
    # Call VALIB
    func(gradU_ct, ctypes.byref(res_vorticity_ct), ctypes.byref(res_strain_ct), ctypes.byref(shear_ct))
 
    return res_vorticity_ct.value, res_strain_ct.value, shear_ct.value
 
def valib_triple_decomposition_batch(gradU_arr, n_steps, accelerator):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition_batch
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_int_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    res_vorticity = np.zeros((n_tuples, 3), dtype=np_real)
    res_strain = np.zeros(n_tuples, dtype=np_real)
    shear = np.zeros(n_tuples, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU_arr.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = res_vorticity.ctypes.data_as(c_real_p)
    res_strain_ct = res_strain.ctypes.data_as(c_real_p)
    shear_ct = shear.ctypes.data_as(c_real_p)
    accelerator_ct = ctypes.c_int(accelerator)
    n_tuples_ct = ctypes.c_int(n_tuples)
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), 
         gradU_ct, 
         ctypes.byref(n_tuples_ct), 
         ctypes.byref(n_steps_ct), 
         res_vorticity_ct, 
         res_strain_ct, 
         shear_ct)
 
    return res_vorticity, res_strain, shear
 
def valib_residual_vorticity(gradU, n_steps, accelerator):
    # Get the VALIB function
    func = get_valib().valib_residual_vorticity
    func.argtypes = [c_real_p, c_int_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output variables
    res_vorticity = np.zeros(3, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = res_vorticity.ctypes.data_as(c_real_p)
    res_strain_ct = c_real()
    shear_ct = c_real()
    accelerator_ct = ctypes.c_int(accelerator)
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), 
         gradU_ct,
         ctypes.byref(n_steps_ct), 
         res_vorticity_ct, 
         ctypes.byref(res_strain_ct), 
         ctypes.byref(shear_ct))
 
    return res_vorticity, res_strain_ct.value, shear_ct.value
 
def valib_residual_vorticity_batch(gradU_arr, n_steps, accelerator):
    # Get the VALIB function
    func = get_valib().valib_residual_vorticity_batch
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_int_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    res_vorticity = np.zeros((n_tuples, 3), dtype=np_real)
    res_strain = np.zeros(n_tuples, dtype=np_real)
    shear = np.zeros(n_tuples, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU_arr.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = res_vorticity.ctypes.data_as(c_real_p)
    res_strain_ct = res_strain.ctypes.data_as(c_real_p)
    shear_ct = shear.ctypes.data_as(c_real_p)
    accelerator_ct = ctypes.c_int(accelerator)
    n_tuples_ct = ctypes.c_int(n_tuples)
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), 
         gradU_ct, 
         ctypes.byref(n_tuples_ct), 
         ctypes.byref(n_steps_ct), 
         res_vorticity_ct, 
         res_strain_ct, 
         shear_ct)
 
    return res_vorticity, res_strain, shear
 
def valib_triple_decomposition_4norms(gradU, n_steps):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition_4norms
    func.argtypes = [c_real_p, c_int_p, c_real_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = c_real()
    res_strain_ct = c_real()
    shear_vort_ct = c_real()
    shear_strain_ct = c_real()
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(gradU_ct, 
         ctypes.byref(n_steps_ct), 
         ctypes.byref(res_vorticity_ct), 
         ctypes.byref(res_strain_ct), 
         ctypes.byref(shear_vort_ct),
         ctypes.byref(shear_strain_ct))
 
    return res_vorticity_ct.value, res_strain_ct.value, shear_vort_ct.value, shear_strain_ct.value
 
def valib_triple_decomposition_4norms_batch(gradU_arr, n_steps, accelerator):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition_4norms_batch
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_int_p, c_real_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    res_vorticity = np.zeros(n_tuples, dtype=np_real)
    res_strain = np.zeros(n_tuples, dtype=np_real)
    shear_vort = np.zeros(n_tuples, dtype=np_real)
    shear_strain = np.zeros(n_tuples, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU_arr.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = res_vorticity.ctypes.data_as(c_real_p)
    res_strain_ct = res_strain.ctypes.data_as(c_real_p)
    shear_vort_ct = shear_vort.ctypes.data_as(c_real_p)
    shear_strain_ct = shear_strain.ctypes.data_as(c_real_p)
    accelerator_ct = ctypes.c_int(accelerator)
    n_tuples_ct = ctypes.c_int(n_tuples)
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), 
         gradU_ct, 
         ctypes.byref(n_tuples_ct), 
         ctypes.byref(n_steps_ct), 
         res_vorticity_ct, 
         res_strain_ct, 
         shear_vort_ct,
         shear_strain_ct)
 
    return res_vorticity, res_strain, shear_vort, shear_strain
 
def valib_triple_decomposition_with_angles(gradU, n_steps):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition_with_angles
    func.argtypes = [c_real_p, c_int_p, c_real_p, c_real_p, c_real_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Get ctypes variables
    gradU_ct = gradU.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = c_real()
    res_strain_ct = c_real()
    shear_ct = c_real()
    alpha_ct = c_real()
    beta_ct = c_real()
    gamma_ct = c_real()
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(gradU_ct, 
         ctypes.byref(n_steps_ct), 
         ctypes.byref(res_vorticity_ct), 
         ctypes.byref(res_strain_ct), 
         ctypes.byref(shear_ct),
         ctypes.byref(alpha_ct),
         ctypes.byref(beta_ct),
         ctypes.byref(gamma_ct))
 
    return res_vorticity_ct.value, res_strain_ct.value, shear_ct.value, alpha_ct.value, beta_ct.value, gamma_ct.value
 
def valib_triple_decomposition_with_angles_batch(gradU_arr, n_steps, accelerator):
    # Get the VALIB function
    func = get_valib().valib_triple_decomposition_with_angles_batch
    func.argtypes = [c_int_p, c_real_p, c_int_p, c_int_p, c_real_p, c_real_p, c_real_p, c_real_p, c_real_p, c_real_p]
    func.restype = None
 
    # Allocate output arrays
    n_tuples = gradU_arr.shape[0]
    res_vorticity = np.zeros(n_tuples, dtype=np_real)
    res_strain = np.zeros(n_tuples, dtype=np_real)
    shear = np.zeros(n_tuples, dtype=np_real)
    alpha = np.zeros(n_tuples, dtype=np_real)
    beta = np.zeros(n_tuples, dtype=np_real)
    gamma = np.zeros(n_tuples, dtype=np_real)
 
    # Get ctypes variables
    gradU_ct = gradU_arr.astype(np_real, copy=False).ctypes.data_as(c_real_p)
    res_vorticity_ct = res_vorticity.ctypes.data_as(c_real_p)
    res_strain_ct = res_strain.ctypes.data_as(c_real_p)
    shear_ct = shear.ctypes.data_as(c_real_p)
    alpha_ct = alpha.ctypes.data_as(c_real_p)
    beta_ct = beta.ctypes.data_as(c_real_p)
    gamma_ct = gamma.ctypes.data_as(c_real_p)
    accelerator_ct = ctypes.c_int(accelerator)
    n_tuples_ct = ctypes.c_int(n_tuples)
    n_steps_ct = ctypes.c_int(n_steps)
 
    # Call VALIB
    func(ctypes.byref(accelerator_ct), 
         gradU_ct, 
         ctypes.byref(n_tuples_ct), 
         ctypes.byref(n_steps_ct), 
         res_vorticity_ct, 
         res_strain_ct, 
         shear_ct,
         alpha_ct,
         beta_ct,
         gamma_ct)
 
    return res_vorticity, res_strain, shear, alpha, beta, gamma