:py:mod:`obr.signac_wrapper.operations` ======================================= .. py:module:: obr.signac_wrapper.operations Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: obr.signac_wrapper.operations.OpenFOAMProject Functions ~~~~~~~~~ .. autoapisummary:: obr.signac_wrapper.operations.is_case obr.signac_wrapper.operations.is_job obr.signac_wrapper.operations.operation_complete obr.signac_wrapper.operations.basic_eligible obr.signac_wrapper.operations.parent_job_is_ready obr.signac_wrapper.operations._link_path obr.signac_wrapper.operations.needs_initialization obr.signac_wrapper.operations.initialize_if_required obr.signac_wrapper.operations.get_args obr.signac_wrapper.operations.execute_operation obr.signac_wrapper.operations.execute_post_build obr.signac_wrapper.operations.execute_pre_build obr.signac_wrapper.operations.start_job_state obr.signac_wrapper.operations.end_job_state obr.signac_wrapper.operations.dispatch_pre_hooks obr.signac_wrapper.operations.dispatch_post_hooks obr.signac_wrapper.operations.set_failure obr.signac_wrapper.operations.copy_on_uses obr.signac_wrapper.operations.controlDict obr.signac_wrapper.operations.MultiCase obr.signac_wrapper.operations.blockMesh obr.signac_wrapper.operations.replaceMesh obr.signac_wrapper.operations.shell obr.signac_wrapper.operations.initialConditions obr.signac_wrapper.operations.fvSolution obr.signac_wrapper.operations.fvSchemes obr.signac_wrapper.operations.transportProperties obr.signac_wrapper.operations.turbulenceProperties obr.signac_wrapper.operations.setKeyValuePair obr.signac_wrapper.operations.has_mesh obr.signac_wrapper.operations.decomposePar obr.signac_wrapper.operations.fetchCase obr.signac_wrapper.operations.is_locked obr.signac_wrapper.operations.refineMesh obr.signac_wrapper.operations.allClean obr.signac_wrapper.operations.checkMesh obr.signac_wrapper.operations.get_number_of_procs obr.signac_wrapper.operations.get_values obr.signac_wrapper.operations.run_cmd_builder obr.signac_wrapper.operations.validate_state_impl obr.signac_wrapper.operations.resetCase obr.signac_wrapper.operations.validateState obr.signac_wrapper.operations.runParallelSolver obr.signac_wrapper.operations.runSerialSolver obr.signac_wrapper.operations.archive obr.signac_wrapper.operations.apply Attributes ~~~~~~~~~~ .. autoapisummary:: obr.signac_wrapper.operations.logger obr.signac_wrapper.operations.generate obr.signac_wrapper.operations.simulate .. py:data:: logger .. py:class:: OpenFOAMProject(*args, **kwargs) Bases: :py:obj:`flow.FlowProject` A signac project class specialized for workflow management. This class is used to define, execute, and submit workflows based on operations and conditions. Users typically interact with this class through its command line interface. This is a typical example of how to use this class: .. code-block:: python @FlowProject.operation def hello(job): print('hello', job) FlowProject().main() Parameters ---------- path : str, optional The project directory. By default, the current working directory (Default value = None). environment : :class:`flow.environment.ComputeEnvironment` An environment to use for scheduler submission. If ``None``, the environment is automatically identified. The default is ``None``. entrypoint : dict A dictionary with two possible keys: ``'executable'`` and ``'path'``. The path represents the location of the script file (the script file must call :meth:`FlowProject.main`). The executable represents the location of the Python interpreter used for the execution of :class:`~.BaseFlowOperation` that are Python functions. .. py:attribute:: filtered_jobs :type: list[signac.job.Job] :value: [] .. py:method:: print_operations() .. py:method:: filter_jobs(filters: list[str]) -> list[signac.job.Job] `filter_jobs` accepts a list of filters. The filters will be applied to all jobs inside the `OpenFOAMProject` instance and the filtered jobs will be returned as a list. .. py:method:: query(jobs: list[signac.job.Job], query: list[obr.core.queries.Query]) -> list[dict] return list of job ids as result of `Query`. .. py:method:: set_entrypoint(entrypoint: dict) Sets the entrypoint for a project, this is useful for submit so that submit writes scripts that call obr run -o instead of the default signac run -o call .. py:method:: group_jobs(jobs, view_id_map: dict[str, str], summarize: int = 0) -> dict[str, list[str]] Returns the list of jobs of the given OpenFOAMProject where the last `summarize` levels are grouped together at the corresponding parent view. Returns a `dict[str, list[str]]` which maps the view path to a list of child jobs. .. py:data:: generate .. py:data:: simulate .. py:function:: is_case(job: signac.job.Job) -> bool .. py:function:: is_job(job: signac.job.Job) -> bool checks OBR_JOB is set to the current job.id used to prevent multiple execution of jobs if --job=id is set .. py:function:: operation_complete(job: signac.job.Job, operation: str) -> bool An operation is considered to be complete if an entry in the job document with same arguments exists and state is success .. py:function:: basic_eligible(job: signac.job.Job, operation: str) -> bool Dispatches to standard checks if operations are eligible for given job this includes: - check for lock, to avoid running operations when calling 'obr run' before operation is finished - check if parent case is ready - operation has been requested for job - copy and link files and folder .. py:function:: parent_job_is_ready(job: signac.job.Job) -> str Checks whether the parent of the given job is ready .. py:function:: _link_path(base: pathlib.Path, dst: pathlib.Path, parent_id: str, copy_instead_link: bool) creates file tree under dst with same folder structure as base but all files are relative symlinks .. py:function:: needs_initialization(job: signac.job.Job) -> bool Check if this job has been initialized already, without performing the initialization .. py:function:: initialize_if_required(job: signac.job.Job) -> bool check if this job has been already linked to The default strategy is to link all files. If a file is modified the modifying operations are responsible for unlinking and copying .. py:function:: get_args(job: signac.job.Job, args: Union[dict, str]) -> Union[dict, str] operation can get args either via function call or it statepoint if no args are passed via function the args from the statepoint are taken also args can be just a str in case of shell scripts .. py:function:: execute_operation(job: signac.job.Job, operation_name: str, operations) -> Literal[True] check whether an operation is requested operation can be simple operations defined by a keyword like blockMesh or operations with parameters defined by a dictionary .. py:function:: execute_post_build(operation_name: str, job: signac.job.Job) check whether an operation is requested operation can be simple operations defined by a keyword like blockMesh or operations with parameters defined by a dictionary .. py:function:: execute_pre_build(operation_name: str, job: signac.job.Job) check whether an operation is requested operation can be simple operations defined by a keyword like blockMesh or operations with parameters defined by a dictionary .. py:function:: start_job_state(_, job: signac.job.Job) -> None .. py:function:: end_job_state(_, job: signac.job.Job) -> Literal[True] .. py:function:: dispatch_pre_hooks(operation_name: str, job: signac.job.Job) just forwards to start_job_state and execute_pre_build .. py:function:: dispatch_post_hooks(operation_name: str, job: signac.job.Job) Forwards to `execute_post_build`, performs md5sum calculation of case files and finishes with `end_job_state` .. py:function:: set_failure(operation_name: str, error, job: signac.job.Job) just forwards to start_job_state and execute_pre_build .. py:function:: copy_on_uses(args: dict, job: signac.job.Job, path: str, target: str) copies the file specified in args['uses'] to path/target .. py:function:: controlDict(job: signac.job.Job, args={}) sets up the controlDict .. py:function:: MultiCase(job: signac.job.Job, args={}) Dummy operation to generate multiple cases .. py:function:: blockMesh(job: signac.job.Job, args={}) .. py:function:: replaceMesh(job: signac.job.Job, args={}) .. py:function:: shell(job: signac.job.Job, args={}) .. py:function:: initialConditions(job: signac.job.Job, args={}) A special operation to allow copying from 0.orig folders .. py:function:: fvSolution(job: signac.job.Job, args={}) .. py:function:: fvSchemes(job: signac.job.Job, args={}) .. py:function:: transportProperties(job: signac.job.Job, args={}) .. py:function:: turbulenceProperties(job: signac.job.Job, args={}) .. py:function:: setKeyValuePair(job: signac.job.Job, args={}) .. py:function:: has_mesh(job: signac.job.Job) -> bool Check whether all mesh files are files (owning) or symlinks (non-owning) TODO check also for .obr files for state of operation .. py:function:: decomposePar(job: signac.job.Job, args={}) .. py:function:: fetchCase(job: signac.job.Job, args={}) .. py:function:: is_locked(job: signac.job.Job) -> bool Cases that are already started are set to tmp_lock dont try to execute them .. py:function:: refineMesh(job: signac.job.Job, args={}) .. py:function:: allClean(job: signac.job.Job, args={}) .. py:function:: checkMesh(job: signac.job.Job, args={}) .. py:function:: get_number_of_procs(job: signac.job.Job) -> int Deduces the number of processors For performance reasons the cache is used to store the number of subdomains .. py:function:: get_values(jobs: list, key: str) -> set find all different statepoint values .. py:function:: run_cmd_builder(job: signac.job.Job, cmd_format: str, args: dict) -> str Builds the cli command to run a OpenFOAM application .. py:function:: validate_state_impl(_: str, job: signac.job.Job) -> None Perform a detailed update of the job state .. py:function:: resetCase(job: signac.job.Job, args={}) -> None Dummy operation that calls resetCase .. py:function:: validateState(job: signac.job.Job, args={}) -> None Dummy operation which forwards to validate_state_impl. The reason for keeping this function is that it can be called from the cli to force a detailed update .. py:function:: runParallelSolver(job: signac.job.Job, args={}) -> str .. py:function:: runSerialSolver(job: signac.job.Job, args={}) .. py:function:: archive(job: signac.job.Job, args={}) -> Literal[True] .. py:function:: apply(*jobs, args={})