Data Model#

Modules in the data_model sections provide functionality for reading, writing, and validation of data. Data products ingested or produced by simtools generally follows the CTAO data model.

File-backed model parameters use validated Astropy ECSV assets as their canonical tabular representation. Structured JSON records are reserved for non-tabular values such as mirror segmentation; sim_telarray text is generated only by the simulator adapters.

data_reader#

Helper module for reading of standardized simtools data products.

data_model.data_reader.read_table_from_file(file_name, schema_file=None, validate=False, metadata_file=None, context=None)[source]#

Read astropy table from file and validate against schema.

Metadata is read from metadata file or from the metadata section of the data file. Schema for validation can be given as argument, or is determined from the metadata associated to the file.

Parameters:

file_name: str or Path: Name of file to be read.
schema_file: str or Path: Name of schema file to be used for validation.
validate: bool: Validate data against schema (if true).
metadata_file: str or Path: Name of metadata file to be read.

Returns:

astropy Table: Table read from file.

Raises:

FileNotFoundError: If file does not exist.

format_checkers#

Custom format checkers for jsonschema validation.

data_model.format_checkers.check_array_element(element)[source]#: Validate array elements for jsonschema.

data_model.format_checkers.check_array_triggers_name(name)[source]#: Validate array trigger names for jsonschema.

data_model.format_checkers.check_astropy_unit(unit_string)[source]#: Validate astropy units (including dimensionless) for jsonschema.

data_model.format_checkers.check_astropy_unit_of_length(unit_string)[source]#: Validate astropy units that this is an astropy unit of length.

data_model.format_checkers.check_astropy_unit_of_time(unit_string)[source]#: Validate astropy units that this is an astropy unit of time.

data_model.format_checkers.check_common_particle_name(name)[source]#: Validate common particle names for jsonschema.

metadata_collector#

Metadata collector for simtools.

This should be the only module in simtools with knowledge on the implementation of the observatory metadata model.

class data_model.metadata_collector.MetadataCollector(args_dict=None, metadata_file_name=None, model_parameter_name=None, observatory='cta', clean_meta=True, schema_version='latest', context=None)[source]#

Collects metadata to describe the current simtools activity and its data products.

Parameters:

args_dict: dict: Command line parameters.
metadata_file_name: str: Name of metadata file (only required when args_dict is None).
model_parameter_name: str: Name of model parameter.
observatory: str: Name of observatory (default: “cta”).
clean_meta: bool: Clean metadata from None values and empty lists (default: True).
schema_version: str: Version of the metadata schema to use (default: ‘latest’).
context: RunContext, optional: Run context providing the activity.

clean_meta_data(meta_dict)[source]#: Remove None values and empty lists/dicts from a metadata dictionary.

collect_meta_data()[source]#: Collect and verify product metadata for each main-level metadata type.

static dump(args_dict, output_file, add_activity_name=False, context=None)[source]#: Write metadata of a new collector to file (see write).

get_data_model_schema_dict()[source]#: Return data model schema dictionary (empty if no valid schema file is given).

get_data_model_schema_file_name()[source]#: Return data model schema file name (command line, metadata, model name, input).

get_site(from_input_meta=False)[source]#: Return the site name from collected (or, if requested, first input) metadata.

get_top_level_metadata()[source]#: Return top level metadata dictionary (with updated activity end time).

write(yml_file=None, keys_lower_case=True, add_activity_name=False)[source]#: Write top level metadata to a yaml file and return its name.

metadata_model#

Definition of metadata model for input to and output of simtools.

Follows CTAO top-level data model definition.

data products submitted to SimPipe (‘input’)
data products generated by SimPipe (‘output’)

data_model.metadata_model.get_default_metadata_dict(schema_file=None, observatory='CTA', schema_version='latest', lower_case=True)[source]#

Return metadata schema with default values.

Follows the CTA Top-Level Data Model.

Parameters:

schema_file: str: Schema file (jsonschema format) used for validation
observatory: str: Observatory name
schema_version: str, optional: Version of the schema to use. If not provided, the latest version is used.
lower_case: bool, optional: If True, all keys in the returned dictionary will be converted to lower case.

Returns:

dict: Reference schema dictionary.

model_data_writer#

Model data writer module.

class data_model.model_data_writer.ModelDataWriter(output_file=None, output_file_format=None, output_path=None, model_reader=None, *, context=None)[source]#

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.

check_for_existing_parameter(parameter_name, instrument, parameter_version)[source]#

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.

get_parameter_type_for_schema(parameter_name, model_parameter_schema_version=None)[source]#

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.

get_validated_parameter_dict(parameter_name, value, instrument, parameter_version, unique_id=None, schema_version=None, unit=None, model_parameter_schema_version=None)[source]#

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.

static prepare_data_dict_for_writing(data_dict)[source]#

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.

validate_and_transform(product_data_table=None, product_data_dict=None, validate_schema_file=None, is_model_parameter=False)[source]#

Validate product data using jsonschema given in metadata.

If necessary, transform product data to match schema.

Parameters:

product_data_table: astropy Table: Model data to be validated.
product_data_dict: dict: Model data to be validated.
validate_schema_file: str: Schema file used in validation of output data.
is_model_parameter: bool: True if data describes a model parameter.

write_data(product_data=None, metadata=None, metadata_output_file=None)[source]#

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.

static 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)[source]#

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.

write_model_parameter_dict_json(file_name, data_dict)[source]#

Write dictionary to model-parameter-style json file.

Parameters:

file_namestr: Name of output file.
data_dictdict: Data dictionary.

Returns:

Path: Resolved output file path.

Raises:

FileNotFoundError: if data writing was not successful.

static write_model_parameter_json(data_dict, output_file)[source]#

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_dictdict: Data dictionary to write.
output_filestr or Path: Path of the output JSON file.

static 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)[source]#

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.

workflow_metadata#

Utilities for workflow-level metadata propagation into model-parameter metadata files.

data_model.workflow_metadata.build_workflow_activity_metadata(args_dict, workflow_activity_id, workflow_start, workflow_end, runtime_environment, workflow_context, context=None)[source]#

Build workflow activity metadata from workflow execution context.

Parameters:

args_dictdict: Workflow application arguments.
workflow_activity_idstr: Workflow-level activity identifier.
workflow_startdatetime: Start time of the workflow.
workflow_enddatetime: End time of the workflow.
runtime_environmentdict or None: Runtime environment definition used for the workflow.
workflow_contextdict: Context with keys ‘site’ and ‘instrument’ for the workflow.

Returns:

dict: Activity block to be injected into model-parameter metadata files.

data_model.workflow_metadata.update_model_parameter_metadata_file(metadata_file, workflow_activity, associated_activities)[source]#

Inject workflow metadata into a model-parameter metadata file.

Parameters:

metadata_filestr or Path: Path to the model-parameter metadata file to update.
workflow_activitydict: Workflow activity metadata block to set as top-level activity metadata.
associated_activitieslist: Ordered activities associated with workflow execution.

Returns:

None: Function updates file in place when it exists.

setting_workflow#

Create and execute versioned workflows for simple model-parameter settings.

data_model.setting_workflow.create_setting_workflow(args, model_reader=None, context=None)[source]#

Create validated input files, or reuse an identical setting.

Parameters:

argsdict: Parameter, instrument, value, parameter_version, description, and optional site, source_url, workflow_runtime_file, output_path (repository root), model_parameter_schema_version, and standard contact arguments.
model_readerobject, optional: Reader used to check for an already submitted parameter version.

Returns:

pathlib.Path: Path to the prepared workflow configuration.

Raises:

ValueError: If inputs are invalid or this version has conflicting input files.

data_model.setting_workflow.run_setting_workflow(config_file, args, context=None)[source]#

Execute a prepared setting in its activity output directory.

Parameters:

config_filepathlib.Path: Configuration returned by create_setting_workflow.
argsdict: Runner options, including model-source options and ignore_runtime_environment.

Returns:

pathlib.Path: Output directory used by this execution.

schema#

Module providing functionality to read and validate dictionaries using schema.

data_model.schema.get_model_parameter_schema(parameter, schema_version=None)[source]#

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.

data_model.schema.get_model_parameter_schema_file(parameter)[source]#

Return schema file path for a given model parameter.

Parameters:

parameter: str: Model parameter name.

Returns:

Path: Schema file path.

data_model.schema.get_model_parameter_schema_files(schema_directory=PosixPath('/home/runner/work/simtools/simtools/src/simtools/schemas/model_parameters'))[source]#

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.

data_model.schema.get_model_parameter_schema_version(schema_version=None)[source]#

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.

data_model.schema.get_parameter_attribute_from_schema(par_name, schema_version, attribute_name)[source]#

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.

data_model.schema.get_schema_file_from_file_metadata(file_name, observatory='cta')[source]#

Extract schema file name from a metadata or data file.

Parameters:

file_namestr or Path: File name to extract schema information from.
observatorystr: Observatory name (default: “cta”).

Returns:

str or None: Schema file name or None if not found.

data_model.schema.get_schema_version_from_data(data, observatory='cta')[source]#

Get schema version from data dictionary.

Parameters:

data: dict: data dictionary.

Returns:

str: Schema version. If not found, returns ‘latest’.

data_model.schema.load_schema(schema_file=None, schema_version='latest')[source]#

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

data_model.schema.validate_deprecation_and_version(data, software_name=None, ignore_software_version=False)[source]#

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.

data_model.schema.validate_dict_using_schema(data, schema_file=None, json_schema=None, ignore_software_version=False, offline=False)[source]#

Validate a data dictionary against a schema.

Parameters:

datadict: Dictionary to be validated.
schema_filestr or Path, optional: Schema file used for validation when json_schema is not provided.
json_schemadict, optional: Preloaded JSON schema used for validation.
ignore_software_versionbool, optional: If True, ignore software version check.
offlinebool, 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.

data_model.schema.validate_schema_from_files(file_directory, file_name=None, schema_file=None, ignore_software_version=False)[source]#

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_directorystr or Path, optional: Directory with files to be validated.
file_namestr or Path, optional: File name pattern to be validated.
schema_filestr, optional: Schema file name provided directly.
ignore_software_versionbool: If True, ignore software version check.

schema_loader#

Cached, dependency-neutral loading of schema definitions.

data_model.schema_loader.get_model_parameter_schema_files(schema_directory)[source]#

Return model-parameter names and schema files from a directory.

Parameters:

schema_directorystr or Path: Directory containing model-parameter schema files.

Returns:

list: Model-parameter names.
list: Corresponding schema file paths.

Raises:

FileNotFoundError: If the directory contains no model-parameter schema files.

data_model.schema_loader.get_schema_for_version(schema, schema_file, schema_version)[source]#

Return a requested version from a parsed schema definition.

Parameters:

schemadict or list: Parsed schema definition or versioned schema documents.
schema_filestr or Path: Schema source used in error and warning messages.
schema_versionstr: Requested schema version, or latest for the first document.

Returns:

dict: Selected schema definition.

Raises:

ValueError: If no version is requested, the schema list is empty, or the requested version is absent.

data_model.schema_loader.load_schema(schema_file, schema_version='latest')[source]#

Load a schema definition from a cached parsed source.

Parameters:

schema_filestr or Path: Local path, URL, or name relative to the packaged schemas directory (for example, model_parameters/num_gains.schema.yml). Local files take precedence over packaged schemas. Absolute paths are used directly.
schema_versionstr: Schema version to return, or latest for the first document.

Returns:

dict: Independent schema definition selected from the cached parsed source.

Raises:

FileNotFoundError: If the schema cannot be loaded locally or from its URL.
ValueError: If the requested schema version is unavailable.

parameter_catalog#

Access model parameter schemas and their catalog metadata.

data_model.parameter_catalog.collection_to_instrument_class_key(collection_name='telescopes')[source]#: Return instrument classes for a model repository collection.

data_model.parameter_catalog.get_collection_name_from_parameter_name(parameter_name)[source]#: Get the model repository collection for a model parameter.

data_model.parameter_catalog.get_model_parameter_scope(collection_name, instrument, parameter_name)[source]#: Return the instrument scope used to store a model parameter.

data_model.parameter_catalog.get_sim_telarray_parameter_scopes(array_element_name, design_model=None, ignore_missing_design_model=False)[source]#: Return the global, design, and telescope scopes for sim_telarray.

data_model.parameter_catalog.get_simulation_software_meta_parameter_mode(parameter_name, software_name='sim_telarray')[source]#: Return the metadata serialization mode for a software mapping.

data_model.parameter_catalog.get_simulation_software_name_from_parameter_name(parameter_name, software_name='sim_telarray')[source]#: Get the parameter name used by a simulation software, if mapped.

data_model.parameter_catalog.instrument_class_key_to_collection(class_name)[source]#: Convert instrument class key to collection name.

data_model.parameter_catalog.is_global_sim_telarray_parameter(parameter_name)[source]#: Return whether a sim_telarray parameter has global scope.

data_model.parameter_catalog.model_parameters(class_key_list=None)[source]#: Get model parameter definitions, optionally filtered by class key.

data_model.parameter_catalog.simulation_software()[source]#: Get simulation software names from the meta schema definition.

data_model.parameter_catalog.site_parameters()[source]#: Return site model parameters.

validate_data#

Validation of data using schema (types, ranges, allowed values and required columns).

class data_model.validate_data.DataValidator(schema_file=None, data_file=None, data_table=None, data_dict=None, check_exact_data_type=True)[source]#

Validate data for type and units following a describing schema; converts or transform data.

Data can be of table or dict format (internally, all data is converted to astropy tables).

Parameters:

schema_file: Path: Schema file describing input data and transformations.
data_file: Path: Input data file.
data_table: astropy.table: Input data table.
data_dict: dict: Input data dict.
check_exact_data_type: bool: Check for exact data type (default: True).

validate_and_transform(is_model_parameter=False, lists_as_strings=False)[source]#

Validate data (and data file) and return the validated dict or table.

Parameters:

is_model_parameter: bool: This is a model parameter (add some data preparation).
lists_as_strings: bool: Convert lists to strings (as needed for model parameters).

Raises:

TypeError: if no data or data table is available.

validate_data_file(is_model_parameter=None)[source]#: Read the data file (successful reading counts as file validation).

static validate_data_files(file_directory=None, file_name=None, is_model_parameter=True, check_exact_data_type=False, schema_file=None)[source]#

Validate data or model parameters in files in a directory or a single file.

Parameters:

file_directory: str or Path: Directory with files to be validated (JSON files, searched recursively).
file_name: str or Path: Name of the file to be validated.
is_model_parameter: bool: This is a model parameter (add some data preparation).
check_exact_data_type: bool: Require exact data type for validation.
schema_file: str or Path: Schema file used for all files (default: from parameter name or file metadata).

static validate_model_parameter(par_dict, par_name=None)[source]#

Validate a simulation model parameter.

Parameters:

par_dict: dict: Data dictionary.
par_name: str: Parameter name (default: ‘parameter’ entry of par_dict).

Returns:

dict: Validated data dictionary.

validate_parameter_and_file_name()[source]#: Check that the file name starts with the parameter and ends with its version.

table_asset#

Read and validate ECSV assets referenced by model parameters.

data_model.table_asset.get_simtel_serialization(schema_dict)[source]#

Return the sim_telarray table serialization contract from a schema.

Parameters:

schema_dictdict: Complete model-parameter schema.

Returns:

dict: The declared serialization contract.

Raises:

ValueError: If the parameter has a sim_telarray table format without a contract.

data_model.table_asset.read_ecsv_asset(path, schema_entry=None, parameter_data=None)[source]#

Read and validate one ECSV model asset.

Parameters:

pathstr or pathlib.Path: ECSV asset path.
schema_entrydict, optional: data entry from the model-parameter schema. Its table_columns are validated when present.
parameter_datadict, optional: JSON parameter record used to validate identity metadata.

Returns:

astropy.table.QTable: Validated table.

data_model.table_asset.resolve_asset_path(value, parameter_file)[source]#

Resolve a model-parameter asset path.

Assets are relative to their parameter JSON file. Absolute paths and path traversal are rejected.

data_model.table_asset.validate_table_asset(table, schema_entry=None, parameter_data=None)[source]#: Validate ECSV columns, units, ranges, and identity metadata in place.

json_validation#

Validation helpers for JSON-compatible structured model values.

data_model.json_validation.validate_finite_json_values(value, path='$')[source]#

Reject non-finite JSON numbers and identify their nested path.

Parameters:

valueobject: JSON-like value to inspect.
pathstr, optional: JSON path used in an error message.

Raises:

ValueError: If a real number is NaN or infinite.

mirror_segmentation#

Validation and serialization of mirror-segmentation model parameters.

data_model.mirror_segmentation.kind_required_fields(parameter_name, schema_version)[source]#: Map schema-declared segmentation kinds to their required fields.

data_model.mirror_segmentation.make_quantity(value, unit)[source]#

Return a JSON-compatible explicit quantity object.

Parameters:

valuefloat: Numeric quantity value.
unitstr: Unit associated with value.

Returns:

dict: Quantity represented by value and unit keys.

data_model.mirror_segmentation.quantity_value(record, field, unit, default=None)[source]#

Return one segmentation quantity converted to the requested unit.

Parameters:

recorddict: Segmentation record containing an explicit value/unit object.
fieldstr: Quantity field name.
unitstr: Unit to which the value is converted.
defaultfloat, optional: Value returned when the optional field is absent.

Returns:

float: Numeric value in unit.

data_model.mirror_segmentation.validate_segments(records, parameter_name, schema_version)[source]#

Validate mirror-segmentation records and return them unchanged.

Parameters:

recordslist of dict: Ring, shape, or polygon records.
parameter_namestr: Mirror-segmentation model parameter whose schema validates the records.
schema_versionstr: Version of the selected parameter schema.

Returns:

list of dict: The validated records.

Raises:

ValueError: If a record is malformed or contains non-finite geometry.