#!/usr/bin/python3
"""Read metadata from sim_telarray files."""
import logging
import re
from functools import cache
from eventio import EventIOFile
from eventio.iact import InputCard
from eventio.simtel import HistoryMeta
from simtools import dependency_versions
from simtools.utils import names
_logger = logging.getLogger(__name__)
LEGACY_METADATA_ALIASES = {
"simtools_corsika_config_version": "simtools_corsika_config_tag",
"simtools_corsika_opt_patch_version": "simtools_corsika_opt_patch_tag",
"simtools_corsika_version": "simtools_corsika_build_id",
"simtools_hessio_version": "simtools_hessio_tag",
"simtools_simtel_version": "simtools_simtel_tag",
"simtools_stdtools_version": "simtools_stdtools_tag",
}
def _decode_dictionary(meta, encoding="utf8"):
"""Decode metadata dictionary."""
def safe_decode(byte_str, encoding, errors="ignore"):
return byte_str.decode(encoding, errors=errors)
try:
return {k.decode(encoding, errors="ignore"): v.decode(encoding) for k, v in meta.items()}
except UnicodeDecodeError as e:
_logger.warning(
f"Unable to decode metadata with encoding {encoding}: {e}. "
"Falling back to 'utf-8' with errors='ignore'."
)
return {safe_decode(k, encoding): safe_decode(v, encoding) for k, v in meta.items()}
[docs]
def get_sim_telarray_telescope_id(telescope_name, file):
"""
Return the telescope ID for a given telescope name in a sim_telarray file.
Translates e.g. 'LSTN-01' to the corresponding telescope ID.
Parameters
----------
telescope_name: str
Name of the telescope.
file: str
Path to the sim_telarray file.
Returns
-------
int, None
Telescope ID. Returns None if not found.
"""
_, telescope_meta = read_sim_telarray_metadata(file)
telescope_name_to_sim_telarray_id = {}
for tel_id in telescope_meta.keys():
_optics_name = telescope_meta[tel_id].get("optics_config_variant", None)
_camera_name = telescope_meta[tel_id].get("camera_config_variant", None)
if _optics_name == _camera_name and _optics_name == telescope_name:
telescope_name_to_sim_telarray_id[telescope_name] = tel_id
return telescope_name_to_sim_telarray_id.get(telescope_name, None)
[docs]
def get_sim_telarray_telescope_id_to_telescope_name_mapping(file):
"""
Return a mapping of telescope IDs to telescope names from a sim_telarray file.
Parameters
----------
file: str
Path to the sim_telarray file.
Returns
-------
dict
Dictionary mapping telescope IDs to telescope names.
"""
_, telescope_meta = read_sim_telarray_metadata(file)
telescope_map = {}
for i, (tel_id, meta) in enumerate(telescope_meta.items()):
try:
telescope_name = names.validate_array_element_name(
meta.get("optics_config_variant", f"Unknown-{tel_id}")
)
except ValueError:
telescope_name = _guess_telescope_name_for_legacy_files(i, file)
if telescope_name is not None:
telescope_map[tel_id] = telescope_name
return telescope_map
def _guess_telescope_name_for_legacy_files(tel_counter, file):
"""
Guess telescope names for legacy prod6 sim_telarray files with incomplete metadata.
Parameters
----------
tel_counter: int
Telescope counter, used to index into the telescope list.
file: str, Path
Path to the sim_telarray file.
Returns
-------
str, None
Guessed telescope name or None if not found.
"""
telescope_list = _legacy_merge_msts(_get_telescope_list_from_input_card(file))
try:
return names.validate_array_element_name(telescope_list[tel_counter])
except IndexError, ValueError:
pass
return None
@cache
def _get_telescope_list_from_input_card(file):
r"""
Return telescope list from CORSIKA input card.
Note hardwired regex pattern with telescope naming convention.
This function is intended for legacy files generated for prod6,
where metadata is incomplete.
Expected format in input card:
.. code-block:: console
TELESCOPE -70.91E2 -52.35E2 45.00E2 12.50E2 # (ID=1) LSTN 01 2B5\n
Parameters
----------
file: str, Path
Path to the sim_telarray file.
Returns
-------
list
List of telescope names as found in CORSIKA input card.
"""
tel_re = re.compile(
r"\s*TELESCOPE\s+"
r"[-\d.E]+\s+[-\d.E]+\s+[-\d.E]+\s+[-\d.E]+\s+"
r"#\s*\(ID=\d+\)\s+"
r"([A-Z0-9]+)\s+" # MSTS / LSTS / SSTS / MST2 / LSTN / MSTN
r"(\d+)\s+" # numeric telescope id
r"[A-Z0-9]+", # internal label (ignored)
re.MULTILINE,
)
with EventIOFile(file) as f:
for o in f:
if isinstance(o, InputCard):
text = o.parse().decode("utf-8")
return [f"{tel_type}-{tel_id}" for tel_type, tel_id in tel_re.findall(text)]
return []
def _legacy_merge_msts(telescopes):
"""
Rename additional 'MST2' telescopes to 'MSTS' or 'MSTN' with incremented IDs.
Expect that main list first element is a telescope with a valid side flag.
This function is very specific and intended for legacy files
generated for prod6, where metadata is incomplete.
Parameters
----------
telescopes: list
List of telescopes
Returns
-------
list
List of telescopes with 'MST2' renamed to 'MSTS' or 'MSTN' and incremented IDs.
"""
tel2_indices = [
index for index, telescope in enumerate(telescopes) if re.search(r"^MST2", telescope)
]
if not tel2_indices:
return telescopes
mst_numbers = [
int(num)
for telescope in telescopes
if "MST" in telescope and not re.search(r"^MST2", telescope)
for num in re.findall(r"\d+", telescope)
]
max_mst_id = max(mst_numbers) if mst_numbers else 0
site = names.get_site_from_array_element_name(telescopes[0])
site_char = "S" if site.lower() == "south" else "N"
merged_tels = list(telescopes)
for i, index in enumerate(tel2_indices):
merged_tels[index] = f"MST{site_char}-{max_mst_id + i + 1:02d}"
return merged_tels
