Cycles runner class

cycles.CyclesRunner(executable, rotation_builder=False) dataclass

Run one or many Cycles simulations with templated inputs.

Manages batch execution of Cycles simulations by generating control files, operation files, and nudge files from templates and parameter dictionaries. Consolidates results into a summary CSV file.

Attributes:
  • executable (str) –

    Absolute path to the Cycles executable binary.

  • rotation_builder (bool) –

    If True, disables spin-up option and enables rotation features.

run(simulations, control_dict, *, summary='summary.csv', operation_template=None, operation_dict=None, calibration_dict=None, options='', rm_input=False, rm_output=False, rm_steady_state_soil=True, silence=True, user_comment='')

Execute a batch of simulations and write a consolidated summary.

Parameters:
  • simulations (SimulationConfig) –

    Simulation rows as list of dicts or a DataFrame.

  • control_dict (dict[str, Any]) –

    Control-file values or callables evaluated per row.

  • summary (str, default: 'summary.csv' ) –

    Summary CSV name written under summary directory.

  • operation_template (Path | str | None, default: None ) –

    Template file for generated operation files.

  • operation_dict (dict[str, Any] | None, default: None ) –

    Substitutions used with operation template.

  • calibration_dict (dict[str, Any] | None, default: None ) –

    Nudge-file values or callables per row.

  • options (str, default: '' ) –

    Cycles command options.

  • rm_input (bool, default: False ) –

    Remove generated input files after each run.

  • rm_output (bool, default: False ) –

    Remove run output directory after each run.

  • rm_steady_state_soil (bool, default: True ) –

    Remove generated steady-state soil file.

  • silence (bool, default: True ) –

    If True, suppress simulation screen output.

  • user_comment (str, default: '' ) –

    Optional text prefixed to summary header comments.
Source code in cycles/cycles_runner.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def run(self, simulations: SimulationConfig, control_dict: dict[str, Any], *,
    summary: str='summary.csv', operation_template: Path | str | None=None, operation_dict: dict[str, Any] | None=None, calibration_dict: dict[str, Any] | None=None,
    options: str='', rm_input: bool=False, rm_output: bool=False, rm_steady_state_soil: bool=True, silence: bool=True, user_comment: str='') -> None:
    """Execute a batch of simulations and write a consolidated summary.

    Args:
        simulations: Simulation rows as list of dicts or a DataFrame.
        control_dict: Control-file values or callables evaluated per row.
        summary: Summary CSV name written under summary directory.
        operation_template: Template file for generated operation files.
        operation_dict: Substitutions used with operation template.
        calibration_dict: Nudge-file values or callables per row.
        options: Cycles command options.
        rm_input: Remove generated input files after each run.
        rm_output: Remove run output directory after each run.
        rm_steady_state_soil: Remove generated steady-state soil file.
        silence: If True, suppress simulation screen output.
        user_comment: Optional text prefixed to summary header comments.
    """
    if isinstance(simulations, pd.DataFrame):
        simulations = simulations.to_dict(orient='records')
    assert isinstance(simulations, list)

    if (operation_template is None) != (operation_dict is None):
        raise ValueError(
            "operation_template and operation_dict must be provided together or not at all. "
            f"Got operation_template={'None' if operation_template is None else repr(operation_template)}, "
            f"operation_dict={'None' if operation_dict is None else '...'}"
        )

    if 's' in options and self.rotation_builder:
        raise ValueError('Spin-up cannot be used with rotation builder.')

    operation_template = Path(operation_template) if operation_template is not None else None
    comment = user_comment + _generate_comment(self.executable, options)
    first_run = True

    SUMMARY_DIR.mkdir(exist_ok=True)

    for s in simulations:
        cxt: SimulationContext = self._resolve(s, control_dict, operation_dict, calibration_dict)
        print(f'{cxt.name} - ', end='')

        self._write_inputs(cxt, operation_template)

        cycles = Cycles(path='.', simulation=cxt.name, executable=self.executable)

        code, _ = cycles.run(options=options, silence=silence)

        if code == 0:
            self._write_summary(cycles, summary, header=first_run, comment=comment)
            first_run = False
            print('Success')
        elif code == 1:
            print('Fail')

        if rm_input:
            self._remove_inputs(cxt)
        if rm_output:
            shutil.rmtree(OUTPUT_DIR / cxt.name, ignore_errors=True)
        if rm_steady_state_soil:
            (INPUT_DIR / f'{cxt.name}_ss.soil').unlink(missing_ok=True)