"""Multi-illuminator simulation orchestration with parallel execution."""
import logging
from pathlib import Path
from simtools.context import RunContext
from simtools.job_execution.execution import map_ordered
from simtools.model.illuminator_visibility import IlluminatorTelescopeVisibility
from simtools.simtel.simulator_light_emission import SimulatorLightEmission
from simtools.utils import value_conversion
_logger = logging.getLogger(__name__)
_NO_RESULTS_MSG = "No simulations have been run yet. Call simulate() first."
def _get_job_model_directory(config, illuminator, telescope, wavelength):
"""Return the isolated generated-model directory for one simulation job."""
output_path = config.get("output_path")
if output_path is None:
return None
wavelength_nm = wavelength.to_value("nm")
job_name = f"{illuminator}_{telescope}_{wavelength_nm:g}nm"
return Path(output_path) / "model" / str(config["model_version"]) / job_name
def _simulate_illuminator_telescope_pair(job_spec):
"""
Simulate a single illuminator-telescope pair.
This function is executed in a separate process.
Parameters
----------
job_spec : dict
Job specification containing:
- illuminator: str - illuminator name
- telescope: str - telescope name
- site: str - site name (North/South)
- label: str - base label for the simulation
- config: dict - light emission configuration
- model_source: dict - serializable model-source selection
Returns
-------
dict
Result dictionary with simulation outcome:
- illuminator: str
- telescope: str
- success: bool
- error: str or None
"""
illuminator = job_spec["illuminator"]
telescope = job_spec["telescope"]
config = job_spec["config"].copy()
try:
context = RunContext.from_spec(job_spec["context"])
model_reader = context.model_reader
# Update configuration for this specific pair
config["telescope"] = telescope
config["light_source"] = illuminator
label = job_spec["label"]
_logger.info(f"Starting simulation for {illuminator} -> {telescope}")
simulator = SimulatorLightEmission(
light_emission_config=config,
telescope=telescope,
label=label,
model_reader=model_reader,
context=context,
)
simulator.simulate()
simulator.validate_simulations()
_logger.info(f"Completed simulation for {illuminator} -> {telescope}")
return {
"illuminator": illuminator,
"telescope": telescope,
"wavelength": job_spec.get("wavelength"),
"success": True,
"error": None,
}
except Exception as exc: # pylint: disable=broad-except
_logger.warning(f"Failed simulation for {illuminator} -> {telescope}: {exc}")
return {
"illuminator": illuminator,
"telescope": telescope,
"wavelength": job_spec.get("wavelength"),
"success": False,
"error": str(exc),
}
[docs]
class MultiIlluminatorSimulator:
"""
Orchestrate parallel simulations of multiple illuminator-telescope pairs.
This class manages simulations for all valid illuminator-telescope combinations
defined in a visibility table, running them in parallel across multiple CPU cores.
Parameters
----------
config : dict
Base configuration for light emission simulations. Will be updated for
each illuminator-telescope pair. Should contain:
- site: str (North/South)
- model_version: str
- number_of_events: int
- other SimulatorLightEmission parameters
visibility_data : list of dict, optional
Structured visibility records with ``illuminator_id``, ``telescope_id``,
and ``visible`` keys.
If not provided, the visibility table is retrieved from the site model
using the site and model_version from config.
label : str, optional
Base label for all simulations. Pair-specific labels will be appended.
max_workers : int or None, optional
Maximum number of parallel worker processes. If None, uses 60% of CPU cores.
If <= 0, uses all available cores.
"""
def __init__(
self,
config,
visibility_data=None,
label=None,
max_workers=None,
backend="local",
backend_config=None,
model_reader=None,
context=None,
):
"""Initialize the multi-illuminator simulator."""
self._logger = logging.getLogger(__name__)
self.context = context or RunContext.from_environment()
self.model_reader = model_reader or self.context.model_reader
if self.model_reader is None:
raise ValueError("A simulation model reader is required for illuminator simulations.")
self.context = self.context.replace(model_reader=self.model_reader)
# Load visibility table from provided data or from the site model
if visibility_data is None:
visibility_data = self._load_visibility_from_site_model(
config, self.model_reader, context=self.context
)
self.visibility = IlluminatorTelescopeVisibility(visibility_data)
self.base_config = config
self.label = label or "multi_illuminator"
self.max_workers = max_workers
self.backend = backend
self.backend_config = backend_config
self.results = None
self._logger.info("Using execution backend %s", self.backend)
@staticmethod
def _load_visibility_from_site_model(config, model_reader=None, *, context):
"""
Load visibility data from the model repository.
Parameters
----------
config : dict
Configuration containing "site" and "model_version" keys.
Returns
-------
list of dict
Structured visibility records.
"""
# Import here to avoid loading SiteModel when visibility_data is provided directly
from simtools.model.site_model import SiteModel # pylint: disable=import-outside-toplevel
site_model = SiteModel(
site=config["site"],
model_version=config["model_version"],
model_reader=model_reader,
context=context,
)
return site_model.get_parameter_value("illuminator_telescope_visibility")
[docs]
def simulate(self, wavelengths=None, illuminators=None, telescopes=None):
"""
Run simulations for all valid illuminator-telescope pairs in parallel.
Optionally simulate across multiple wavelengths. If wavelengths are not
specified and not in base_config, all model wavelengths will be used.
Parameters
----------
wavelengths : list of astropy.units.Quantity, optional
List of wavelengths to simulate. If None, uses wavelength from base_config
if present, otherwise all model wavelengths.
illuminators : list of str, optional
Restrict simulations to specific illuminators. If None, simulate all.
telescopes : list of str, optional
Restrict simulations to specific telescopes. If None, simulate all.
Returns
-------
list of dict
List of simulation results. Each dict contains:
- illuminator: str
- telescope: str
- wavelength: astropy.units.Quantity (if wavelength simulation)
- success: bool
- error: str or None
"""
# Get all valid pairs
all_pairs = self.visibility.get_valid_pairs()
# Apply filters first (before using a pair to fetch wavelengths)
if illuminators is not None:
all_pairs = [(ill, tel) for ill, tel in all_pairs if ill in illuminators]
if telescopes is not None:
all_pairs = [(ill, tel) for ill, tel in all_pairs if tel in telescopes]
if not all_pairs:
self._logger.warning("No valid pairs to simulate after filtering")
self.results = []
return self.results
# Determine wavelengths: explicit param > config > model
if wavelengths is not None:
wavelengths = value_conversion.ensure_list(wavelengths)
elif "wavelength" in self.base_config and self.base_config["wavelength"] is not None:
wavelengths = value_conversion.ensure_list(self.base_config["wavelength"])
else:
# Fetch from model using a sample pair
self._logger.info("No wavelength specified, fetching from model")
sample_illuminator, sample_telescope = all_pairs[0]
config_for_query = self.base_config.copy()
config_for_query["telescope"] = sample_telescope
config_for_query["light_source"] = sample_illuminator
wavelengths = SimulatorLightEmission.get_available_wavelengths_from_config(
config_for_query, model_reader=self.model_reader, context=self.context
)
self._logger.info(
f"Using {len(wavelengths)} wavelengths from model: "
f"{[w.value for w in wavelengths]} nm"
)
# Build job specs for all wavelength-pair combinations
job_specs = []
model_source = self.model_reader.source_config
if not isinstance(model_source, dict):
model_source = None
for wavelength in wavelengths:
for illuminator, telescope in all_pairs:
# Create config with wavelength
config_with_wl = self.base_config.copy()
config_with_wl["wavelength"] = wavelength
model_directory = _get_job_model_directory(
self.base_config, illuminator, telescope, wavelength
)
if model_directory is not None:
config_with_wl["model_directory"] = model_directory
job_spec = {
"illuminator": illuminator,
"telescope": telescope,
"wavelength": wavelength,
"site": self.base_config.get("site"),
"label": self.label,
"config": config_with_wl,
"context": self.context.to_spec(),
}
job_specs.append(job_spec)
self._logger.info(
f"Simulating {len(all_pairs)} pair(s) x {len(wavelengths)} wavelength(s) "
f"= {len(job_specs)} jobs with {self.max_workers} workers"
)
# Execute in parallel
self.results = map_ordered(
_simulate_illuminator_telescope_pair,
job_specs,
backend=self.backend,
max_workers=self.max_workers,
backend_config=self.backend_config,
context=self.context,
)
# Log summary
successful = sum(1 for r in self.results if r["success"])
failed = len(self.results) - successful
self._logger.info(
f"Simulation complete: {successful} successful, {failed} failed "
f"across {len(wavelengths)} wavelength(s)"
)
return self.results
