# -*- coding: utf-8 -*-
#
# This file is subject to the terms and conditions defined in
# file 'LICENSE.txt', which is part of this source code package.
#
#
import multiprocessing as mp
import os
import pickle
from functools import partial
from typing import List
from Muscat.Bridges.CGNSBridge import CGNSToMesh, MeshToCGNS
from Muscat.Containers.Mesh import Mesh
from plaid.containers.dataset import Dataset
from plaid.containers.sample import Sample
from tqdm import tqdm
from mmgp.morphing import Morphing
from mmgp.utils import reset_folder
from mmgp.FE_utils import compute_FE_projection_operators
import plaid
[docs]
def pretreat_sample(configuration: dict, common_morphed_mesh: Mesh, i_sample: int) -> dict:
"""Compute and Save the precomputed FE projection and inverse FE projection operators.
Args:
configuration (dict): A dictionary containing various parameters and settings for the pretreatment process. It should include the following keys:
- 'case_name' (str): The name or identifier for the specific case or scenario being analyzed.
- 'base_name' (str): The name of a specific base on which to pretreat samples.
- 'zone_name' (str): The name of a specific zone on which to pretreat samples.
- 'morphing' (dict): A dictionary containing morphing-related parameters with the following keys:
- algo (str): A string indicating the morphing algorithm to be used. This parameter specifies the technique for mesh parametrization or transformation.
- options (str): A string containing options or settings for the chosen morphing algorithm.
- init_dataset_location (str): A string specifying the location or path to the initial dataset or data source used for the calculations.
- generated_data_folder (str): A string representing the folder where the morphed mesh and transported fields, the common morphed mesh and projected coordinate fields and fields of interest, and precomputed FE projection and inverse FE projection operators will be saved.
common_morphed_mesh (Mesh): The target common morphed mesh.
i_sample (int): An integer representing the index or identifier of the sample being processed.
Hint:
The following samples will alsos be saved in the generated data folder:
- Sample with morphed mesh and transported fields
- Sample with common morphed mesh and projected coordinate fields and fields of interest
Caution:
This function will load the initial PLAID dataset. Make sure it has been created and is located correctly.
"""
case_name = configuration['case_name']
base_name = configuration["base_name"]
zone_name = configuration["zone_name"]
morphing_algo = configuration["morphing"]["algo"]
morphing_options = configuration["morphing"]["options"]
init_dataset_location = configuration['init_dataset_location']
generated_data_folder = configuration['generated_data_folder']
# Load the provided dataset
dataset = Dataset()
dataset._load_from_dir_(
os.path.join(
init_dataset_location,
"dataset"),
ids=[i_sample])
field_names = dataset.get_field_names(
zone_name=zone_name, base_name=base_name)
if "OriginalIds" in field_names:
field_names.remove("OriginalIds")
scalar_names = dataset.get_scalar_names()
sample = dataset[i_sample]
cgns_mesh = sample.get_mesh()
mesh = CGNSToMesh(cgns_mesh)
# Compute the morphing of the current mesh
morphing = Morphing(algo=morphing_algo, options=morphing_options)
morphed_mesh = morphing.transform(mesh)
coord_names = ["X", "Y", "Z"]
# Create and save a sample with morphed mesh and transported fields
morph_sample = Sample()
tree = MeshToCGNS(morphed_mesh, exportOriginalIDs=False, tagsAsFields=False)
morph_sample.add_tree(tree)
for fname in field_names:
field = sample.get_field(fname, zone_name, base_name)
morph_sample.add_field(fname, field, zone_name, base_name)
for j in range(mesh.GetPointsDimensionality()):
coord_field = mesh.nodes[:, j]
morph_sample.add_field(
"coord_" +
coord_names[j],
coord_field,
zone_name,
base_name)
for sname in scalar_names:
scalar = sample.get_scalar(sname)
morph_sample.add_scalar(sname, scalar)
morph_sample.save(
os.path.join(
generated_data_folder,
f"{case_name}_morphed/dataset/samples/sample_{i_sample:09d}"))
# Compute the two FE projection operators
proj_operator, inv_proj_operator = compute_FE_projection_operators(
morphed_mesh, common_morphed_mesh)
# Create and save a sample with common morphed mesh and projected
# coordinate fields and fields of interest
morph_proj_sample = Sample()
tree = MeshToCGNS(
common_morphed_mesh,
exportOriginalIDs=False,
tagsAsFields=False)
morph_proj_sample.add_tree(tree)
for fname in field_names:
field = sample.get_field(fname, zone_name, base_name)
morph_proj_sample.add_field(
fname, proj_operator.dot(field), zone_name, base_name)
for j in range(mesh.GetPointsDimensionality()):
coord_field = mesh.nodes[:, j]
morph_proj_sample.add_field(
"coord_" + coord_names[j],
proj_operator.dot(coord_field),
zone_name,
base_name)
for sname in scalar_names:
scalar = sample.get_scalar(sname)
morph_proj_sample.add_scalar(sname, scalar)
morph_proj_sample.save(
os.path.join(
generated_data_folder,
f"{case_name}_morphed_and_projected/dataset/samples/sample_{i_sample:09d}"))
# Save precomputed FE projection and inverse FE projection operators
FE_interpolation_operators = {
"projOperator": proj_operator,
"invProjOperator": inv_proj_operator
}
with open(os.path.join(generated_data_folder, "FEInterpolationOperators", f"sample_{i_sample:09d}.pkl"), 'wb') as file:
pickle.dump(FE_interpolation_operators, file)
return FE_interpolation_operators
[docs]
def pre_process(configuration: dict) -> List[dict]:
"""Parallel apply pretreat_sample function on the entire Dataset.
Args:
configuration (dict): A dictionary containing various parameters and settings for the pretreatment process. It should include the following keys:
- 'init_dataset_location' (str): A string specifying the location or path to the initial dataset or data source used for the calculations.
- 'generated_data_folder' (str): A string representing the folder where the morphed mesh and transported fields, the common morphed mesh, projected coordinate fields, fields of interest, and precomputed FE projection and inverse FE projection operators will be saved.
- 'case_name' (str): The name or identifier for the specific case or scenario being analyzed.
- 'morphing' (dict): A dictionary containing morphing-related parameters with the following keys:
- 'algo' (str): A string indicating the morphing algorithm to be used.
- 'options' (str): A string containing options or settings for the chosen morphing algorithm.
- 'common_mesh_index' (int): An integer specifying the index for the common mesh.
Caution:
This function will load the initial PLAID dataset. Make sure it has been created and is located correctly.
"""
init_dataset_location = configuration['init_dataset_location']
generated_data_folder = configuration['generated_data_folder']
case_name = configuration['case_name']
morphing_algo = configuration["morphing"]["algo"]
morphing_options = configuration["morphing"]["options"]
common_mesh_index = configuration["common_mesh_index"]
if 'verbose' in configuration:
verbose = configuration['verbose']
else:
verbose = True
# Reset workspace
reset_folder(
os.path.join(
generated_data_folder, "FEInterpolationOperators"))
reset_folder(
os.path.join(
generated_data_folder, f"{case_name}_morphed"))
reset_folder(
os.path.join(
generated_data_folder,
f"{case_name}_morphed_and_projected"))
# ---------------------------------------------------------
# Load Dataset
dataset = Dataset()
dataset._load_from_dir_(
os.path.join(
init_dataset_location, "dataset"),
ids=[common_mesh_index],
verbose=verbose)
cgns_mesh = dataset[common_mesh_index].get_mesh()
median_mesh = CGNSToMesh(cgns_mesh)
# Transforms the mesh into a common mesh
morphing = Morphing(
algo = morphing_algo,
options = morphing_options)
common_morphed_mesh = morphing.transform(median_mesh)
# ---------------------------------------------------------
# Thread pool-based sample pre-processing
number_of_samples = plaid.get_number_of_samples(os.path.join(init_dataset_location, "dataset"))
n_cores = mp.cpu_count()
n_parallel_tasks = max(1,int(min(number_of_samples, n_cores)/4))
if verbose:
print("Preprocessing dataset on " + str(n_parallel_tasks) + " threads:")
# result = []
# for i in tqdm(range(number_of_samples)):
# result.append(pretreat_sample(configuration, common_morphed_mesh, i))
with mp.Pool(n_parallel_tasks) as pool:
results = list(tqdm(
pool.imap(partial(pretreat_sample, configuration, common_morphed_mesh), range(number_of_samples)),
total=number_of_samples, disable=not (verbose)
))
return results