"""Model data writer module."""
import logging
from pathlib import Path
import packaging.version
from astropy.io.registry.base import IORegistryError
import simtools.utils.general as gen
from simtools.context import RunContext
from simtools.data_model import schema, validate_data
from simtools.data_model.json_validation import validate_finite_json_values
from simtools.data_model.metadata_collector import MetadataCollector
from simtools.io import ascii_handler
from simtools.io.output_paths import OutputPaths
from simtools.utils import names, value_conversion
[docs]
class ModelDataWriter:
"""
Writer for simulation model data and metadata.
Parameters
----------
output_file: str
Name of output file.
output_file_format: str
Format of output file.
output_path: str or Path
Path to output file.
"""
def __init__(
self,
output_file=None,
output_file_format=None,
output_path=None,
model_reader=None,
*,
context=None,
):
"""Initialize model data writer."""
self._logger = logging.getLogger(__name__)
self.context = context or RunContext.from_environment()
self.model_reader = model_reader or self.context.model_reader
self.schema_dict = {}
self.output_label = "default"
selected_output_path = output_path or self.context.args.get("output_path")
if selected_output_path is not None:
paths = OutputPaths(
output_path=selected_output_path,
model_path=self.context.paths.model_path,
)
self.context = self.context.replace(paths=paths)
self.io_handler = self.context.paths
try:
self.output_file = (
self.io_handler.get_output_file(
file_name=output_file, output_path_label=self.output_label
)
if self.output_label in self.io_handler.output_path
else None
)
except TypeError:
self.output_file = None
self.output_file_format = self._derive_data_format(output_file_format, self.output_file)
[docs]
@staticmethod
def write_product_data(
output_file=None,
metadata=None,
product_data=None,
output_file_format="ascii.ecsv",
validate_schema_file=None,
metadata_output_file=None,
context=None,
):
"""
Write model data and metadata (as static method).
Parameters
----------
output_file: string or Path
Name of output file (args["output_file"] is used if this parameter is not set).
metadata: MetadataCollector object
Metadata to be written.
product_data: astropy Table
Model data to be written
output_file_format: str
Format of output file.
validate_schema_file: str
Schema file used in validation of output data.
metadata_output_file: str or Path, optional
Separate output path for application metadata. If omitted, metadata is
written next to the product data.
"""
writer = ModelDataWriter(
output_file=output_file,
output_file_format=output_file_format,
context=context,
)
skip_output_validation = writer.context.args.get("skip_output_validation", True)
if validate_schema_file is not None and not skip_output_validation:
product_data = writer.validate_and_transform(
product_data_table=product_data,
validate_schema_file=validate_schema_file,
)
writer.write_data(
metadata=metadata,
product_data=product_data,
metadata_output_file=metadata_output_file,
)
[docs]
@staticmethod
def write_model_parameter(
parameter_name,
value,
instrument,
parameter_version,
output_file,
output_path=None,
metadata_input_dict=None,
unit=None,
model_parameter_schema_version=None,
check_for_existing_parameter=True,
model_reader=None,
context=None,
):
"""
Generate model repository-style model parameter dict and write it to json file.
Parameters
----------
parameter_name: str
Name of the parameter.
value: any
Value of the parameter.
instrument: str
Name of the instrument.
parameter_version: str
Version of the parameter.
output_file: str
Name of output file.
output_path: str or Path
Path to output file.
metadata_input_dict: dict
Input to metadata collector.
unit: str
Unit of the parameter value (if applicable and value is not of type astropy Quantity).
model_parameter_schema_version: str, None
Version of the model parameter schema (if None, use schema version from schema dict).
check_for_existing_parameter: bool
If True, check whether the parameter version exists before writing.
model_reader: object, optional
Reader used for the existing-parameter check.
Returns
-------
dict
Validated parameter dictionary.
"""
writer = ModelDataWriter(
output_file=output_file,
output_file_format="json",
output_path=output_path,
model_reader=model_reader,
context=context,
)
if check_for_existing_parameter and not writer.context.args.get(
"ignore_existing_parameter_version", False
):
writer.check_for_existing_parameter(parameter_name, instrument, parameter_version)
output_file = writer.io_handler.get_output_file(
output_file, output_path_label=writer.output_label
)
unique_id = None
metadata = None
if metadata_input_dict is not None:
metadata_input_dict = dict(metadata_input_dict)
metadata_input_dict["output_file"] = Path(output_file).name
metadata_input_dict["output_file_format"] = Path(output_file).suffix.lstrip(".")
if metadata_input_dict.get("activity_id") is None:
activity_id = gen.extract_uuid7_from_path(output_file)
if activity_id is not None:
metadata_input_dict["activity_id"] = activity_id
metadata = MetadataCollector(args_dict=metadata_input_dict, context=context)
unique_id = (
metadata.get_top_level_metadata().get("cta", {}).get("product", {}).get("id")
)
_json_dict = writer.get_validated_parameter_dict(
parameter_name,
value,
instrument,
parameter_version,
unique_id,
model_parameter_schema_version=model_parameter_schema_version,
unit=unit,
)
output_file = writer.write_model_parameter_dict_json(output_file, _json_dict)
if metadata is not None:
metadata.write(output_file)
return _json_dict
[docs]
def check_for_existing_parameter(self, parameter_name, instrument, parameter_version):
"""
Check if a parameter with the same version exists in the model repository.
Parameters
----------
parameter_name: str
Name of the parameter.
instrument: str
Name of the instrument.
parameter_version: str
Version of the parameter.
Raises
------
ValueError
If parameter with the same version exists in the model repository.
"""
model_reader = self.model_reader or self.context.model_reader
if model_reader is None:
raise RuntimeError("No simulation model reader is configured.")
if not model_reader.is_configured():
return
try:
model_reader.get_model_parameter(
parameter=parameter_name,
parameter_version=parameter_version,
site=names.get_site_from_array_element_name(instrument),
array_element_name=instrument,
)
except ValueError:
pass # parameter does not exist - expected behavior
else:
raise ValueError(
f"Parameter {parameter_name} with version {parameter_version} already exists."
)
[docs]
def get_validated_parameter_dict(
self,
parameter_name,
value,
instrument,
parameter_version,
unique_id=None,
schema_version=None,
unit=None,
model_parameter_schema_version=None,
):
"""
Get validated parameter dictionary.
Parameters
----------
parameter_name: str
Name of the parameter.
value: any
Value of the parameter.
instrument: str
Name of the instrument.
parameter_version: str
Version of the parameter.
schema_version: str
Version of the schema.
unique_id: str
Unique ID of the parameter set (from metadata).
unit: str
Unit of the parameter value (if applicable and value is not an astropy Quantity).
model_parameter_schema_version: str, None
Version of the model parameter schema (if None, use schema version from schema dict).
Returns
-------
dict
Validated parameter dictionary.
"""
self._logger.debug(f"Getting validated parameter dictionary for {instrument}")
self.schema_dict, schema_file = self._read_schema_dict(
parameter_name, model_parameter_schema_version
)
if unit is None:
value, unit = value_conversion.split_value_and_unit(value)
validate_finite_json_values(value)
data_dict = {
"schema_version": schema.get_model_parameter_schema_version(schema_version),
"parameter": parameter_name,
"instrument": instrument,
"site": names.get_site_from_array_element_name(instrument),
"parameter_version": parameter_version,
"unique_id": unique_id,
"value": value,
"unit": unit,
"type": self._get_parameter_type(),
"file": self._parameter_is_a_file(),
"model_parameter_schema_version": model_parameter_schema_version
or self.schema_dict.get("schema_version", "0.1.0"),
}
return self.validate_and_transform(
product_data_dict=data_dict,
validate_schema_file=schema_file,
is_model_parameter=True,
)
def _read_schema_dict(self, parameter_name, schema_version):
"""
Read schema dict for given parameter name and version.
Use newest schema version if schema_version is None.
Parameters
----------
parameter_name: str
Name of the parameter.
schema_version: str
Schema version.
Returns
-------
dict
Schema dictionary.
"""
schema_file = schema.get_model_parameter_schema_file(parameter_name)
schemas = ascii_handler.collect_data_from_file(schema_file)
if isinstance(schemas, list):
if schema_version is None:
return self._find_highest_schema_version(schemas), schema_file
for entry in schemas:
if entry.get("schema_version") == schema_version:
return entry, schema_file
else:
return schemas, schema_file
raise ValueError(f"Schema version {schema_version} not found for {parameter_name}")
def _find_highest_schema_version(self, schema_list):
"""
Find entry with highest schema_version in a list of schema dicts.
Parameters
----------
schema_list: list
List of schema dictionaries.
Returns
-------
dict
Schema dictionary with highest schema_version.
"""
try:
valid_entries = [entry for entry in schema_list if "schema_version" in entry]
except TypeError as exc:
raise TypeError("No valid schema versions found in the list.") from exc
return max(valid_entries, key=lambda e: packaging.version.Version(e["schema_version"]))
[docs]
def get_parameter_type_for_schema(self, parameter_name, model_parameter_schema_version=None):
"""
Return parameter type(s) from a selected model-parameter schema.
This method loads the schema for ``parameter_name`` and an optional
``model_parameter_schema_version``, then extracts the ``type`` field from
each data entry. If all entries share the same type, a single string is
returned; otherwise a list of types is returned.
Parameters
----------
parameter_name: str
Name of the model parameter.
model_parameter_schema_version: str or None
Explicit model-parameter schema version to use. If None, the newest
available schema version is selected.
Returns
-------
str or list[str]
Parameter type, reduced to a single value when all schema entries
have the same type.
"""
schema_dict, _ = self._read_schema_dict(parameter_name, model_parameter_schema_version)
parameter_types = [data["type"] for data in schema_dict["data"]]
return (
parameter_types[0]
if all(data_type == parameter_types[0] for data_type in parameter_types)
else parameter_types
)
def _get_parameter_type(self):
"""
Return parameter type(s) from the currently loaded schema.
This helper reads ``self.schema_dict`` (expected to be populated by
``_read_schema_dict`` beforehand), extracts ``type`` values from its
data entries, and reduces the result to a single string when all types
are identical.
Returns
-------
str or list[str]
Parameter type derived from ``self.schema_dict``.
"""
_parameter_type = [data["type"] for data in self.schema_dict["data"]]
return (
_parameter_type[0]
if all(t == _parameter_type[0] for t in _parameter_type)
else _parameter_type
)
def _parameter_is_a_file(self):
"""
Check if parameter is a file.
Returns
-------
bool
True if parameter is a file.
"""
try:
return self.schema_dict["data"][0]["type"] == "file"
except KeyError, IndexError:
pass
return False
[docs]
def write_data(self, product_data=None, metadata=None, metadata_output_file=None):
"""
Write model data and metadata.
Parameters
----------
product_data: astropy Table
Model data to be written
metadata: MetadataCollector object
Metadata to be written.
metadata_output_file: str or Path, optional
Separate output path for application metadata. If omitted, metadata is
written next to the product data.
Raises
------
FileNotFoundError
if data writing was not successful.
"""
if product_data is None:
return
if metadata is not None:
product_data.meta.update(
value_conversion.change_dict_keys_case(metadata.get_top_level_metadata(), True)
)
self._logger.info(f"Writing data to {self.output_file}")
if isinstance(product_data, dict) and Path(self.output_file).suffix == ".json":
self.write_model_parameter_dict_json(self.output_file, product_data)
return
try:
product_data.write(self.output_file, format=self.output_file_format, overwrite=True)
except IORegistryError:
self._logger.error(f"Error writing model data to {self.output_file}.")
raise
if metadata is not None:
metadata.write(
metadata_output_file or self.output_file,
add_activity_name=True,
)
[docs]
def write_model_parameter_dict_json(self, file_name, data_dict):
"""
Write dictionary to model-parameter-style json file.
Parameters
----------
file_name : str
Name of output file.
data_dict : dict
Data dictionary.
Raises
------
FileNotFoundError
if data writing was not successful.
Returns
-------
Path
Resolved output file path.
"""
data_dict = ModelDataWriter.prepare_data_dict_for_writing(data_dict)
output_file = self.io_handler.get_output_file(
file_name, output_path_label=self.output_label
)
self._logger.info(f"Writing data to {output_file}")
ModelDataWriter.write_model_parameter_json(data_dict, output_file)
return output_file
[docs]
@staticmethod
def write_model_parameter_json(data_dict, output_file):
"""
Write model parameter dictionary to JSON file.
Centralizes the JSON serialization options (sort_keys, numpy_types,
compact_numeric_lists) so all callers use consistent settings.
Parameters
----------
data_dict : dict
Data dictionary to write.
output_file : str or Path
Path of the output JSON file.
"""
ascii_handler.write_data_to_file(
data=data_dict,
output_file=output_file,
sort_keys=True,
numpy_types=True,
compact_numeric_lists=isinstance(data_dict.get("value"), dict),
)
[docs]
@staticmethod
def prepare_data_dict_for_writing(data_dict):
"""
Prepare data dictionary for writing to json file.
Ensure sim_telarray style lists as strings 'type' and 'unit' entries.
Replace "None" with "null" for unit field.
Replace list of equal units with single unit string.
Parameters
----------
data_dict: dict
Dictionary with lists.
Returns
-------
dict
Dictionary with lists converted to strings.
"""
try:
data_dict["unit"] = value_conversion.normalize_dimensionless_unit(data_dict["unit"])
if isinstance(data_dict["unit"], str):
data_dict["unit"] = data_dict["unit"].replace("None", "null")
elif isinstance(data_dict["unit"], list):
data_dict["unit"] = [
unit.replace("None", "null") if isinstance(unit, str) else unit
for unit in data_dict["unit"]
]
if not data_dict["unit"]:
data_dict["unit"] = None
elif all(u == data_dict["unit"][0] for u in data_dict["unit"]):
data_dict["unit"] = data_dict["unit"][0]
except KeyError:
pass
return data_dict
@staticmethod
def _derive_data_format(product_data_format, output_file=None):
"""
Derive data format and ensure conformance with astropy data format naming.
If product_data_format is None and output_file is given, derive format
from output_file suffix.
Parameters
----------
product_data_format: string
format identifier
"""
if product_data_format is None and output_file is not None:
product_data_format = Path(output_file).suffix.lstrip(".")
return "ascii.ecsv" if product_data_format == "ecsv" else product_data_format
