Runners for simulation software#

Support modules for running simtools, CORSIKA, and sim_telarray.

Science workflows#

simtools_runner#

Tools for running applications in the simtools framework.

runners.simtools_runner.apply_execution_overrides(configurations, args, model_source, env_file)[source]#: Apply inherited settings before recording or running applications.

runners.simtools_runner.get_explicit_env_file(args_dict)[source]#: Return the runner env file when it was explicitly supplied on the CLI.

runners.simtools_runner.model_source_options(args_dict)[source]#: Return configured model-source options for child applications.

runners.simtools_runner.prepare_runtime_environment(runtime_environment_file)[source]#

Read and prepare a standalone runtime environment.

Parameters:

runtime_environment_filestr or pathlib.Path: YAML file with a top-level runtime_environment mapping.

Returns:

tuple[dict, list[str]]: Runtime-environment configuration and prepared runtime command.

Raises:

ValueError: If the YAML content is not a mapping or lacks runtime_environment.
jsonschema.ValidationError: If the configuration does not conform to the runtime-environment schema.

runners.simtools_runner.prepare_workflow(args_dict, replacements=None)[source]#: Resolve a workflow and its backend configuration without executing it.

runners.simtools_runner.read_runtime_environment(runtime_environment)[source]#: Read the runtime environment (e.g. docker runtime) and return the required command.

runners.simtools_runner.resolve_nested_configuration_files(configurations, replacements)[source]#: Resolve placeholders in referenced backend configuration files.

runners.simtools_runner.run_applications(args_dict, run_time=None, replacements=None, context=None)[source]#

Run simtools applications step-by-step as defined in a configuration file.

Parameters:

args_dictdict: Command line arguments. Optional keys: overwrite_collection_files (allow overwriting collection files with the same basename) and provenance_path (False omits the resolved workflow file).
run_timelist or None: Prepared runtime command, reused instead of preparing the runtime environment.
replacementsdict or None: Placeholder replacements applied recursively; __CONFIG_DIRECTORY__ is added automatically.
contextRunContext, optional: Run context used to report software versions and collect workflow metadata.

stages#

Simulation stages (e.g. CORSIKA, sim_telarray), the data products connecting them, and simulation chains.

Simulation stages and the chains built from them.

A stage is one simulation program (e.g. CORSIKA or sim_telarray). Stages are connected by the data products they consume and produce, either through files or as a stream. A chain is an ordered list of stage names. Stages are looked up by name in the STAGES registry, which maps each name to a factory returning the stage class; the factories defer imports until the stage is used. Additional stages are added with register_stage(). SimulationChain checks that each stage consumes the data product of the stage before it when the chain is created.

class runners.stages.SimulationChain(stages)[source]#

Ordered simulation stages, checked for matching data products.

classmethod from_software(simulation_software)[source]#: Return the chain of a simulation_software value (e.g. corsika_sim_telarray).

has(name)[source]#: Return whether the chain contains a stage.

property reads_input_file#: Return whether the first stage reads an input file.

property software#: Return the simulation_software value of this chain.

property stage_classes#: Return the stage classes in chain order.

property starts_with_showers#: Return whether the first stage generates showers (consumes no input).

class runners.stages.SimulationStage(*args, **kwargs)[source]#

One program of a simulation chain.

environment()[source]#: Return environment variables needed to run the stage.

get_resources(runtime=None)[source]#: Return the computing resources used.

classmethod instances(inputs, label=None)[source]#: Return the stage instances for a run (e.g. one per array model).

output_file(run_number)[source]#: Return the file the stage writes when its output is not streamed.

prepare(run_number, input_file=None, output=None)[source]#

Write the stage inputs and return the shell lines that run the stage.

Parameters:

run_numberint: Run number.
input_filestr or Path or None: Input file, STDIN when reading a stream, or None for the first stage.
outputStreamTo or None: Downstream commands when the output is streamed; None writes the stage’s file.

Returns:

list[str]: Shell lines.

class runners.stages.StageInputs(array_models, simulation_parameters, shower_configurations, args, context=None)[source]#: What a simulation provides to create its stage instances.

class runners.stages.StreamTo(commands)[source]#: Output of a stage that is streamed to downstream stage commands (shell lines).

runners.stages.get_stage_class(name)[source]#: Return the stage class registered under a name.

runners.stages.register_stage(name, factory)[source]#

Register a stage under a name, replacing any stage already registered under it.

Parameters:

namestr: Stage name used in simulation chains.
factorycallable: Function without arguments returning the stage class.

chain_runner#

Run script for a chain of simulation stages.

class runners.chain_runner.ChainRunner(chain, inputs, label=None)[source]#

Write the run script of a simulation chain.

Stages that both stream are connected by a stream (the upstream stage receives the downstream commands); otherwise the upstream output file is the downstream input. A stage may run as several instances (e.g. one per array model).

Parameters:

chainSimulationChain: Ordered simulation stages.
inputsStageInputs: Array models, simulation parameters, shower configurations and arguments.
labelstr: Instance label.

environment()[source]#: Return the environment variables of all stages.

get_resources(runtime=None)[source]#: Return computing resources used (reported by the first stage).

prepare_run(run_number, sub_script, input_file=None, extra_commands=None)[source]#

Write the run script for one run.

Parameters:

run_numberint: Run number.
sub_scriptstr or Path: Run script to write.
input_filestr or Path, optional: Input file of the first stage (when it does not generate showers).
extra_commandsstr or list[str], optional: Commands written before the simulation commands.

corsika_stage#

CORSIKA as a simulation stage (air showers and Cherenkov photons).

class runners.corsika_stage.CorsikaStage(corsika_config, label=None, sequential=False, save_output=False, context=None)[source]#

CORSIKA air-shower simulation stage.

Writes the CORSIKA input card and the commands that run CORSIKA. A streamed output is passed to the downstream commands with multipipe_corsika.

Parameters:

corsika_configCorsikaConfig: CORSIKA configuration.
labelstr: Instance label.
sequentialbool: Run multipipe_corsika in sequential mode (no parallel CORSIKA/downstream execution).
save_outputbool: Also save the streamed CORSIKA output to a compressed file.

environment()[source]#: Return the CORSIKA interaction-model environment variables.

get_resources(runtime=None)[source]#: Return computing resources used.

classmethod instances(inputs, label=None)[source]#: Return the CORSIKA stage of a run (the first shower configuration defines it).

output_file(run_number)[source]#: Return the CORSIKA output file.

prepare(run_number, input_file=None, output=None)[source]#

Write the CORSIKA input card (and stream configuration) and return the run lines.

Parameters:

run_numberint: Run number.
input_fileNone: CORSIKA reads no input product.
outputStreamTo or None: Downstream commands receiving the CORSIKA output stream; None writes the CORSIKA output file.

Returns:

list[str]: Shell lines running CORSIKA.

sim_telarray_stage#

sim_telarray as a simulation stage (telescope and camera response).

class runners.sim_telarray_stage.SimTelArrayStage(simulation_parameters, array_model, label=None, context=None)[source]#

sim_telarray array simulation stage, one instance per array model.

Parameters:

simulation_parametersSimulationParameters: Physical run settings.
array_modelArrayModel: Telescope and site models.
labelstr: Instance label.

environment()[source]#: Return the sim_telarray environment variables.

property file_list#: Files of this stage by file type.

get_resources(runtime=None)[source]#: Return computing resources used.

classmethod instances(inputs, label=None)[source]#: Return one sim_telarray stage per array model.

output_file(run_number)[source]#: Return the sim_telarray output file.

prepare(run_number, input_file=None, output=None)[source]#

Write the sim_telarray configuration and return the run lines.

Parameters:

run_numberint: Run number.
input_filestr or Path: CORSIKA output file, or STDIN when reading a stream.
outputNone: sim_telarray writes its output file.

Returns:

list[str]: Shell lines running sim_telarray (one per run of a run set; one line when reading a stream).

property runner_service#: File naming of this stage.

simtel_runner#

Base class for running sim_telarray simulations.

class runners.simtel_runner.SimtelRunner(label=None, config=None, array_model=None, context=None)[source]#

Base class for running simulations based on the sim_telarray software stack.

The sim_telarray software stack includes sim_telarray itself and auxiliary tools, LightEmission, and other software packages.

Parameters:

label: str: Instance label. Important for output file naming.
config: CorsikaConfig or dict: Configuration parameters.

get_resources(runtime=None)[source]#: Return computing resources used.

make_run_command(run_number=None, input_file=None)[source]#: Make the sim_telarray run command (to implemented in subclasses).

run(test=False, input_file=None, run_number=None)[source]#

Make run command and run sim_telarray.

Parameters:

test: bool: If True, make simulations faster.
input_file: str or Path: Full path of the input CORSIKA file.
run_number: int: Run number.

runners.simtel_runner.sim_telarray_env_as_string()[source]#: Return the sim_telarray environment variables as a string.

runner_services#

Base service methods for simulation runners.

class runners.runner_services.RunnerServices(config, run_type, label=None, array_model=None, context=None)[source]#

Base service methods for simulation runners.

Includes file naming and directory management.

Parameters:

config: CorsikaConfig, dict: Configuration parameters.
run_typestr: Type of simulation runner.
labelstr: Label.

get_file_name(file_type, run_number=None)[source]#

Get a file name depending on file type and run number.

Parameters:

file_typestr: The type of file (determines the file suffix).
run_numberint: Run number.

Returns:

str: File name with full path.

Raises:

ValueError: If file_type is unknown.

get_resources(runtime=None)[source]#

Return resources used by a simulation run.

Parameters:

runtime: float, optional: Wall-clock runtime in seconds measured by the Python subprocess caller.

Returns:

dict: run time and number of simulated events

load_data_directory()[source]#

Create and return directory for the given run type.

Returns:

Path: Path to the created directory.

load_files(run_number=None)[source]#

Load files required for the simulation run.

Parameters:

run_number: int: Run number.

Returns:

dict: Dictionary containing paths to files required for the simulation run.

runners.runner_services.validate_corsika_run_number(run_number)[source]#

Validate run number and return it.

Parameters:

run_number: int: Run number.

Returns:

int: Run number.

Raises:

ValueError: If run_number is not a valid value (< 1 or > 999999).

runners.runner_services.warn_if_file_name_exceeds_dirac_limit(file_name, file_type)[source]#

Warn when a generated file name exceeds the current DIRAC limit.

Parameters:

file_namestr: Final file name without its directory path.
file_typestr: Internal simtools file type used to generate the file name.