GitHub

romtools.workflows.workflow_utils 

 1from typing import Iterable
 2import os
 3import shutil
 4import subprocess
 5
 6
 7def create_empty_dir(dir_name: str):
 8    '''
 9    Create empty directory
10    '''
11    if os.path.exists(dir_name):
12        pass
13    else:
14        os.makedirs(dir_name)
15
16
17def setup_directory(source_dir: str,
18                    target_dir: str,
19                    files2link: Iterable = (),
20                    files2copy: Iterable = ()):
21    '''
22    Create new directory and populate with files
23    '''
24    create_empty_dir(target_dir)
25    for file in files2copy:
26        shutil.copy(f'{source_dir}/{file}',
27                    f'{target_dir}/{os.path.basename(file)}')
28    for file in files2link:
29        os.symlink(f'{source_dir}/{file}',
30                   f'{target_dir}/{os.path.basename(file)}')
31
32
33def run_model(module: str = None,
34              pre_script: str = None,
35              executable: str = 'bash',
36              num_procs: int = 1,
37              directory: str = '.',
38              **kwargs):
39    '''
40    Execute external model
41    '''
42    execution_str = f'module load {module};' if module is not None else ''
43    execution_str += pre_script if pre_script is not None else ''
44    execution_str += f'mpirun -n {num_procs} {executable}'
45    for flag, value in kwargs:
46        execution_str += f' {flag} {value}'
47    subprocess.run(execution_str, shell=True, check=True, cwd=directory)
def create_empty_dir(dir_name: str): 
 8def create_empty_dir(dir_name: str):
 9    '''
10    Create empty directory
11    '''
12    if os.path.exists(dir_name):
13        pass
14    else:
15        os.makedirs(dir_name)

Create empty directory

def setup_directory( source_dir: str, target_dir: str, files2link: Iterable = (), files2copy: Iterable = ()): 
18def setup_directory(source_dir: str,
19                    target_dir: str,
20                    files2link: Iterable = (),
21                    files2copy: Iterable = ()):
22    '''
23    Create new directory and populate with files
24    '''
25    create_empty_dir(target_dir)
26    for file in files2copy:
27        shutil.copy(f'{source_dir}/{file}',
28                    f'{target_dir}/{os.path.basename(file)}')
29    for file in files2link:
30        os.symlink(f'{source_dir}/{file}',
31                   f'{target_dir}/{os.path.basename(file)}')

Create new directory and populate with files

def run_model( module: str = None, pre_script: str = None, executable: str = 'bash', num_procs: int = 1, directory: str = '.', **kwargs): 
34def run_model(module: str = None,
35              pre_script: str = None,
36              executable: str = 'bash',
37              num_procs: int = 1,
38              directory: str = '.',
39              **kwargs):
40    '''
41    Execute external model
42    '''
43    execution_str = f'module load {module};' if module is not None else ''
44    execution_str += pre_script if pre_script is not None else ''
45    execution_str += f'mpirun -n {num_procs} {executable}'
46    for flag, value in kwargs:
47        execution_str += f' {flag} {value}'
48    subprocess.run(execution_str, shell=True, check=True, cwd=directory)

Execute external model