Source code for fluidsimfoam.invoke.tasks

"""Invoke tasks to be used from the ```` file in the solvers

.. autodata:: clean
.. autodata:: block_mesh
.. autodata:: surface_feature_extract
.. autodata:: snappy_hex_mesh
.. autodata:: polymesh
.. autodata:: set_fields
.. autodata:: decompose_par
.. autodata:: run


import re
from pathlib import Path
from subprocess import Popen
from time import sleep, time

import invoke.context
from invoke import task
from invoke.exceptions import Exit

from fluidsimfoam.foam_input_files import parse

from .context import Context

invoke.tasks.Context = Context

[docs]@task def clean(context): """clean with foamCleanTutorials""""foamCleanTutorials", warn=True) for path in Path(".").glob(".data_fluidsim/*_called*"): path.unlink()
[docs]@task def block_mesh(context): """Run ``blockMesh``""" context.run_appl_once("blockMesh")
[docs]@task def surface_feature_extract(context): """Run ``surfaceFeatureExtract``""" context.run_appl_once("surfaceFeatureExtract")
[docs]@task def snappy_hex_mesh(context): """Run ``snappyHexMesh -overwrite``""" context.run_appl_once( "snappyHexMesh -overwrite", parallel_if_needed=True, path_decompose_par_dict=PATH_DECOMPOSE_PAR_DICT_MESH, )
[docs]@task(pre=[block_mesh, surface_feature_extract, snappy_hex_mesh]) def polymesh(context): """Create the polymesh directory"""
[docs]@task def set_fields(context, force=False): """Run ``setFields``""" context.run_appl_once("setFields")
[docs]@task def decompose_par(context): """Run ``decomposePar``""" context.run_appl_once("decomposePar")
[docs]@task(pre=[polymesh, set_fields, decompose_par]) def run(context): """Main target to launch a simulation""" with open("system/controlDict") as file: ctr_dict = ctr_dict = parse(ctr_dict) application = ctr_dict.children["application"] end_time = ctr_dict["endTime"] start_time = ctr_dict["startTime"] path_run = Path.cwd() if context.parallel: # Options '-n' and '-np' are synonymous, but msmpi only supports '-n' command = [ context.mpi_command, "-n", str(context.nsubdoms), application, "-parallel", ] else: command = application path_log = path_run / f"log_{context.time_as_str}.txt" print(f"Starting simulation in \n{path_run}") with open(path_log, "w") as file_log: file_log.write(f"{start_time = }\n{end_time = }\n") file_log.flush() print(f"{end_time = }") pattern_time = re.compile(r"\nTime = ([\d]+\.?[\d]+)") t_start = time() process = Popen(command, stdout=file_log, stderr=file_log) t_last = time() - 10.0 while process.poll() is None: sleep(0.2) t_now = time() if t_now - t_last > 2: t_last = t_now log_size = path_log.stat().st_size with open(path_log, "r") as file_log_read:, log_size - 1000)) log = if log: groups = pattern_time.findall(log) if groups: time_simul = float(groups[-1]) print( f"eq_time: {time_simul:12.3f} " f"({100 * time_simul/end_time:6.2f} %), " f"clock_time: {t_now - t_start: 12.3f} s" ) print(f"Simulation done. path_run:\n{path_run}") if process.returncode: with open(path_log, "r") as file_log_read:, log_size - 5000)) tail = print( f"Error: simulation finished with return code {process.returncode}\n" f"Tail of log file:\n{tail}" ) raise Exit(process.returncode)