Source code for

# -*- coding: utf-8 -*-

CML Parametric Run Manager

import os
import logging
import itertools
    from import Mapping
except ImportError:  # pragma: no cover
    from collections import Mapping

import numpy as np

from ..utils import osutils
from import CaelusDict
from .case import CMLSimCollection

_lgr = logging.getLogger(__name__)

[docs]def normalize_variable_param(varspec): """Helper function to normalize the different run matrix options""" if isinstance(varspec, Mapping): start = float(varspec["start"]) step = float(varspec.get("step", 1)) stop = float(varspec["stop"]) + 0.5*step return np.arange(start, stop, step) if isinstance(varspec, (list, np.ndarray)): return varspec return [varspec]
[docs]def iter_case_params(sim_options, case_name_func): """Normalize the keys and yield all possible run setups""" casefmt = sim_options.get("case_format", "case_{idx:04d}") const_params = sim_options.get("constant_parameters", CaelusDict()) run_matrix = sim_options["run_matrix"] code = None extract_vars = None if "apply_transforms" in sim_options: code = sim_options.apply_transforms.code extract_vars = sim_options.apply_transforms.get( "extract_vars", None) idx = 1 for i, group in enumerate(run_matrix): ropts = {k: normalize_variable_param(v) for k, v in group.items()} rkeys = ropts.keys() rvalues = ropts.values() for j, vals in enumerate(itertools.product(*rvalues)): rdict = CaelusDict(zip(rkeys, vals)) rdict.update(const_params) myglobs = dict(**rdict) myglobs['np'] = np mylocs = {} if code is not None: exec(code, myglobs, mylocs) if extract_vars is not None: for k in extract_vars: rdict[k] = mylocs[k] else: rdict.update(mylocs) case_params = CaelusDict(rdict) rdict['idx'] = idx # Global index rdict['gid'] = i # Group index rdict['cid'] = j # Case index (within this group) case_name = case_name_func( case_format=casefmt, case_params=rdict) yield (case_name, case_params) idx += 1
[docs]class CMLParametricRun(CMLSimCollection): """A class to handle parametric runs""" _json_public_ = ("name sim_dict case_names _udf_script".split()) def __init__(self, name, sim_dict, env=None, basedir=None): """ Args: name (str): Unique name for this parametric run sim_dict (CaelusDict): Dictionary with simulation settings env (CMLEnv): CML execution environment basedir (path): Path where the parametric run directories are created """ super(CMLParametricRun, self).__init__(name, env, basedir) #: Dictionary containing the run settings self.sim_dict = sim_dict self.udf = self.udf_instance(self.udf_script, self.udf_params) self.udf.sim_init_udf(simcoll=self, is_reload=False) @property def udf_script(self): """Return the UDF script""" if not hasattr(self, "_udf_script"): self._udf_script = self.sim_dict.pop("udf_script", None) if self._udf_script is not None: self._udf_script = osutils.abspath(self._udf_script) return self._udf_script @property def udf_params(self): """Return the parameters for UDF script""" return self.sim_dict.get("udf_params", None)
[docs] def setup(self): """Setup the parametric case directories""" simcfg = self.sim_dict tmpl_info = CaelusDict(simcfg.template) tmpl_dir = tmpl_info.pop("path") setup_params = simcfg.simulation_setup runconf = simcfg.run_configuration if not osutils.path_exists(tmpl_dir): raise FileNotFoundError( "Cannot find case template directory: %s"%tmpl_dir) cases = [] if "run_matrix" in setup_params: osutils.ensure_directory(self.casedir) cases = [self.setup_case(cname, tmpl_dir, cparams, runconf, tmpl_info) for cname, cparams in iter_case_params(setup_params, self.udf.sim_case_name)] self.cases = [case for case in cases if case is not None] self.case_names = [ for case in self.cases] fname = os.path.join(self.casedir, "caelus_sim.yaml") with open(fname, 'w') as fh: simfile = CaelusDict(simulation=simcfg) simfile.to_yaml(fh)
[docs] def setup_case(self, cname, tmpl_dir, cparams, runconf, clone_opts): """Helper function to setup the cases""" cdir = os.path.join(self.casedir, cname) osutils.ensure_directory(os.path.dirname(cdir)) skip_setup = self.udf.case_setup_prologue( name=cname, case_params=cparams, run_config=runconf) if skip_setup: return None case = self.simulation_class()( cname, cml_env=self.env, basedir=self.casedir, parent=self) case.clone(tmpl_dir, **clone_opts) case.run_config = runconf case.udf = self.udf with osutils.set_work_dir(cdir): case.update() cmlctrls = case.cmlControls cmlctrls.write() self.udf.case_setup_epilogue(case) return case