"""Module providing functionality to read and validate dictionaries using schema."""
import logging
import urllib.error
import urllib.request
from functools import lru_cache
from pathlib import Path
import jsonschema
from astropy.table import Table
from referencing import Registry, Resource
from simtools import dependencies
from simtools.constants import (
METADATA_JSON_SCHEMA,
MODEL_PARAMETER_METASCHEMA,
MODEL_PARAMETER_SCHEMA_PATH,
SCHEMA_PATH,
)
from simtools.data_model import format_checkers, schema_loader
from simtools.io import ascii_handler
from simtools.utils import names, value_conversion
from simtools.version import check_version_constraint
_logger = logging.getLogger(__name__)
[docs]
def get_model_parameter_schema_files(schema_directory=MODEL_PARAMETER_SCHEMA_PATH):
"""
Return list of parameters and schema files located in schema file directory.
Returns
-------
list
List of parameters found in schema file directory.
list
List of schema files found in schema file directory.
"""
return schema_loader.get_model_parameter_schema_files(schema_directory)
[docs]
def get_model_parameter_schema_file(parameter):
"""
Return schema file path for a given model parameter.
Parameters
----------
parameter: str
Model parameter name.
Returns
-------
Path
Schema file path.
"""
schema_file = MODEL_PARAMETER_SCHEMA_PATH / f"{parameter}.schema.yml"
if not schema_file.exists():
raise FileNotFoundError(f"Schema file not found: {schema_file}")
return schema_file
[docs]
def get_model_parameter_schema(parameter, schema_version=None):
"""
Return schema dictionary for a given model parameter and schema version.
Parameters
----------
parameter: str
Model parameter name.
schema_version: str, optional
Schema version. If not provided, the latest version is used.
Returns
-------
dict
Independent schema dictionary.
"""
schema_file = get_model_parameter_schema_file(parameter)
return load_schema(schema_file, schema_version or "latest")
[docs]
def get_model_parameter_schema_version(schema_version=None):
"""
Validate and return schema versions.
If no schema_version is given, the most recent version is provided.
Parameters
----------
schema_version: str
Schema version.
Returns
-------
str
Schema version.
"""
schemas = ascii_handler.collect_data_from_file(MODEL_PARAMETER_METASCHEMA)
if schema_version is None and schemas:
return schemas[0].get("schema_version")
if any(schema.get("schema_version") == schema_version for schema in schemas):
return schema_version
raise ValueError(f"Schema version {schema_version} not found in {MODEL_PARAMETER_METASCHEMA}.")
[docs]
def validate_dict_using_schema(
data, schema_file=None, json_schema=None, ignore_software_version=False, offline=False
):
"""
Validate a data dictionary against a schema.
Parameters
----------
data : dict
Dictionary to be validated.
schema_file : str or Path, optional
Schema file used for validation when ``json_schema`` is not provided.
json_schema : dict, optional
Preloaded JSON schema used for validation.
ignore_software_version : bool, optional
If True, ignore software version check.
offline : bool, optional
If True, skip remote ``meta_schema_url`` validation.
Returns
-------
dict or None
The validated data, or ``None`` if no schema was provided.
Raises
------
jsonschema.exceptions.ValidationError
If validation fails.
"""
if json_schema is None and schema_file is None:
_logger.warning(f"No schema provided for validation of {data}")
return None
if json_schema is None:
schema_version = data.get("meta_schema_version") or get_schema_version_from_data(data)
json_schema = load_schema(schema_file, schema_version)
validate_deprecation_and_version(data, ignore_software_version=ignore_software_version)
validator = jsonschema.Draft6Validator(
schema=json_schema,
format_checker=format_checkers.format_checker,
registry=Registry(retrieve=_retrieve_yaml_schema_from_uri),
)
try:
validator.validate(instance=data)
except jsonschema.exceptions.ValidationError as exc:
_logger.error(f"Validation failed using schema: {json_schema} for data: {data}")
raise exc
if not offline:
_validate_meta_schema_url(data)
_logger.debug(f"Successful validation of data using schema ({json_schema.get('name')})")
return data
def _validate_meta_schema_url(data):
"""Validate meta_schema_url if present in data."""
if not isinstance(data, dict):
return
meta_schema_url = data.get("meta_schema_url")
if meta_schema_url is not None:
try:
with urllib.request.urlopen(meta_schema_url, timeout=5) as response:
url_exists = response.status == 200
except urllib.error.URLError, AttributeError:
url_exists = False
if not url_exists:
raise FileNotFoundError(f"Meta schema URL does not exist: {meta_schema_url}")
@lru_cache
def _retrieve_yaml_schema_from_uri(uri):
"""Load schema from a file URI."""
path = SCHEMA_PATH / Path(uri.removeprefix("file:/"))
contents = ascii_handler.collect_data_from_file(file_name=path)
return Resource.from_contents(contents)
[docs]
def get_schema_version_from_data(data, observatory="cta"):
"""
Get schema version from data dictionary.
Parameters
----------
data: dict
data dictionary.
Returns
-------
str
Schema version. If not found, returns 'latest'.
"""
schema_version = data.get("schema_version") or data.get("SCHEMA_VERSION")
if schema_version:
return schema_version
reference_version = data.get(observatory.upper(), {}).get("REFERENCE", {}).get(
"VERSION"
) or data.get(observatory.lower(), {}).get("reference", {}).get("version")
if reference_version:
return reference_version
return "latest"
[docs]
def load_schema(schema_file=None, schema_version="latest"):
"""
Load parameter schema from file.
Parameters
----------
schema_file: str
Path to schema file.
schema_version: str
Schema version.
Returns
-------
schema: dict
Independent schema dictionary.
Raises
------
FileNotFoundError
if schema file is not found
"""
schema_file = schema_file or METADATA_JSON_SCHEMA
schema = schema_loader.load_schema(schema_file, schema_version)
_add_array_elements("InstrumentTypeElement", schema)
return schema
def _get_array_element_list():
"""Build complete list of array elements including design types."""
elements = set(names.array_elements().keys())
for array_element in names.array_elements():
for design_type in names.array_element_design_types(array_element):
elements.add(f"{array_element}-{design_type}")
return sorted(elements)
def _add_array_elements(key, schema):
"""
Add list of array elements to schema.
Avoids having to list all array elements in multiple schema.
Assumes an element [key]['enum'] is a list of elements.
Parameters
----------
key: str
Key in schema dictionary
schema: dict
Schema dictionary
Returns
-------
dict
Schema dictionary with added array elements.
"""
array_elements = _get_array_element_list()
def update_enum(sub_schema):
if "enum" in sub_schema and isinstance(sub_schema["enum"], list):
sub_schema["enum"] = list(set(sub_schema["enum"] + array_elements))
else:
sub_schema["enum"] = array_elements
def recursive_search(sub_schema, target_key):
if target_key in sub_schema:
update_enum(sub_schema[target_key])
return
for v in sub_schema.values():
if isinstance(v, dict):
recursive_search(v, target_key)
recursive_search(schema, key)
return schema
[docs]
def validate_deprecation_and_version(data, software_name=None, ignore_software_version=False):
"""
Check if data contains deprecated parameters or version mismatches.
Parameters
----------
data: dict
Data dictionary to check.
software_name: str or None
Name of the software to check version against. If None, use complete list
ignore_software_version: bool
If True, ignore software version check.
"""
if not isinstance(data, dict):
return
data_name = data.get("name", "<unknown>")
if data.get("deprecated", False):
note = data.get("deprecation_note", "(no deprecation note provided)")
_logger.warning(f"Data for {data_name} is deprecated. Note: {note}")
for sw in data.get("simulation_software", []):
name, constraint = sw.get("name"), sw.get("version")
if not name or not constraint:
continue
if software_name is not None and name.lower() != software_name.lower():
continue
software_version = dependencies.get_software_version(name)
if check_version_constraint(software_version, constraint):
continue
msg = f"{data_name}: version {software_version} of {name} does not match {constraint}."
if ignore_software_version:
_logger.warning(f"{msg}, but version check is ignored.")
else:
raise ValueError(msg)
[docs]
def validate_schema_from_files(
file_directory, file_name=None, schema_file=None, ignore_software_version=False
):
"""
Validate a schema file or several files in a directory.
Files to be validated are taken from file_directory and file_name pattern.
The schema is either given as command line argument, read from the meta_schema_url or from
the metadata section of the data dictionary.
Parameters
----------
file_directory : str or Path, optional
Directory with files to be validated.
file_name : str or Path, optional
File name pattern to be validated.
schema_file : str, optional
Schema file name provided directly.
ignore_software_version : bool
If True, ignore software version check.
"""
if file_directory and file_name:
file_list = sorted(Path(file_directory).rglob(file_name))
else:
file_list = [Path(file_name)] if file_name else []
for _file_name in file_list:
try:
data = ascii_handler.collect_data_from_file(file_name=_file_name)
except FileNotFoundError as exc:
raise FileNotFoundError(f"Error reading schema file from {_file_name}") from exc
data = data if isinstance(data, list) else [data]
try:
for data_dict in data:
validate_dict_using_schema(
data_dict,
_get_schema_file_name(schema_file, _file_name, data_dict),
ignore_software_version=ignore_software_version,
)
except Exception as exc:
raise ValueError(f"Validation of file {_file_name} failed") from exc
_logger.info(f"Successful validation of file {_file_name}")
def _get_schema_file_name(schema_file=None, file_name=None, data_dict=None):
"""
Get schema file name from metadata, data dict, or from file.
Parameters
----------
schema_file : str, optional
Schema file name provided directly.
file_name : str or Path, optional
File name to extract schema information from.
data_dict : dict, optional
Dictionary with metaschema information.
Returns
-------
str or None
Schema file name.
"""
if schema_file is not None:
return schema_file
if data_dict:
# Check for meta_schema_url (metadata) or schema_url (data with embedded schema info)
if url := data_dict.get("meta_schema_url"):
return url
if url := data_dict.get("schema_url"):
return url
if file_name:
return get_schema_file_from_file_metadata(file_name)
return None
def _extract_schema_url_from_metadata_dict(metadata, observatory="cta"):
"""Extract schema URL from metadata dictionary."""
for key in (observatory, observatory.lower()):
url = metadata.get(key, {}).get("product", {}).get("data", {}).get("model", {}).get("url")
if url:
return url
return None
[docs]
def get_parameter_attribute_from_schema(par_name, schema_version, attribute_name):
"""
Return one attribute from model-parameter schema data entries.
Parameters
----------
par_name: str
Name of the parameter.
schema_version: str
Schema version to look up.
attribute_name: str
Attribute to read from data entries (e.g. "type", "unit").
Returns
-------
str or list or None
Attribute value as scalar for single-entry schemas, list for multi-entry schemas,
or None for unsupported schema data structures.
"""
schema_dict = get_model_parameter_schema(par_name, schema_version)
data = schema_dict.get("data", [])
if isinstance(data, list):
values = [
_normalize_parameter_schema_attribute(attribute_name, item.get(attribute_name))
for item in data
]
return values[0] if len(values) == 1 else values
if isinstance(data, dict):
return _normalize_parameter_schema_attribute(attribute_name, data.get(attribute_name))
return None
def _normalize_parameter_schema_attribute(attribute_name, value):
"""Normalize schema attribute values for public helper functions."""
if attribute_name == "unit":
return value_conversion.normalize_dimensionless_unit(value)
return value
