Source code for simtel.simulator_light_emission

"""Light emission simulation (e.g. illuminators or flashers)."""

import logging
import shlex
from pathlib import Path

import astropy.units as u
import numpy as np

from simtools.atmosphere.transmission import AtmosphericTransmission
from simtools.context import RunContext
from simtools.job_execution import execution
from simtools.job_execution.process_accounting import build_accounting_command
from simtools.model.calibration_model import CalibrationModel
from simtools.model.model_utils import (
    initialize_simulation_models,
    read_overwrite_model_parameter_dict,
)
from simtools.model_repository.asset_names import get_simtel_table_file_name
from simtools.ray_tracing.backends.interface import LightSourceRequest, get_backend
from simtools.runners import runner_services
from simtools.runners.simtel_runner import SimtelRunner, sim_telarray_env_as_string
from simtools.sim_events import validate_run_outputs
from simtools.simtel.photon_bunch_file import (
    bypass_optics_level,
    camera_bunches,
    write_photon_bunches,
)
from simtools.simtel.simtel_seeds import instance_seed_pairs
from simtools.simulation.configuration import get_light_source_writer, get_model_writer
from simtools.simulation.light_source_profiles import angular_profile, pulse_profile
from simtools.utils import value_conversion
from simtools.utils.geometry import (
    fiducial_radius_from_shape,
    ground_direction_to_telescope_frame,
    ground_to_telescope_frame,
)

AXES_OFFSETS = "axes_offsets"
CORSIKA_OBSERVATION_LEVEL = "corsika_observation_level"
FLASHER_PHOTONS = "flasher_photons"
FLASHER_POSITION = "flasher_position"
FLASHER_WAVELENGTH = "flasher_wavelength"
FLAT_FIELDING = "flat_fielding"
LIGHT_SOURCE_TYPE = "light_source_type"
LIGHT_SOURCE_POSITION = "light_source_position"
LIGHT_SOURCE_POINTING = "light_source_pointing"
NUMBER_OF_EVENTS = "number_of_events"
RUN_NUMBER = "run_number"
TELESCOPE_SPHERE_RADIUS = "telescope_sphere_radius"


[docs] class SimulatorLightEmission(SimtelRunner): """ Light emission simulation (e.g. illuminators or flashers). Uses the sim_telarray LightEmission package to simulate the light emission. Parameters ---------- light_emission_config : dict, optional Configuration for the light emission (e.g. number of events, model names) telescope : str, optional Telescope name. label : str, optional Label for the simulation """
[docs] @staticmethod def get_available_wavelengths_from_config(config, model_reader=None, context=None): """ Get available wavelengths from model configuration without full initialization. This is a lightweight method that only initializes the calibration model to retrieve wavelengths, without telescope/site models or file I/O. Parameters ---------- config : dict Configuration dictionary with site, light_source, and model_version Returns ------- list of astropy.units.Quantity List of available wavelengths with units Raises ------ ValueError If required configuration keys are missing """ site = config.get("site") model_version = config.get("model_version") light_source = config.get("light_source") if site is None or model_version is None or light_source is None: raise ValueError( "Missing required configuration keys for wavelength query: " f"site={site}, model_version={model_version}, light_source={light_source}" ) calibration_model = CalibrationModel( site=site, calibration_device_model_name=light_source, model_version=model_version, label=f"temp_wavelength_query_{light_source}", overwrite_model_parameter_dict=read_overwrite_model_parameter_dict(context=context), model_reader=model_reader, context=context, ) wavelengths = calibration_model.get_parameter_value_with_unit(FLASHER_WAVELENGTH) return value_conversion.ensure_list(wavelengths)
def __init__( self, light_emission_config, telescope=None, label=None, model_reader=None, context=None ): """Initialize SimulatorLightEmission.""" light_emission_config = dict(light_emission_config) self._logger = logging.getLogger(__name__) self.configuration_writers = {} self._seed_pair = None self.context = context or RunContext.from_environment() self.model_reader = model_reader or self.context.model_reader self.io_handler = self.context.paths telescope = telescope or light_emission_config.get("telescope") label = f"{label}_{telescope}" if label else telescope wavelength = light_emission_config.get("wavelength") if wavelength is not None: # always include in the label for consistency label = f"{label}_{wavelength.to(u.nm).value:.0f}nm" super().__init__(label=label, config=light_emission_config, context=self.context) self.submission_files = runner_services.RunnerServices( light_emission_config, run_type="sub", label=label, context=self.context ) model_kwargs = { "label": label, "site": light_emission_config.get("site"), "telescope_name": telescope, "calibration_device_name": light_emission_config.get("light_source"), "calibration_device_type": light_emission_config.get(LIGHT_SOURCE_TYPE), "model_version": light_emission_config.get("model_version"), "model_reader": self.model_reader, "context": self.context, } if light_emission_config.get("model_directory") is not None: model_kwargs["model_directory"] = light_emission_config["model_directory"] self.telescope_model, self.site_model, self.calibration_model = ( initialize_simulation_models(**model_kwargs) ) get_model_writer(self.telescope_model).write_sim_telarray_config_file( additional_models=self.site_model ) self.light_emission_config = self._initialize_light_emission_configuration( light_emission_config ) def _initialize_light_emission_configuration(self, config): """Initialize light emission configuration from the calibration model.""" flasher_type = self.calibration_model.get_parameter_value("flasher_type") if flasher_type: config[LIGHT_SOURCE_TYPE] = flasher_type.lower() if config.get("run_mode") is None: config["run_mode"] = config[LIGHT_SOURCE_TYPE] if config.get(FLASHER_PHOTONS) is None: config[FLASHER_PHOTONS] = self.calibration_model.get_parameter_value(FLASHER_PHOTONS) else: photons = value_conversion.parse_typed_sequence(config[FLASHER_PHOTONS], int) if len(photons) == 1: self.calibration_model.overwrite_model_parameter(FLASHER_PHOTONS, photons[0]) photons = photons[0] config[FLASHER_PHOTONS] = photons if config.get("wavelength") is not None: requested = config["wavelength"].to(u.nm) allowed = [ wl.to(u.nm) for wl in value_conversion.ensure_list( self.calibration_model.get_parameter_value_with_unit(FLASHER_WAVELENGTH) ) ] # Small tolerance for floating point comparison if not any(abs(requested - wl) < 0.5 * u.nm for wl in allowed): raise ValueError( f"Wavelength {requested.value} nm is not supported by the " f"illuminator. Allowed wavelengths are: {[wl.value for wl in allowed]} nm" ) closest_wavelength = min(allowed, key=lambda wl: abs(wl - requested)) self.calibration_model.overwrite_model_parameter(FLASHER_WAVELENGTH, closest_wavelength) config["wavelength"] = closest_wavelength self._logger.info(f"Using wavelength: {closest_wavelength.value} nm") if config.get(LIGHT_SOURCE_POSITION) is not None: config[LIGHT_SOURCE_POSITION] = ( np.array(config[LIGHT_SOURCE_POSITION], dtype=float) * u.m ) return config @staticmethod def _repeat_or_map_events(events, n_photon_levels): """Apply ff-1m event mapping rules to event counts.""" if n_photon_levels == 1: return [events[0]] if len(events) == n_photon_levels: return events if len(events) == 1: return [events[0]] * n_photon_levels raise ValueError( "Invalid number_of_events list length. Use one value or one value per photon intensity." )
[docs] def build_flasher_event_and_photon_sequences(self): """Build ff-1m-compatible events/photons sequences.""" photons_int = value_conversion.parse_typed_sequence( self.light_emission_config.get(FLASHER_PHOTONS), int ) events = value_conversion.parse_typed_sequence( self.light_emission_config.get(NUMBER_OF_EVENTS, 1), int ) events_int = self._repeat_or_map_events(events, len(photons_int)) return events_int, photons_int
[docs] def simulate(self): """Simulate light emission.""" run_script = self.prepare_run() execution.run_command( run_script, stdout=self.submission_files.get_file_name("sub_out"), stderr=self.submission_files.get_file_name("sub_err"), )
[docs] def prepare_run(self): """Write the bash run script with the light-emission command and return its path.""" script_file = self.submission_files.get_file_name(file_type="sub_script") output_file = self.runner_service.get_file_name(file_type="sim_telarray_output") if output_file.exists(): raise FileExistsError( f"sim_telarray output file exists, cancelling simulation: {output_file}" ) if self._backend() is not None: self._write_camera_photons(self.runner_service.get_file_name(file_type="iact_output")) lines = self.make_run_command() script_file.write_text("".join(lines), encoding="utf-8") return script_file
def _backend(self): """Return the ray-tracing backend class if it traces light, else None.""" backend = get_backend(context=self.context) return backend if backend.traces_light_sources else None def _seeds(self): """Return the (instrument seed, simulation seed) of this run.""" if self._seed_pair is None: self._seed_pair = instance_seed_pairs(context=self.context)[0] return self._seed_pair def _dual_mirror(self): """Return True for dual-mirror telescopes.""" return self.telescope_model.get_parameter_value("mirror_class") == 2 def _bypass_optics(self): """Return sim_telarray's Bypass_Optics value: photons arrive at the camera entrance.""" if self._backend() is not None: return bypass_optics_level(self._dual_mirror()) return 1 if self.light_emission_config[LIGHT_SOURCE_TYPE] == FLAT_FIELDING else 0 def _write_camera_photons(self, iact_output): """Write the photons the ray-tracing backend brings to the camera entrance.""" level_m = ( self.site_model.get_parameter_value_with_unit(CORSIKA_OBSERVATION_LEVEL).to(u.m).value ) cal = self.calibration_model shape, width_ns, decay_ns = cal.get_parameter_value("flasher_pulse_shape") source = { "angular_profile": angular_profile( cal.get_parameter_value("flasher_angular_distribution"), cal.get_parameter_value_with_unit("flasher_angular_distribution_width") .to(u.deg) .value, ), "pulse": pulse_profile(shape, width_ns, decay_ns), "wavelength_nm": cal.get_parameter_value_with_unit(FLASHER_WAVELENGTH).to(u.nm).value, "seeds": self._seeds(), } if self.light_emission_config[LIGHT_SOURCE_TYPE] == FLAT_FIELDING: request, radius, transmission, emission_altitude_m = self._flasher_request( source, level_m ) else: request, radius, transmission, emission_altitude_m = self._illuminator_request( source, level_m ) backend = self._backend()( self.telescope_model, self.site_model, self.label, request.telescope_zenith, False ) dual_mirror = self._dual_mirror() camera_rotation = ( self.telescope_model.get_parameter_value_with_unit("camera_rotate").to(u.deg).value if dual_mirror else 0.0 ) events = ( np.concatenate( [ camera_bunches( block | {"weight": block["weight"] * transmission}, dual_mirror=dual_mirror, emission_altitude_m=emission_altitude_m, camera_rotation_deg=camera_rotation, ) for block in pulse ] ) for pulse in backend.camera_photons(request) ) write_photon_bunches( iact_output, events, observation_level_m=level_m, telescope_radius_m=radius, atmospheric_profile=self.site_model.get_parameter_table("atmospheric_profile"), run_number=self.light_emission_config.get(RUN_NUMBER) or 1, ) def _photons_per_pulse(self): """Return the photons emitted in each light pulse (one entry per event).""" events, photons = self.build_flasher_event_and_photon_sequences() return tuple(n for count, n in zip(events, photons) for _ in range(count)) def _flasher_request(self, source, level_m): """Flasher light to the camera (as ff-1m); return (request, radius, 1, altitude).""" x, y, z = ( c.to(u.m).value for c in self.calibration_model.get_parameter_value_with_unit(FLASHER_POSITION)[:3] ) # Primary-mirror frame to telescope frame. z -= self.telescope_model.get_parameter_value_with_unit("mirror_offset").to(u.m).value radius = fiducial_radius_from_shape( self.telescope_model.get_parameter_value_with_unit("camera_body_diameter") .to(u.m) .value, self.telescope_model.get_parameter_value("camera_body_shape"), ) request = LightSourceRequest( position_m=(x, y, z), axis=(0.0, 0.0, 1.0), photons_per_pulse=self._photons_per_pulse(), telescope_zenith=0.0, inside_telescope=True, camera_radius_m=radius, **source, ) distance = self.calculate_distance_focal_plane_calibration_device().to(u.m).value return request, radius, 1.0, level_m + distance def _illuminator_request(self, source, level_m): """Illuminator light via the atmosphere; return (request, radius, transmission, level).""" theta, phi = self._get_telescope_pointing() position = self.get_illuminator_position() telescope = self._get_telescope_position_ground_with_axis_offset(*position) radius = self.telescope_model.get_parameter_value_with_unit(TELESCOPE_SPHERE_RADIUS) telescope, radius = [c.to(u.m).value for c in telescope], radius.to(u.m).value position = np.array([c.to(u.m).value for c in position]) relative = position - np.array(telescope) if relative[2] == 0: raise ValueError("Horizontal illuminator paths require a nonzero altitude difference") pointing = np.asarray(self.get_illuminator_pointing_vector(position * u.m), dtype=float) transmission = AtmosphericTransmission( self.site_model.get_parameter_table("atmospheric_transmission") ).transmission( source["wavelength_nm"], level_m + position[2], level_m + telescope[2], np.linalg.norm(relative) / abs(relative[2]), ) request = LightSourceRequest( position_m=tuple(ground_to_telescope_frame(relative, theta, phi)), axis=tuple( ground_direction_to_telescope_frame(pointing / np.linalg.norm(pointing), theta, phi) ), photons_per_pulse=self._photons_per_pulse(), telescope_zenith=theta, **source, ) return request, radius, float(transmission), level_m
[docs] def make_run_command(self, run_number=None, input_file=None): # pylint: disable=unused-argument """Light emission and sim_telarray run command (a backend writes the photons beforehand).""" iact_output = self.runner_service.get_file_name(file_type="iact_output") light_emission = ( "" if self._backend() is not None else f"{get_light_source_writer(self).make_command(iact_output)}\n\n" ) return [ "#!/usr/bin/env bash\n", "set -e -o pipefail\n", light_emission, ( f"[ -s '{iact_output}' ] || " f"{{ echo 'LightEmission did not produce IACT file' >&2; exit 1; }}\n\n" ), f"{self._make_simtel_script()}\n\n", f"rm -f '{iact_output}'\n\n", ]
def _get_telescope_pointing(self): """Return (theta, phi) telescope pointing for the light source type and position.""" if self.light_emission_config[LIGHT_SOURCE_TYPE] == FLAT_FIELDING: return 0.0, 0.0 if self.light_emission_config.get(LIGHT_SOURCE_POSITION) is not None: self._logger.info("Using fixed (vertical up) telescope pointing.") return 0.0, 0.0 illuminator_position = self.get_illuminator_position() _, angles = self._calibration_pointing_direction(*illuminator_position) return angles[0], angles[1] def _calibration_pointing_direction(self, x_cal=None, y_cal=None, z_cal=None): """Return [pointing vector, telescope and beam angles] for an off-telescope device.""" if x_cal is None or y_cal is None or z_cal is None: x_cal, y_cal, z_cal = self.calibration_model.get_parameter_value_with_unit( "array_element_position_ground" ) x_cal, y_cal, z_cal = [coord.to(u.m).value for coord in (x_cal, y_cal, z_cal)] cal_vect = np.array([x_cal, y_cal, z_cal]) x_tel, y_tel, z_tel = self._get_telescope_position_ground_with_axis_offset( x_cal=x_cal * u.m, y_cal=y_cal * u.m, z_cal=z_cal * u.m, ) x_tel, y_tel, z_tel = [coord.to(u.m).value for coord in (x_tel, y_tel, z_tel)] tel_vect = np.array([x_tel, y_tel, z_tel]) direction_vector = tel_vect - cal_vect # pointing vector from calibration device to telescope pointing_vector = np.round(direction_vector / np.linalg.norm(direction_vector), 6) # Calculate telescope theta and phi angles tel_theta = 180 - np.round( np.rad2deg(np.arccos(direction_vector[2] / np.linalg.norm(direction_vector))), 6 ) tel_phi = 180 - np.round( np.rad2deg(np.arctan2(direction_vector[1], direction_vector[0])), 6 ) # Calculate source beam theta and phi angles direction_vector_inv = direction_vector * -1 source_theta = np.round( np.rad2deg(np.arccos(direction_vector_inv[2] / np.linalg.norm(direction_vector_inv))), 6, ) source_phi = np.round( np.rad2deg(np.arctan2(direction_vector_inv[1], direction_vector_inv[0])), 6 ) return pointing_vector.tolist(), [tel_theta, tel_phi, source_theta, source_phi] def _get_telescope_position_ground_with_axis_offset(self, x_cal=None, y_cal=None, z_cal=None): """Return the telescope ground position corrected for the illuminator-axis offsets.""" if self.light_emission_config.get(LIGHT_SOURCE_POSITION) is not None and not ( self.uses_telescope_position_file(self.light_emission_config.get(LIGHT_SOURCE_POINTING)) ): return (0.0 * u.m,) * 3 x_tel, y_tel, z_tel = self.telescope_model.get_parameter_value_with_unit( "array_element_position_ground" ) axes_offsets = self.telescope_model.get_parameter_value_with_unit(AXES_OFFSETS) offset_1 = axes_offsets[0].to(u.m).value offset_2 = axes_offsets[1].to(u.m).value if len(axes_offsets) > 1 else 0.0 if np.isclose(offset_1, 0.0) and np.isclose(offset_2, 0.0): return x_tel, y_tel, z_tel if x_cal is None or y_cal is None or z_cal is None: x_cal, y_cal, z_cal = self.calibration_model.get_parameter_value_with_unit( "array_element_position_ground" ) x_tel_m = x_tel.to(u.m).value y_tel_m = y_tel.to(u.m).value z_tel_m = z_tel.to(u.m).value delta_x = x_cal.to(u.m).value - x_tel_m delta_y = y_cal.to(u.m).value - y_tel_m delta_z = z_cal.to(u.m).value - z_tel_m norm_horizontal = np.hypot(delta_x, delta_y) # Angle from North (y-axis) towards East (x-axis) phi = np.arctan2(delta_x, delta_y) el = np.arctan2(delta_z, norm_horizontal) # Unit vectors in telescope-pointing convention (phi, el). u_pointing = np.array( [ np.cos(el) * np.sin(phi), np.cos(el) * np.cos(phi), np.sin(el), ], dtype=float, ) u_xy = np.array([np.sin(phi), np.cos(phi), 0.0], dtype=float) altitude_axis = np.array([np.cos(phi), -np.sin(phi), 0.0], dtype=float) v_perpendicular = np.cross(altitude_axis, u_pointing) norm_v = np.linalg.norm(v_perpendicular) v_perpendicular /= norm_v corrected_position = np.array([x_tel_m, y_tel_m, z_tel_m], dtype=float) corrected_position += offset_1 * u_xy + offset_2 * v_perpendicular return ( corrected_position[0] * u.m, corrected_position[1] * u.m, corrected_position[2] * u.m, )
[docs] def write_telescope_position_file(self, illuminator_position=None): """ Write the telescope positions to a telescope_position file. The file will contain lines in the format: x y z r in cm Returns ------- Path The path to the generated telescope_position file. """ if illuminator_position is None: illuminator_position = self.get_illuminator_position() x_cal, y_cal, z_cal = illuminator_position x_tel, y_tel, z_tel = self._get_telescope_position_ground_with_axis_offset( x_cal=x_cal, y_cal=y_cal, z_cal=z_cal, ) x_tel, y_tel, z_tel = [coord.to(u.cm).value for coord in (x_tel, y_tel, z_tel)] radius = self.telescope_model.get_parameter_value_with_unit(TELESCOPE_SPHERE_RADIUS) radius = radius.to(u.cm).value # Convert radius to cm tel = self.get_file_name_token(self.light_emission_config.get("telescope") or "telescope") cal = self.get_file_name_token( self.light_emission_config.get("light_source") or "calibration" ) telescope_position_file = ( self.io_handler.get_output_directory("light_emission") / f"telescope_position_{tel}_{cal}.dat" ) telescope_position_file.write_text(f"{x_tel} {y_tel} {z_tel} {radius}\n", encoding="utf-8") return telescope_position_file
[docs] def get_illuminator_position(self): """Return the illuminator emission position in ground coordinates.""" pos = self.light_emission_config.get(LIGHT_SOURCE_POSITION) if pos is not None: return pos x_pos, y_pos, z_pos = self.calibration_model.get_parameter_value_with_unit( "array_element_position_ground" ) tower_height = self.calibration_model.get_parameter_value_with_unit( "illuminator_tower_height" ) return x_pos, y_pos, z_pos + tower_height
[docs] def get_illuminator_pointing_vector(self, pos=None): """Return illuminator pointing vector; prefer explicit config if available.""" pointing_vector = self.light_emission_config.get(LIGHT_SOURCE_POINTING) if pointing_vector is not None: return pointing_vector if pos is None: pos = self.get_illuminator_position() x_cal, y_cal, z_cal = pos return self._calibration_pointing_direction(x_cal, y_cal, z_cal)[0]
[docs] @staticmethod def uses_telescope_position_file(pointing_vector): """Return False for the downward beam used with an artificial telescope at the origin.""" try: vec = np.asarray(pointing_vector, dtype=float) except TypeError, ValueError: return True if vec.size < 3 or not np.all(np.isfinite(vec[:3])): return True is_default_down = np.allclose(vec[:3], [0.0, 0.0, -1.0], atol=1e-6) return not is_default_down
[docs] @staticmethod def get_file_name_token(value): """Return a filename token for a light source or telescope name.""" return "".join(ch if (ch.isalnum() or ch in ("-", "_")) else "_" for ch in str(value))
def _make_simtel_script(self): """Return the command to run sim_telarray on the output of the previous step.""" bypass = self._bypass_optics() theta, phi = (0.0, 0.0) if bypass else self._get_telescope_pointing() simtel_bin = str(self.context.software.sim_telarray_exe) parts = [ simtel_bin, f"-I{self.telescope_model.config_file_directory}", "-c", str(self.telescope_model.config_file_path), "-DNUM_TELESCOPES=1", ] altitude = ( self.site_model.get_parameter_value_with_unit(CORSIKA_OBSERVATION_LEVEL).to(u.m).value ) atmosphere = self.site_model.get_parameter_value("atmospheric_transmission") if str(atmosphere).lower().endswith(".ecsv"): # generated sim_telarray table parameter_data = self.site_model.parameters.get("atmospheric_transmission", {}) atmosphere = get_simtel_table_file_name(parameter_data) or ( f"atmospheric_transmission-{Path(self.telescope_model.config_file_path).stem}.dat" ) options = [ ("altitude", altitude), ("atmospheric_transmission", atmosphere), ("TRIGGER_TELESCOPES", "1"), ("TELTRIG_MIN_SIGSUM", "2"), ("PULSE_ANALYSIS", "-30"), ("MAXIMUM_TELESCOPES", 1), ("telescope_theta", f"{theta}"), ("telescope_phi", f"{phi}"), ] if bypass: options.append(("Bypass_Optics", f"{bypass}")) options.append(("random_seed", ",".join(str(seed) for seed in self._seeds()))) input_file = self.runner_service.get_file_name(file_type="iact_output") output_file = self.runner_service.get_file_name(file_type="sim_telarray_output") histo_file = self.runner_service.get_file_name(file_type="sim_telarray_histogram") options += [ ("power_law", "2.68"), ("input_file", f"{input_file}"), ("output_file", f"{output_file}"), ("histogram_file", f"{histo_file}"), ] for key, value in options: parts.extend(["-C", f"{key}={value}"]) log_file = self.runner_service.get_file_name(file_type="sim_telarray_log") command = build_accounting_command( parts, self.runner_service.get_file_name(file_type="sim_telarray_resources"), "sim_telarray", self.light_emission_config.get(RUN_NUMBER, 1), log_file=log_file, ) return sim_telarray_env_as_string() + shlex.join(command) + "\n"
[docs] def calculate_distance_focal_plane_calibration_device(self): """ Return the distance between the focal plane and a flasher (focal length + focus offset - z). Returns ------- astropy.units.Quantity Distance between calibration device and focal plane. """ focal_length = self.telescope_model.get_parameter_value_with_unit("focal_length").to(u.m) flasher_z = self.calibration_model.get_parameter_value_with_unit(FLASHER_POSITION)[2].to( u.m ) return focal_length + self._vertical_focus_offset() - flasher_z
def _vertical_focus_offset(self): """Return the vertical-telescope focus offset (zenith terms only if both are non-zero).""" values = self.telescope_model.get_parameter_value_with_unit("focus_offset") if values is None: return 0.0 * u.m values = value_conversion.ensure_list(values) offset = values[0].to(u.m) if len(values) < 4 or not (values[2].value and values[3].value): return offset reference = values[1].to(u.rad).value return ( offset + (values[2].value * (1.0 - np.cos(reference)) - values[3].value * np.sin(reference)) * values[0].unit.to(u.m) * u.m )
[docs] def validate_simulations(self): """Validate that the simulations were successful.""" validate_run_outputs.validate_sim_telarray( data_files=Path(self.runner_service.get_file_name(file_type="sim_telarray_output")), log_files=Path(self.runner_service.get_file_name(file_type="sim_telarray_log")), array_models=None, )