"""Validation class for ChemKED schema."""
import re
from importlib import resources
from warnings import warn
import habanero
import httpx2 as httpx
import numpy as np
import pint
import yaml
from cerberus import SchemaError, Validator
from . import schemas
from .orcid import search_orcid
units = pint.UnitRegistry()
"""Unit registry to contain the units used in PyKED"""
units.define("cm3 = centimeter**3")
units.define("m3 = meter**3")
units.define("mm3 = millimeter**3")
units.define("Torr = 133.322368 pascal")
units.define("m2 = meter**2")
units.define("cm6 = centimeter**6")
units.define("molecule = 1 / 6.02214076e23 mol")
def _normalize_unit_str(val_str):
"""Normalize implicit multiplication and negative exponents for Pint."""
val_str = str(val_str)
parts = val_str.split(" ", 1)
if len(parts) == 1:
return val_str
num, unit_str = parts
unit_str = re.sub(r"(?<=\w)_(?=\w)", " ", unit_str)
unit_str = re.sub(r"([a-zA-Z]+)(-\d+)", r"\1**\2", unit_str)
unit_str = re.sub(r"(?<=\w) +(?=\w)", " * ", unit_str)
return f"{num} {unit_str}"
Q_ = units.Quantity
crossref_api = habanero.Crossref(mailto="prometheus@pr.omethe.us")
# Load the ChemKED schema definition file
schema_file = resources.files(schemas) / "chemked_schema.yaml"
with schema_file.open("r", encoding="utf-8") as f:
schema_list = f.readlines()
inc_start = None
inc_end = None
inc_list = []
no_includes = False
for l_num, line in enumerate(schema_list):
if line.startswith("!include"):
if no_includes: # pragma: no cover
raise SchemaError("All included files must be first in the main schema")
if inc_start is None:
inc_start = l_num
if inc_end is not None: # pragma: no cover
raise SchemaError("All included files must be first in the main schema")
inc_fname = resources.files(schemas) / line.split("!include")[1].strip()
with inc_fname.open("r", encoding="utf-8") as f:
inc_list.extend(f.readlines())
else:
if not line.strip() or line.startswith("#") or line.startswith("---"):
continue
if inc_start is None: # pragma: no cover
no_includes = True
if inc_start is not None and inc_end is None:
inc_end = l_num
schema_list[inc_start:inc_end] = inc_list
schema = yaml.safe_load("".join(schema_list))
# These top-level keys in the schema serve as references for lower-level keys.
# They are removed to prevent conflicts due to required variables, etc.
for key in [
"author",
"value-unit-required",
"value-unit-optional",
"composition",
"ignition-type",
"uncertainty-metadata",
"value-with-uncertainty",
"value-without-uncertainty",
"time-shift",
"laminar-burning-velocity-measurement-schema",
"speciation-measurement-schema",
"ignition-delay-schema",
"time-history",
]:
if key in schema:
del schema[key]
# Relative tolerance on the sum of a mole or mass fraction composition. Compositions in the
# literature are reported to a fixed number of digits, so their sums land near 1.0 rather than on
# it, and a tolerance tighter than the reported precision rejects otherwise sound data.
composition_sum_tolerance = 1.0e-3
# The quantity that identifies each kind of datapoint, and the apparatus each can be measured on.
# The datapoints schema is an anyof, so without these an experiment-type could be paired with
# datapoints of another kind, or with an apparatus that cannot make that measurement.
experiment_datapoint_keys = {
"ignition delay": "ignition-delay",
"laminar burning velocity measurement": "laminar-burning-velocity",
"speciation measurement": "concentration-profiles",
}
experiment_apparatus_kinds = {
"ignition delay": [
"shock tube",
"rapid compression machine",
],
"laminar burning velocity measurement": [
"counterflow twin flame",
"heat flux burner",
"bunsen burner",
"outwardly propagating spherical flame",
],
"speciation measurement": [
"jet stirred reactor",
"flow reactor",
"burner stabilized flame",
"shock tube",
],
}
# SI units for available value-type properties
property_units = {
"temperature": "kelvin",
"compressed-temperature": "kelvin",
"pressure": "pascal",
"compressed-pressure": "pascal",
"ignition-delay": "second",
"first-stage-ignition-delay": "second",
"pressure-rise": "1.0 / second",
"compression-time": "second",
"volume": "meter**3",
"time": "second",
"piston position": "meter",
"emission": "dimensionless",
"absorption": "dimensionless",
"concentration": "mole/meter**3",
"stroke": "meter",
"clearance": "meter",
"compression-ratio": "dimensionless",
"equivalence-ratio": "dimensionless",
"laminar-burning-velocity": "meter / second",
"distance": "meter",
"flow-rate": "kilogram / meter**2 / second",
"residence-time": "second",
"reactor-volume": "meter**3",
"volumetric-flow-in-reference-state": "meter**3 / second",
"environment-temperature": "kelvin",
"global-heat-exchange-coefficient": "watt / meter**2 / kelvin",
"exchange-area": "meter**2",
"reactor-length": "meter",
"reactor-diameter": "meter",
"pressure-in-reference-state": "pascal",
"temperature-in-reference-state": "kelvin",
}
[docs]
def compare_name(given_name, family_name, question_name):
"""Compares a name in question to a specified name separated into given and family.
The name in question ``question_name`` can be of varying format, including
"Kyle E. Niemeyer", "Kyle Niemeyer", "K. E. Niemeyer", "KE Niemeyer", and
"K Niemeyer". Other possibilities include names with hyphens such as
"Chih-Jen Sung", "C. J. Sung", "C-J Sung".
Examples:
>>> compare_name('Kyle', 'Niemeyer', 'Kyle E Niemeyer')
True
>>> compare_name('Chih-Jen', 'Sung', 'C-J Sung')
True
Args:
given_name (`str`): Given (or first) name to be checked against.
family_name (`str`): Family (or last) name to be checked against.
question_name (`str`): The whole name in question.
Returns:
`bool`: The return value. True for successful comparison, False otherwise.
"""
# lowercase everything
given_name = given_name.lower()
family_name = family_name.lower()
question_name = question_name.lower()
# rearrange names given as "last, first middle"
if "," in question_name:
name_split = question_name.split(",")
name_split.reverse()
question_name = " ".join(name_split).strip()
# remove periods
question_name = question_name.replace(".", "")
given_name = given_name.replace(".", "")
family_name = family_name.replace(".", "")
# split names by , <space> - .
given_name = list(filter(None, re.split(r"[, \-.]+", given_name)))
num_family_names = len(list(filter(None, re.split("[, .]+", family_name))))
# split name in question by , <space> - .
name_split = list(filter(None, re.split(r"[, \-.]+", question_name)))
first_name = [name_split[0]]
if len(name_split) > 2:
first_name += list(name_split[1:-num_family_names])
if len(first_name) > 1 and len(given_name) == len(first_name):
# both have same number of first and middle names/initials
for i in range(1, len(first_name)):
first_name[i] = first_name[i][0]
given_name[i] = given_name[i][0]
elif len(given_name) != len(first_name):
min_names = min(len(given_name), len(first_name))
first_name = first_name[:min_names]
given_name = given_name[:min_names]
# first initial
if len(first_name[0]) == 1 or len(given_name[0]) == 1:
given_name[0] = given_name[0][0]
first_name[0] = first_name[0][0]
# first and middle initials combined
if len(first_name[0]) > 1 or len(given_name[0]) > 1:
given_name[0] = given_name[0][0]
first_name[0] = name_split[0][0]
# Hyphenated last name may need to be reconnected
if num_family_names == 1 and "-" in family_name:
num_hyphen = family_name.count("-")
family_name_compare = "-".join(name_split[-(num_hyphen + 1) :])
else:
family_name_compare = " ".join(name_split[-num_family_names:])
return given_name == first_name and family_name == family_name_compare
[docs]
class OurValidator(Validator):
"""Custom validator with rules for Quantities and references."""
def _validate_isvalid_experiment(self, isvalid_experiment, field, value):
"""Checks that the datapoints and the apparatus match the stated experiment type.
The datapoints schema accepts any of the three kinds of datapoint, so on its own it cannot
tell that, say, ignition delay datapoints were filed under a speciation measurement.
Args:
isvalid_experiment (`bool`): flag from schema indicating the check is to be done
field (`str`): 'datapoints'
value (`list`): the datapoints
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
experiment_type = self.document.get("experiment-type")
if experiment_type not in experiment_datapoint_keys:
# An unallowed experiment-type is reported by the allowed rule instead
return
required_key = experiment_datapoint_keys[experiment_type]
for idx, datapoint in enumerate(value):
if not isinstance(datapoint, dict):
continue
if required_key not in datapoint:
self._error(
field,
f"datapoint {idx} has no {required_key}, which every "
f"{experiment_type} datapoint requires",
)
apparatus = self.document.get("apparatus")
allowed_kinds = experiment_apparatus_kinds[experiment_type]
if isinstance(apparatus, dict) and apparatus.get("kind") not in [None, *allowed_kinds]:
self._error(
field,
f"a {experiment_type} cannot be measured in a {apparatus['kind']}; "
f"allowed kinds are {allowed_kinds}",
)
def _validate_isvalid_t_range(self, isvalid_t_range, field, values):
"""Checks that the temperature ranges given for thermo data are valid
Args:
isvalid_t_range (`bool`): flag from schema indicating T range is to be checked
field (`str`): T_range
values (`list`): List of temperature values indicating low, middle, and high ranges
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
if all(isinstance(v, (float, int)) for v in values):
# If no units given, assume Kelvin
T_low = Q_(values[0], "K")
T_mid = Q_(values[1], "K")
T_hi = Q_(values[2], "K")
elif all(isinstance(v, str) for v in values):
T_low = Q_(values[0])
T_mid = Q_(values[1])
T_hi = Q_(values[2])
else:
self._error(
field,
"The temperatures in the range must all be either with units or "
"without units, they cannot be mixed",
)
return False
if min([T_low, T_mid, T_hi]) != T_low:
self._error(field, "The first element of the T_range must be the lower limit")
if max([T_low, T_mid, T_hi]) != T_hi:
self._error(field, "The last element of the T_range must be the upper limit")
def _validate_isvalid_unit(self, isvalid_unit, field, value):
"""Checks for appropriate units using Pint unit registry.
Args:
isvalid_unit (`bool`): flag from schema indicating units to be checked.
field (`str`): property associated with units in question.
value (`dict`): dictionary of values from file associated with this property.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
quantity = 1.0 * units(value["units"])
try:
quantity.to(property_units[field])
except pint.DimensionalityError:
self._error(
field,
f"incompatible units; should be consistent with {property_units[field]}",
)
def _validate_isvalid_history(self, isvalid_history, field, value):
"""Checks that the given time history is properly formatted.
Args:
isvalid_history (`bool`): flag from schema indicating history to be checked.
field (`str`): property associated with history in question.
value (`dict`): dictionary of values from file associated with this property.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
# Check the type has appropriate units
history_type = value["type"]
if history_type.endswith("emission"):
history_type = "emission"
elif history_type.endswith("absorption"):
history_type = "absorption"
quantity = 1.0 * (units(value["quantity"]["units"]))
try:
quantity.to(property_units[history_type])
except pint.DimensionalityError:
self._error(
field,
f"incompatible units; should be consistent with {property_units[history_type]}",
)
# Check that time has appropriate units
time = 1.0 * (units(value["time"]["units"]))
try:
time.to(property_units["time"])
except pint.DimensionalityError:
self._error(
field,
f"incompatible units; should be consistent with {property_units['time']}",
)
# Check that the values have the right number of columns
n_cols = len(value["values"][0])
max_cols = (
max(
value["time"]["column"],
value["quantity"]["column"],
value.get("uncertainty", {}).get("column", 0),
)
+ 1
)
if n_cols > max_cols:
self._error(field, "too many columns in the values")
elif n_cols < max_cols:
self._error(field, "not enough columns in the values")
def _validate_isvalid_speciation(self, isvalid_speciation, field, value):
"""Checks that a speciation datapoint's profile rows are well formed.
Each concentration profile row must carry one value per declared
independent variable, followed by the measured amount, with an
optional trailing uncertainty.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
n_independent = len(value.get("independent-variables", []))
if n_independent < 1:
self._error(field, "at least one independent-variable is required")
return
for profile in value.get("concentration-profiles", []):
species = profile.get("species-name", "?")
for row in profile.get("values", []):
if len(row) not in (n_independent + 1, n_independent + 2):
self._error(
field,
"concentration-profiles row for "
f"{species} must have {n_independent + 1} or "
f"{n_independent + 2} columns ({n_independent} "
"independent-variable(s) + amount [+ uncertainty]); "
f"got {len(row)}",
)
def _validate_isvalid_quantity(self, isvalid_quantity, field, value):
"""Checks for valid given value and appropriate units.
Args:
isvalid_quantity (`bool`): flag from schema indicating quantity to be checked.
field (`str`): property associated with quantity in question.
value (`list`): list whose first element is a string representing a value with units
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
if isinstance(value[0], dict):
return
quantity = self._parse_quantity(field, value[0])
if quantity is None:
return
expected_units = property_units.get(field)
if expected_units is None:
# No dimensional check is configured for this property. Zero is meaningful for the
# properties that land here, a composition amount above all, since a species can be
# measured as absent, so only a negative value is an error. `isvalid_composition`
# takes the same view of an amount.
if quantity.magnitude < 0:
self._error(field, "value must not be negative")
return
low_lim = 0.0 * units(expected_units)
try:
if quantity <= low_lim:
self._error(
field,
f"value must be greater than 0.0 {expected_units}",
)
except pint.DimensionalityError:
self._error(
field,
f"incompatible units; should be consistent with {expected_units}",
)
def _parse_quantity(self, field, value):
"""Read a value into a Pint quantity, reporting an error if it cannot be read.
Args:
field (`str`): property the value belongs to, used in the error message
value (`str` or `float`): the value as written in the file
Returns:
`~pint.Quantity`, or `None` if the value could not be read
"""
try:
return Q_(_normalize_unit_str(value))
except Exception: # noqa: BLE001 - Pint raises several unrelated types for bad input
self._error(field, f"{value!r} is not a value with units that can be understood")
return None
def _check_uncertainty_value(self, field, key, value, kind):
"""Check one uncertainty value against the kind of uncertainty it declares.
A relative uncertainty is a fraction of the quantity it describes, so it carries no units.
An absolute one is in the units of that quantity, and is checked against them.
Args:
field (`str`): property the uncertainty describes
key (`str`): which uncertainty key is being checked, used in the error message
value (`str` or `float`): the uncertainty value
kind (`str`): ``relative`` or ``absolute``
"""
if kind != "relative":
self._validate_isvalid_quantity(True, field, [value])
return
quantity = self._parse_quantity(field, value)
if quantity is None:
return
if not quantity.dimensionless:
self._error(
field,
f"a relative {key} is a fraction of the {field} value, so it must be "
f"dimensionless; got {value!r}",
)
def _check_uncertainty_metadata(self, field, metadata):
"""Checks that uncertainty metadata includes a value, not only labels."""
uncertainty_value_keys = {
"uncertainty",
"upper-uncertainty",
"lower-uncertainty",
}
has_uncertainty_value = any(
metadata.get(key) is not None for key in uncertainty_value_keys
)
has_evaluated_sd_value = metadata.get("evaluated-standard-deviation") is not None
if has_uncertainty_value or has_evaluated_sd_value:
return True
self._error(
field,
"uncertainty metadata must contain at least one uncertainty value "
"(uncertainty, upper-uncertainty, lower-uncertainty) or an "
f"evaluated-standard-deviation value; got: {dict(metadata) or 'empty dict'}",
)
return False
def _check_uncertainty_metadata_values(self, field, uncertainty_dict):
"""Validate the values contained in an uncertainty metadata mapping."""
if not self._check_uncertainty_metadata(field, uncertainty_dict):
return
uncertainty_type = uncertainty_dict.get("uncertainty-type")
for key in ["uncertainty", "upper-uncertainty", "lower-uncertainty"]:
if uncertainty_dict.get(key) is not None:
self._check_uncertainty_value(
field, key, uncertainty_dict[key], uncertainty_type
)
if uncertainty_dict.get("evaluated-standard-deviation") is not None:
self._check_uncertainty_value(
field,
"evaluated-standard-deviation",
uncertainty_dict["evaluated-standard-deviation"],
uncertainty_dict.get("evaluated-standard-deviation-type"),
)
def _validate_isvalid_profile_uncertainty(
self, isvalid_profile_uncertainty, field, value
):
"""Validate uncertainty metadata associated with a concentration profile.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
if value and isinstance(value[0], dict):
self._check_uncertainty_metadata_values(field, value[0])
def _validate_isvalid_uncertainty(self, isvalid_uncertainty, field, value):
"""Checks for valid given value and appropriate units with uncertainty.
Args:
isvalid_uncertainty (`bool`): flag from schema indicating uncertainty to be checked
field (`str`): property associated with the quantity in question.
value (`list`): list with the string of the value of the quantity and a dictionary of
the uncertainty
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
self._validate_isvalid_quantity(True, field, value)
if len(value) > 1:
self._check_uncertainty_metadata_values(field, value[1])
def _validate_isvalid_reference(self, isvalid_reference, field, value):
"""Checks valid reference metadata using DOI (if present).
Args:
isvalid_reference (`bool`): flag from schema indicating reference to be checked.
field (`str`): 'reference'
value (`dict`): dictionary of reference metadata.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
if "doi" in value:
try:
ref = crossref_api.works(ids=value["doi"])["message"]
except (httpx.HTTPStatusError, habanero.RequestError):
self._error(field, "DOI not found")
return
except httpx.ConnectError:
warn("network not available, DOI not validated.")
return
# Assume that the reference returned by the DOI lookup always has a container-title
ref_container = ref.get("container-title")[0]
# TODO: Add other container types: value.get('journal') or value.get('report') or ...
# note that there's a type field in the ref that is journal-article, proceedings-article
container = value.get("journal")
if container is None or container != ref_container:
self._error(field, f"journal should be {ref_container}")
# Assume that the reference returned by DOI lookup always has a year
ref_year = ref.get("published-print") or ref.get("published-online")
ref_year = ref_year["date-parts"][0][0]
year = value.get("year")
if year is None or year != ref_year:
self._error(field, f"year should be {ref_year}")
# Volume number might not be in the reference
ref_volume = ref.get("volume")
volume = value.get("volume")
if ref_volume is None:
if volume is not None:
self._error(
field,
"Volume was specified in the YAML but is not present in the DOI reference.",
)
else:
if volume is None or int(volume) != int(ref_volume):
self._error(field, f"volume should be {ref_volume}")
# Pages might not be in the reference
ref_pages = ref.get("page")
pages = value.get("pages")
if ref_pages is None:
if pages is not None:
self._error(
field,
"Pages were specified in the YAML but are not present in "
"the DOI reference.",
)
else:
if pages is None or pages != ref_pages:
self._error(field, f"pages should be {ref_pages}")
# check that all authors present
authors = value["authors"][:]
author_names = [a["name"] for a in authors]
for author in ref["author"]:
# find using family name
author_match = next(
(
a
for a in authors
if compare_name(author["given"], author["family"], a["name"])
),
None,
)
# error if missing author in given reference information
if author_match is None:
self._error(
field,
f"Missing author: {author['given']} {author['family']}",
)
else:
author_names.remove(author_match["name"])
# validate ORCID if given
orcid = author.get("ORCID")
if orcid:
# Crossref may give ORCID as http://orcid.org/####-####-####-####
# so need to strip the leading URL
orcid = orcid[orcid.rfind("/") + 1 :]
if "ORCID" in author_match:
if author_match["ORCID"] != orcid:
self._error(
field,
f"{author_match['name']} ORCID does not match that in "
f"reference. Reference: {orcid}. "
f"Given: {author_match['ORCID']}",
)
else:
# ORCID not given, suggest adding it
warn(f"ORCID {orcid} missing for {author_match['name']}")
# check for extra names given
if len(author_names) > 0:
self._error(field, f"Extra author(s) given: {', '.join(author_names)}")
def _validate_isvalid_orcid(self, isvalid_orcid, field, value):
"""Checks for valid ORCID if given.
Args:
isvalid_orcid (`bool`): flag from schema indicating ORCID to be checked.
field (`str`): 'author'
value (`dict`): dictionary of author metadata.
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
if isvalid_orcid and "ORCID" in value:
try:
res = search_orcid(value["ORCID"])
except httpx.ConnectError:
warn("network not available, ORCID not validated.")
return
except httpx.HTTPStatusError:
self._error(field, f"ORCID incorrect or invalid for {value['name']}")
return
family_name = res["name"]["family-name"]["value"]
given_name = res["name"]["given-names"]["value"]
if not compare_name(given_name, family_name, value["name"]):
self._error(
field,
f"Name and ORCID do not match. Name supplied: {value['name']}. "
f"Name associated with ORCID: {given_name} {family_name}",
)
def _validate_isvalid_composition(self, isvalid_composition, field, value):
"""Checks for valid specification of composition.
Args:
isvalid_composition (bool): flag from schema indicating
composition to be checked.
field (str): 'composition'
value (dict): dictionary of composition
The rule's arguments are validated against this schema:
{'type': 'boolean'}
"""
composition_kind = value["kind"]
fraction_kinds = ["mass fraction", "mole fraction"]
percent_kinds = ["mole percent"]
concentration_kinds = ["mol/cm3", "mol/m3", "mol/L", "mol/dm3"]
sum_amount = 0.0
total_amount = None
if composition_kind in fraction_kinds:
low_lim = 0.0
up_lim = 1.0
total_amount = 1.0
elif composition_kind in percent_kinds:
low_lim = 0.0
up_lim = 100.0
total_amount = 100.0
elif composition_kind in concentration_kinds:
low_lim = 0.0
up_lim = None
else:
self._error(
field,
'composition kind must be "mole percent", "mass fraction", "mole fraction", '
'"mol/cm3", "mol/m3", "mol/L", or "mol/dm3"',
)
return False
for sp in value["species"]:
amount = sp["amount"][0]
sum_amount += amount
# Check that amount within bounds, based on kind specified
if amount < low_lim:
self._error(
field,
f"Species {sp['species-name']} {composition_kind} "
f"must be greater than {low_lim:.1f}",
)
elif up_lim is not None and amount > up_lim:
self._error(
field,
f"Species {sp['species-name']} {composition_kind} "
f"must be less than {up_lim:.1f}",
)
# Make sure mole/mass fraction sum to 1, allowing for the round-off in published tables
if total_amount is not None and not np.isclose(
total_amount, sum_amount, rtol=composition_sum_tolerance
):
self._error(
field,
f"Species {composition_kind}s do not sum to {total_amount:.1f}: {sum_amount:f}",
)
# TODO: validate InChI, SMILES, or atomic-composition
