"""Base service methods for simulation runners."""
import logging
from collections.abc import Mapping
from simtools.context import RunContext
from simtools.io.file_type import product_value
from simtools.simulation.parameters import SimulationParameters
_logger = logging.getLogger(__name__)
_DIRAC_MAX_FILENAME_LENGTH = 128
_RUN_NUMBER = "run_number"
_SUB_DIRECTORY_TYPE = "sub_dir_type"
_FILE_SUFFIX = "suffix"
_SUBMISSION_DIRECTORY = "sub"
_SIM_TELARRAY_HISTOGRAM_PRODUCT = "sim_telarray_histogram_output"
_RUNNER_FILE_PLACEMENT = {
"corsika_input": _RUN_NUMBER,
"corsika_output": _RUN_NUMBER,
"corsika_log": _RUN_NUMBER,
"corsika_err": _RUN_NUMBER,
"corsika_resources": _RUN_NUMBER,
"iact_output": _RUN_NUMBER,
"sim_telarray_output": _RUN_NUMBER,
"sim_telarray_histogram": _RUN_NUMBER,
"sim_telarray_log": _RUN_NUMBER,
"sim_telarray_err": _RUN_NUMBER,
"sim_telarray_event_data": _RUN_NUMBER,
"sim_telarray_resources": _RUN_NUMBER,
"light_emission_log": _RUN_NUMBER,
"light_emission_err": _RUN_NUMBER,
"light_emission_resources": _RUN_NUMBER,
"multi_pipe_config": _RUN_NUMBER,
"multi_pipe_script": _RUN_NUMBER,
"multi_pipe_resources": _RUN_NUMBER,
"sub_out": _SUBMISSION_DIRECTORY,
"sub_err": _SUBMISSION_DIRECTORY,
"sub_log": _SUBMISSION_DIRECTORY,
"sub_script": _SUBMISSION_DIRECTORY,
}
_RUNNER_PRODUCT_IDS = {"sim_telarray_histogram": _SIM_TELARRAY_HISTOGRAM_PRODUCT}
FILES_AND_PATHS = {
file_type: {
_FILE_SUFFIX: product_value(
_RUNNER_PRODUCT_IDS.get(file_type, file_type), "default_suffix"
),
_SUB_DIRECTORY_TYPE: directory_type,
}
for file_type, directory_type in _RUNNER_FILE_PLACEMENT.items()
}
[docs]
def warn_if_file_name_exceeds_dirac_limit(file_name, file_type):
"""Warn when a generated file name exceeds the current DIRAC limit.
Parameters
----------
file_name : str
Final file name without its directory path.
file_type : str
Internal simtools file type used to generate the file name.
"""
if len(file_name) <= _DIRAC_MAX_FILENAME_LENGTH:
return
_logger.warning(
"Generated %s file name exceeds DIRAC filename limit (%s > %s): %s",
file_type,
len(file_name),
_DIRAC_MAX_FILENAME_LENGTH,
file_name,
)
[docs]
def validate_corsika_run_number(run_number):
"""
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).
"""
if not float(run_number).is_integer() or run_number < 1 or run_number > 999999:
raise ValueError(
f"Invalid type of run number ({run_number}) - it must be an uint < 1000000."
)
return run_number
[docs]
class RunnerServices:
"""
Base service methods for simulation runners.
Includes file naming and directory management.
Parameters
----------
config: CorsikaConfig, dict
Configuration parameters.
run_type : str
Type of simulation runner.
label : str
Label.
"""
def __init__(self, config, run_type, label=None, array_model=None, context=None):
"""Initialize RunnerServices."""
self._logger = logging.getLogger(__name__)
self.label = label
self.array_model = array_model or getattr(config, "array_model", None)
self.config = getattr(config, "simulation_parameters", config)
self.run_type = run_type
self.context = context or RunContext.from_environment()
self.directory = self.load_data_directory()
[docs]
def load_data_directory(self):
"""
Create and return directory for the given run type.
Returns
-------
Path
Path to the created directory.
"""
directory = self.context.paths.get_output_directory(self.run_type)
self._logger.debug(f"Data directories for {self.run_type}: {directory}")
return directory
[docs]
def load_files(self, run_number=None):
"""
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.
"""
run_files = {}
for key in FILES_AND_PATHS:
if key.startswith(self.run_type.lower()):
run_files[key] = self.get_file_name(file_type=key, run_number=run_number)
for key, file_path in run_files.items():
self._logger.debug(f"{key}: {file_path}")
return run_files
def _get_file_basename(self, run_number, is_multi_pipe=False):
"""
Get the base name for the simulation files.
Parameters
----------
run_number: int
Run number.
file_type: str
File type.
Returns
-------
str
Base name for the simulation files.
"""
if isinstance(self.config, SimulationParameters):
return self._get_file_base_name_from_simulation_parameters(run_number, is_multi_pipe)
if isinstance(self.config, Mapping):
return self._get_file_base_name_from_core_config()
raise ValueError(f"Invalid configuration type: {type(self.config)}")
def _get_file_base_name_from_core_config(self):
"""Get file base name from core configuration."""
cfg = self.config
run_mode = cfg.get("run_mode")
zenith_angle = cfg.get("zenith_angle")
azimuth_angle = cfg.get("azimuth_angle")
light_source_suffix = (
f"{cfg['light_source']}_"
if run_mode == "illuminator" and cfg.get("light_source")
else ""
)
parts = [
f"{run_mode}_" if run_mode else "",
f"{self._get_run_number_string(cfg.get(_RUN_NUMBER))}_" if _RUN_NUMBER in cfg else "",
f"za{round(zenith_angle):02}deg_" if isinstance(zenith_angle, (int, float)) else "",
f"azm{round(azimuth_angle):03}deg_" if isinstance(azimuth_angle, (int, float)) else "",
f"{cfg['site']}_" if cfg.get("site") else "",
light_source_suffix,
f"{cfg['layout']}_" if cfg.get("layout") else "",
cfg.get("model_version", ""),
f"_{self.label}" if self.label else "",
]
return "".join(parts)
def _get_file_base_name_from_simulation_parameters(self, run_number, is_multi_pipe=False):
"""Get file base name from physical simulation settings."""
zenith = self.config.zenith_min
vc_high = self.config.viewcone_max
if self.config.run_mode is not None and self.config.run_mode != "":
primary_name = self.config.run_mode
else:
primary_name = (
self.config.primary_particle.name if self.config.primary_particle else None
)
if primary_name == "gamma" and vc_high > 0:
primary_name = "gamma_diffuse"
file_label = f"_{self.label}" if self.label else ""
run_number_string = self._get_run_number_string(run_number)
prefix = f"{primary_name}_{run_number_string}_" if primary_name else f"{run_number_string}_"
return (
prefix
+ f"za{round(zenith):02}deg_"
+ f"azm{round(self.config.azimuth_angle):03}deg_"
+ f"{self.array_model.site}_"
+ (f"{self.array_model.layout_name}_" if self.array_model.layout_name else "")
+ (self.array_model.model_version if not is_multi_pipe else "")
+ file_label
)
def _get_sub_directory(self, run_number, dir_path):
"""
Return sub directory with / without run number.
Parameters
----------
run_number: int
Run number.
dir_path: Path
Parent directory path.
Returns
-------
Path
Child directory path.
"""
sub_dir = dir_path / self._get_run_number_string(run_number)
sub_dir.mkdir(parents=True, exist_ok=True)
return sub_dir
[docs]
def get_file_name(self, file_type, run_number=None):
"""
Get a file name depending on file type and run number.
Parameters
----------
file_type : str
The type of file (determines the file suffix).
run_number : int
Run number.
Returns
-------
str
File name with full path.
Raises
------
ValueError
If file_type is unknown.
"""
file_name = self._get_file_basename(run_number, file_type.startswith("multi_pipe"))
try:
desc = FILES_AND_PATHS[file_type]
except KeyError as exc:
raise ValueError(f"Unknown file type: {file_type}") from exc
if desc[_SUB_DIRECTORY_TYPE] == _RUN_NUMBER:
dir_path = self._get_sub_directory(run_number, self.directory)
else:
dir_path = self.directory
output_file_name = f"{file_name}{desc[_FILE_SUFFIX]}"
warn_if_file_name_exceeds_dirac_limit(output_file_name, file_type)
return dir_path / output_file_name
@staticmethod
def _get_run_number_string(run_number):
"""
Get run number string as used for the simulation file names (ex. run000014).
Parameters
----------
run_number: int
Run number.
Returns
-------
str
Run number string.
"""
if run_number is None:
return ""
return f"run{validate_corsika_run_number(run_number):06d}"
[docs]
def get_resources(self, runtime=None):
"""
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
"""
if isinstance(self.config, SimulationParameters):
return {
"runtime": runtime,
"n_events": int(self.config.shower_events),
}
self._logger.warning("Number of events cannot be determined from non-CORSIKA config.")
return {"runtime": runtime, "n_events": None}
