"""Array model represents an observatory consisting of site, telescopes, and further devices."""
import logging
import os
import tarfile
from pathlib import Path
import astropy.units as u
from astropy.table import QTable
from simtools.context import RunContext
from simtools.data_model import data_reader, schema
from simtools.io.ascii_handler import to_builtin
from simtools.model.calibration_model import CalibrationModel
from simtools.model.model_parameter import InvalidModelParameterError
from simtools.model.model_utils import read_overwrite_model_parameter_dict
from simtools.model.site_model import SiteModel
from simtools.model.telescope_model import TelescopeModel
from simtools.simulation.configuration import get_model_writer
from simtools.utils import names
[docs]
class ArrayModel:
"""
Representation of an observatory consisting of site, telescopes, and further devices.
Parameters
----------
model_version: str
Model version.
label: str, optional
Instance label. Used for output file naming.
site: str, optional
Site name.
layout_name: str, optional
Layout name.
array_elements: Union[str, Path, List[str]], optional
Array element definitions (list of array element or path to file with
the array element positions).
calibration_device_types: List[str], optional
List of calibration device types (e.g., 'flat_fielding') attached to each telescope.
overwrite_model_parameters: dict or str, optional
Dictionary with model parameters to overwrite, or file path to YAML file.
"""
def __init__(
self,
model_version,
label=None,
site=None,
layout_name=None,
array_elements=None,
calibration_device_types=None,
overwrite_model_parameters=None,
ignore_software_version=False,
model_directory_subdir=None,
model_reader=None,
context=None,
):
"""Initialize ArrayModel."""
self._logger = logging.getLogger(__name__)
self.model_version = model_version
self.ignore_software_version = ignore_software_version
self.context = context or RunContext.from_environment()
self.model_reader = model_reader or self.context.model_reader
if self.model_reader is None:
raise RuntimeError("No simulation model reader is configured.")
self.label = label
self.layout_name = (
layout_name[0]
if isinstance(layout_name, list) and len(layout_name) == 1
else layout_name
)
self._config_file_path = None
self._config_file_directory = None
self.io_handler = self.context.paths
self.overwrite_model_parameter_dict = read_overwrite_model_parameter_dict(
overwrite_model_parameters, context=self.context
)
self.array_elements, self.site_model, self.telescope_models, self.calibration_models = (
self._initialize(site, array_elements, calibration_device_types)
)
if model_directory_subdir:
self._config_file_directory = self.io_handler.get_model_configuration_directory(
model_version=self.model_version, sub_dir=model_directory_subdir
)
models = [self.site_model, *self.telescope_models.values()]
for model in models:
model._config_file_directory = self._config_file_directory
self.configuration_writers = {}
self.sim_telarray_seed = None
def _initialize(self, site, array_elements_config, calibration_device_types):
"""
Initialize ArrayModel taking different configuration options into account.
Parameters
----------
site: str
Site name.
array_elements_config: Union[str, Path, List[str]]
Array element definitions.
calibration_device_types: str
Calibration device types.
Returns
-------
dict
Dict with telescope positions.
SiteModel
Site model.
dict
Dict with telescope models.
"""
self._logger.debug(f"Getting site parameters from the model repository ({site})")
site_model = SiteModel(
site=names.validate_site_name(site),
model_version=self.model_version,
label=self.label,
overwrite_model_parameter_dict=self.overwrite_model_parameter_dict,
ignore_software_version=self.ignore_software_version,
model_reader=self.model_reader,
context=self.context,
)
# Case 1: array_elements is a file name
if isinstance(array_elements_config, str | Path):
array_elements = self._load_array_element_positions_from_file(
array_elements_config, site_model.site
)
# Case 2: array elements is a list of elements
elif isinstance(array_elements_config, list) and len(array_elements_config) > 0:
array_elements = self._get_array_elements_from_list(array_elements_config, site_model)
# Case 3: array elements defined in the model repository by array layout name
elif self.layout_name is not None:
array_elements = self._get_array_elements_from_list(
site_model.get_array_elements_for_layout(self.layout_name)
)
else:
raise ValueError(
"No array elements found. "
"Possibly missing valid layout name or missing telescope list."
)
telescope_models, calibration_models = self._build_telescope_models(
site_model, array_elements, calibration_device_types
)
return array_elements, site_model, telescope_models, calibration_models
@property
def config_file_path(self):
"""
Return the path of the array config file for sim_telarray.
Returns
-------
Path
Path of the exported config file for sim_telarray.
"""
return self._config_file_path or self.get_config_directory().joinpath(
names.sim_telarray_config_file_name(
array_name=self.layout_name, site=self.site_model.site
)
)
@property
def number_of_telescopes(self):
"""
Return the number of telescopes.
Returns
-------
int
Number of telescopes.
"""
return len(self.telescope_models)
@property
def site(self):
"""
Return site.
Returns
-------
str
Site name.
"""
return self.site_model.site
def _build_telescope_models(self, site_model, array_elements, calibration_device_types):
"""
Build telescope models for all telescopes of this array.
Adds calibration device models, if requested through the calibration_device_types argument.
Calibration device models are stored in a dictionary with the telescope name as key (to
identify the calibration device model on a given telescope).
Includes reading of telescope model parameters from the model repository.
The array is defined in the array_elements dictionary. Array element positions
are read from the model repository if no values are given in this dictionary.
Parameters
----------
site_model: SiteModel
Site model.
array_elements: dict
Dict with array elements.
calibration_device_types: List[str]
List of calibration device types (e.g., 'flat_fielding')
Returns
-------
dict, dict
Dictionaries with telescope and calibration models.
"""
telescope_models, calibration_models = {}, {}
for element_name in array_elements:
if names.get_collection_name_from_array_element_name(element_name) != "telescopes":
continue
telescope_models[element_name] = TelescopeModel(
site=site_model.site,
telescope_name=element_name,
model_version=self.model_version,
label=self.label,
overwrite_model_parameter_dict=self.overwrite_model_parameter_dict,
ignore_software_version=self.ignore_software_version,
model_reader=self.model_reader,
context=self.context,
)
calibration_models[element_name] = self._build_calibration_models(
telescope_models[element_name],
site_model,
calibration_device_types,
)
return telescope_models, calibration_models
def _build_calibration_models(self, telescope_model, site_model, calibration_device_types):
"""
Build calibration device models for all telescopes in the array.
A telescope can have multiple calibration devices of different types.
Returns
-------
dict
Dict with calibration device models.
"""
calibration_models = {}
for calibration_device_type in calibration_device_types or []:
device_name = telescope_model.get_calibration_device_name(calibration_device_type)
if device_name is None:
continue
calibration_models[device_name] = CalibrationModel(
site=site_model.site,
calibration_device_model_name=device_name,
model_version=self.model_version,
label=self.label,
overwrite_model_parameter_dict=self.overwrite_model_parameter_dict,
model_reader=self.model_reader,
context=self.context,
)
return calibration_models
[docs]
def print_telescope_list(self):
"""Print list of telescopes."""
for tel_name, data in self.telescope_models.items():
print(f"Name: {tel_name}\t Model: {data.name}")
[docs]
def export_config_files(self, simulation_software="sim_telarray"):
"""Export array and telescope configurations for selected software.
Parameters
----------
simulation_software : str
Registered configuration format.
"""
return get_model_writer(
self, simulation_software, context=self.context
).export_config_files()
[docs]
def export_simtel_telescope_config_files(self):
"""Export sim_telarray configuration files for all telescopes into the model directory."""
return get_model_writer(self, context=self.context).export_simtel_telescope_config_files()
[docs]
def export_sim_telarray_config_file(self):
"""Export sim_telarray configuration file for the array into the model directory."""
return get_model_writer(self, context=self.context).export_sim_telarray_config_file()
[docs]
def export_all_simtel_config_files(self):
"""
Export sim_telarray config file for the array and for each individual telescope.
Config files are exported into the output model directory.
"""
return get_model_writer(self, context=self.context).export_all_simtel_config_files()
[docs]
def get_config_directory(self):
"""
Get the path of the array config directory for sim_telarray.
Returns
-------
Path
Path of the config directory path for sim_telarray.
"""
if self._config_file_directory is None:
self._config_file_directory = self.io_handler.get_model_configuration_directory(
model_version=self.model_version
)
return self._config_file_directory
[docs]
def pack_model_files(self):
"""
Pack all model files into a tar.gz archive.
Returns
-------
Path
Path of the packed model files archive.
"""
model_files = list(Path(self.get_config_directory()).rglob("*"))
if not model_files:
self._logger.warning("No model files found to pack.")
return None
archive_name = self.get_config_directory() / f"model_files_{self.model_version}.tar.gz"
files = [Path(path) for path in model_files]
base = Path(os.path.commonpath([path.resolve() for path in files])).resolve()
if any(not path.is_file() or not path.resolve().is_relative_to(base) for path in files):
raise ValueError("Model files contain an unsafe path.")
# Archive creation, not expansion of untrusted input: files are validated above and
# added non-recursively.
with tarfile.open(archive_name, "w:gz", compresslevel=1) as archive: # NOSONAR
for path in files:
archive.add(
path,
arcname=str(Path(f"model/{self.model_version}") / path.name),
recursive=False,
)
self._logger.info(f"Packed model files into {archive_name}")
return archive_name
def _load_array_element_positions_from_file(self, array_elements_file, site):
"""
Load array element (e.g. telescope) positions from a file into a dict.
Dictionary format: {telescope_name: {position_x: x, position_y: y, position_z: z}}
Parameters
----------
array_elements_file: Union[str, Path]
Path to the file with the array element positions.
site: str
Site name used when building the returned parameter dictionaries.
Returns
-------
dict
Dict with telescope positions.
"""
table = data_reader.read_table_from_file(
file_name=array_elements_file, context=self.context
)
return {
row["telescope_name"]: self._get_telescope_position_parameter(
row["telescope_name"], site, row["position_x"], row["position_y"], row["position_z"]
)
for row in table
}
def _get_telescope_position_parameter(
self, telescope_name, site, x, y, z, parameter_version=None
):
"""
Return telescope position parameters in model-repository format.
Parameters
----------
telescope_name: str
Name of the telescope.
site: str
Site name.
x: astropy.Quantity
X ground position.
y: astropy.Quantity
Y ground position.
z: astropy.Quantity
Z ground position.
Returns
-------
dict
Dict with telescope position parameters.
"""
return {
"schema_version": schema.get_model_parameter_schema_version(),
"parameter": "array_element_position_ground",
"instrument": telescope_name,
"site": site,
"parameter_version": parameter_version,
"unique_id": None,
"value": [x.to("m").value, y.to("m").value, z.to("m").value],
"unit": "m",
"type": "float64",
"file": False,
"model_parameter_schema_version": "0.1.0",
}
def _get_array_elements_from_list(self, array_elements_list, site_model=None):
"""
Return dictionary with array elements from a list of telescope names.
Input list can contain telescope names (e.g, LSTN-01) or a telescope
type (e.g., MSTN). In the latter case, all telescopes of this specific
type are added.
Parameters
----------
array_elements_list: list
List of telescope names.
site_model: SiteModel
Site model.
Returns
-------
dict
Dict with array elements.
"""
array_elements_dict = {}
for name in array_elements_list:
try:
array_elements_dict[names.validate_array_element_name(name)] = None
except ValueError:
array_elements_dict.update(self._get_all_array_elements_of_type(name, site_model))
return array_elements_dict
def _get_all_array_elements_of_type(self, array_element_type, site_model):
"""
Return all array elements of a specific type.
Parameters
----------
array_element_type : str
Type of the array element (e.g. LSTN, MSTS)
site_model: SiteModel
Site model.
Returns
-------
dict
Dict with array elements.
"""
return self._get_array_elements_from_list(
site_model.get_array_elements_of_type(array_element_type)
)
[docs]
def export_array_elements_as_table(
self, coordinate_system="ground", include_calibration_array_elements=False
):
"""
Export array elements positions to astropy table.
Parameters
----------
coordinate_system: str
Positions are exported in this coordinate system.
include_calibration_array_elements: bool
Include calibration array elements in output table.
Returns
-------
astropy.table.QTable
Astropy table with the array-elements layout information.
"""
table = QTable(meta={"array_name": self.layout_name, "site": self.site_model.site})
name, pos_x, pos_y, pos_z, tel_r = [], [], [], [], []
for tel_name, data in self.telescope_models.items():
name.append(tel_name)
xyz = data.position(coordinate_system=coordinate_system)
pos_x.append(xyz[0])
pos_y.append(xyz[1])
pos_z.append(xyz[2])
try:
# Add tests of KeyError after calibration elements gain positions.
tel_r.append(data.get_parameter_value_with_unit("telescope_sphere_radius"))
except (
KeyError,
InvalidModelParameterError,
): # not all array elements have a sphere radius
tel_r.append(float("nan") * u.m)
if include_calibration_array_elements:
calibration_position_parameter = f"array_element_position_{coordinate_system}"
for array_element_name in self.array_elements:
if (
names.get_collection_name_from_array_element_name(array_element_name)
!= "calibration_devices"
):
continue
calibration_model = CalibrationModel(
site=self.site_model.site,
calibration_device_model_name=array_element_name,
model_version=self.model_version,
label=self.label,
overwrite_model_parameter_dict=self.overwrite_model_parameter_dict,
model_reader=self.model_reader,
context=self.context,
)
xyz = calibration_model.get_parameter_value_with_unit(
calibration_position_parameter
)
name.append(array_element_name)
pos_x.append(xyz[0])
pos_y.append(xyz[1])
pos_z.append(xyz[2])
try:
tel_r.append(
calibration_model.get_parameter_value_with_unit(
"array_element_sphere_radius"
)
)
except KeyError, InvalidModelParameterError:
tel_r.append(float("nan") * u.m)
table["telescope_name"] = name
if coordinate_system == "ground":
table["position_x"] = pos_x
table["position_y"] = pos_y
table["position_z"] = pos_z
elif coordinate_system == "utm":
table["utm_east"] = pos_x
table["utm_north"] = pos_y
table["altitude"] = pos_z
table["sphere_radius"] = tel_r
table.sort("telescope_name")
return table
[docs]
def initialize_seeds(self, zenith_angle=None, azimuth_angle=None):
"""Initialize sim_telarray seeds for instrument and shower simulations."""
return get_model_writer(self, context=self.context).initialize_seeds(
zenith_angle=zenith_angle, azimuth_angle=azimuth_angle
)
