"""Schema compliance checking for HED schemas.
This module is designed for HED 8.3+ schemas, which carry full domain and
range metadata on their attributes. It can be run on any loaded schema, but
schemas earlier than 8.3 will produce extensive known compliance errors
because they lack the attribute metadata that 8.3 introduced.
The checker validates domain constraints, range constraints, and semantic
rules for all entries in a schema. The schema's own Attributes and
Properties sections define which attributes are valid for each section
(domain) and what type of value each attribute takes (range). This checker
is data-driven by that metadata rather than hard-coding parallel validator
dictionaries.
"""
import re
from collections import defaultdict
from functools import partial
import pandas as pd
from semantic_version import Version
from hed.errors.error_reporter import ErrorHandler, sort_issues
from hed.errors.error_types import (
ErrorContext,
ErrorSeverity,
SchemaAttributeErrors,
SchemaErrors,
SchemaWarnings,
)
from hed.errors.exceptions import HedFileError
from hed.schema import hed_cache
from hed.schema.hed_schema import HedKey, HedSchema, HedSectionKey
from hed.schema.hed_schema_constants import ANY_UNITS_CLASS
from hed.schema.schema_io import df_constants
from hed.schema.schema_validation import attribute_validators
from hed.schema.schema_validation.compliance_summary import ComplianceSummary
from hed.schema.schema_validation.hed_id_validator import HedIDValidator
from hed.schema.schema_validation.validation_util import (
get_allowed_characters_by_name,
get_problem_indexes,
schema_version_for_library,
validate_schema_description,
validate_schema_tag,
validate_schema_term,
)
# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------
[docs]
def check_compliance(hed_schema, check_for_warnings=True, name=None, error_handler=None):
"""Check a HED schema for compliance.
Parameters:
hed_schema (HedSchema): HedSchema object to check for hed3 compliance.
check_for_warnings (bool): If True, check for formatting issues like
invalid characters, capitalization, etc.
name (str): If present, will use as filename for context.
error_handler (ErrorHandler or None): Used to report errors.
Uses a default one if none passed in.
Returns:
list: A list of all warnings and errors found. Each issue is a dict.
The returned list has an additional ``compliance_summary``
attribute (ComplianceSummary) providing a structured report.
Raises:
ValueError: If *hed_schema* is not a ``HedSchema`` instance.
"""
if not isinstance(hed_schema, HedSchema):
raise ValueError("To check compliance of a HedGroupSchema, call self.check_compliance on the schema itself.")
error_handler = error_handler or ErrorHandler(check_for_warnings)
validator = SchemaValidator(hed_schema, error_handler)
issues = []
name = name or hed_schema.filename
error_handler.push_error_context(ErrorContext.FILE_NAME, name)
issues += validator.check_if_prerelease_version()
issues += validator.check_prologue_epilogue()
issues += validator.check_invalid_characters()
issues += validator.check_attributes()
issues += validator.check_duplicate_names()
issues += validator.check_library_partner_rules()
issues += validator.check_reserved_tags_file()
issues += validator.check_redundant_units()
issues += validator.check_any_units_class()
issues += validator.check_units_unique_across_classes()
issues += validator.check_duplicate_hed_ids()
issues += validator.check_extras_columns()
issues += validator.check_annotation_attribute_values()
error_handler.pop_error_context()
issues = sort_issues(issues)
return _IssuesListWithSummary(issues, validator.summary)
class _IssuesListWithSummary(list):
"""A list subclass that carries a ``compliance_summary`` attribute."""
def __init__(self, issues, summary):
super().__init__(issues)
self.compliance_summary = summary
# ---------------------------------------------------------------------------
# Constants — domain / range mappings
# ---------------------------------------------------------------------------
#: Map from domain property name to the HedSectionKey it gates.
DOMAIN_TO_SECTION = {
HedKey.TagDomain: HedSectionKey.Tags,
HedKey.UnitClassDomain: HedSectionKey.UnitClasses,
HedKey.UnitDomain: HedSectionKey.Units,
HedKey.UnitModifierDomain: HedSectionKey.UnitModifiers,
HedKey.ValueClassDomain: HedSectionKey.ValueClasses,
}
#: Reverse map — section to domain property name.
SECTION_TO_DOMAIN = {v: k for k, v in DOMAIN_TO_SECTION.items()}
#: The five primary content sections.
CONTENT_SECTIONS = frozenset(DOMAIN_TO_SECTION.values())
# ---------------------------------------------------------------------------
# Validator class
# ---------------------------------------------------------------------------
[docs]
class SchemaValidator:
"""Validates a loaded HedSchema for compliance.
The five content sections (Tags, UnitClasses, Units, UnitModifiers,
ValueClasses) are validated using range and domain metadata that the
schema itself provides in its Attributes and Properties sections.
Typical usage is through :func:`check_compliance`.
"""
# -- Range validators ---------------------------------------------------
# Each range property maps to zero or more validator functions with the
# signature: validator(hed_schema, tag_entry, attribute_name) -> list[dict]
_range_validators = {
HedKey.BoolRange: [], # bool attributes are present/absent — no value check
HedKey.TagRange: [
partial(attribute_validators.item_exists_check, section_key=HedSectionKey.Tags),
],
HedKey.NumericRange: [
attribute_validators.is_numeric_value,
],
HedKey.StringRange: [], # free-form string — no structural validation
HedKey.UnitClassRange: [
partial(attribute_validators.item_exists_check, section_key=HedSectionKey.UnitClasses),
],
HedKey.UnitRange: [
attribute_validators.unit_exists,
],
HedKey.ValueClassRange: [
partial(attribute_validators.item_exists_check, section_key=HedSectionKey.ValueClasses),
],
}
# -- Semantic validators ------------------------------------------------
# Extra checks for specific attributes, beyond what range covers.
_semantic_validators = {
HedKey.TakesValue: [attribute_validators.tag_is_placeholder_check],
HedKey.UnitClass: [
attribute_validators.tag_is_placeholder_check,
attribute_validators.single_unit_class_check,
attribute_validators.unit_class_requires_numeric_check,
],
HedKey.ValueClass: [attribute_validators.tag_is_placeholder_check],
HedKey.DeprecatedFrom: [attribute_validators.tag_is_deprecated_check],
HedKey.ConversionFactor: [attribute_validators.conversion_factor],
HedKey.AllowedCharacter: [attribute_validators.allowed_characters_check],
HedKey.InLibrary: [attribute_validators.in_library_check],
}
# -----------------------------------------------------------------------
# Construction
# -----------------------------------------------------------------------
[docs]
def __init__(self, hed_schema, error_handler):
self.hed_schema = hed_schema
self.error_handler = error_handler
self._id_validator = HedIDValidator(hed_schema)
self.summary = ComplianceSummary(
schema_name=hed_schema.filename or "",
schema_version=hed_schema.version_number,
)
# -----------------------------------------------------------------------
# Top-level checks
# -----------------------------------------------------------------------
[docs]
def check_if_prerelease_version(self):
"""Warn if this schema version is newer than all known released versions."""
self.summary.start_check(
"prerelease_version",
"Check if schema version is newer than all known released versions.",
)
issues = []
libraries = self.hed_schema.library.split(",")
versions = self.hed_schema.version_number.split(",")
for library, version in zip(libraries, versions, strict=False):
all_known = hed_cache.get_hed_versions(library_name=library, check_prerelease=False)
if not all_known or Version(all_known[0]) < Version(version):
issues += ErrorHandler.format_error(
SchemaWarnings.SCHEMA_PRERELEASE_VERSION_USED,
version,
all_known,
)
if self.hed_schema.with_standard:
all_known = hed_cache.get_hed_versions(check_prerelease=False)
if not all_known or Version(all_known[0]) < Version(self.hed_schema.with_standard):
issues += ErrorHandler.format_error(
SchemaWarnings.SCHEMA_PRERELEASE_VERSION_USED,
self.hed_schema.with_standard,
all_known,
)
self.error_handler.add_context_and_filter(issues)
self.summary.record_issues(len(issues))
return issues
[docs]
def check_prologue_epilogue(self):
"""Validate characters in the prologue and epilogue."""
self.summary.start_check(
"prologue_epilogue",
"Validate characters in prologue and epilogue text.",
)
self.summary.add_sub_check("prologue character validation")
self.summary.add_sub_check("epilogue character validation")
issues = []
char_set = get_allowed_characters_by_name(["text", "newline"])
for label, text in [("Prologue", self.hed_schema.prologue), ("Epilogue", self.hed_schema.epilogue)]:
for _, index in get_problem_indexes(text, char_set):
issues += ErrorHandler.format_error(
SchemaWarnings.SCHEMA_PROLOGUE_CHARACTER_INVALID,
char_index=index,
source_string=text,
section_name=label,
)
self.error_handler.add_context_and_filter(issues)
self.summary.record_issues(len(issues))
return issues
[docs]
def check_invalid_characters(self):
"""Validate characters in entry names and descriptions."""
self.summary.start_check(
"invalid_characters",
"Validate characters in entry names and descriptions.",
)
self.summary.add_sub_check("tag name capitalization")
self.summary.add_sub_check("term/name character validation")
self.summary.add_sub_check("description character validation")
issues = []
for section_key in HedSectionKey:
checked = 0
skipped = 0
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(section_key))
for entry in self.hed_schema[section_key].values():
if entry.has_attribute(HedKey.DeprecatedFrom):
skipped += 1
continue
checked += 1
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, str(entry))
validator = validate_schema_tag if section_key == HedSectionKey.Tags else validate_schema_term
new_issues = validator(entry) + validate_schema_description(entry)
self.error_handler.add_context_and_filter(new_issues)
issues += new_issues
self.error_handler.pop_error_context()
self.summary.record_section(section_key, checked, skipped)
self.error_handler.pop_error_context()
self.summary.record_issues(len(issues))
return issues
# -----------------------------------------------------------------------
# Attribute checking — the core domain + range logic
# -----------------------------------------------------------------------
[docs]
def check_attributes(self):
"""Validate every attribute on every entry in every section.
For each attribute this performs three layers of checking:
1. **Domain** — the attribute is valid for the entry's section.
Any attribute not in the section's ``valid_attributes`` was
already flagged as ``_unknown_attributes`` during loading;
those are reported here.
2. **Range** — the attribute value matches the range type declared
on the attribute's own definition (boolRange, tagRange, etc.).
3. **Semantic** — extra attribute-specific rules (e.g. takesValue
requires a placeholder entry, deprecatedFrom version must exist).
"""
self.summary.start_check(
"attributes",
"Validate attribute domains, ranges, and semantic constraints.",
)
self.summary.add_sub_check("unknown/invalid attribute detection (domain)")
self.summary.add_sub_check("deprecated attribute usage")
self.summary.add_sub_check("range-based value validation")
self.summary.add_sub_check("semantic attribute validation")
self.summary.add_sub_check("hedId validation")
issues = []
for section_key in HedSectionKey:
entry_count = 0
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(section_key))
for entry in self.hed_schema[section_key].values():
entry_count += 1
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, entry.name)
issues += self._check_entry_attributes(entry)
self.error_handler.pop_error_context()
self.summary.record_section(section_key, entry_count)
self.error_handler.pop_error_context()
self.summary.record_issues(len(issues))
return issues
[docs]
def check_duplicate_hed_ids(self):
"""Check for duplicate hedId values across all schema sections."""
self.summary.start_check(
"duplicate_hed_ids",
"Check for duplicate hedId values within or across schema sections.",
)
issues = []
seen_ids: dict[str, tuple[str, str]] = {} # maps hedId string → (first tag name, section key)
for section_key in HedSectionKey:
section = self.hed_schema[section_key]
self.summary.record_section(section_key, len(section))
for entry in section.values():
hed_id = entry.attributes.get(HedKey.HedID)
if not hed_id:
continue
if hed_id in seen_ids:
first_tag_name, first_section_key = seen_ids[hed_id]
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(section_key))
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, entry.name)
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_HED_ID_INVALID,
entry.name,
new_id=hed_id,
duplicate_tag=first_tag_name,
duplicate_tag_section=first_section_key,
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
else:
seen_ids[hed_id] = (entry.name, section_key.value)
self.summary.record_issues(len(issues))
return issues
[docs]
def check_duplicate_names(self):
"""Check for duplicate entry names across library merges."""
self.summary.start_check(
"duplicate_names",
"Check for duplicate entry names within or across library merges.",
)
issues = []
for section_key in HedSectionKey:
self.summary.record_section(section_key, len(self.hed_schema[section_key]))
for name, dups in self.hed_schema[section_key].duplicate_names.items():
libraries = {e.has_attribute(HedKey.InLibrary) for e in dups}
code = (
SchemaErrors.SCHEMA_DUPLICATE_FROM_LIBRARY
if len(libraries) == 2
else SchemaErrors.SCHEMA_DUPLICATE_NODE
)
issues += self.error_handler.format_error_with_context(
code,
name,
duplicate_tag_list=[e.name for e in dups],
section=section_key,
)
self.summary.record_issues(len(issues))
return issues
# Attributes that are tool bookkeeping, not schema content, and so never a Properties difference.
_PROPERTY_COMPARE_EXCLUDED = (HedKey.InLibrary, HedKey.HedID)
[docs]
def check_library_partner_rules(self):
"""Check the rules a library file follows (Appendix B SCHEMA_LIBRARY_INVALID j, k, l, m, n).
Some rules depend on the file form, which ``HedSchema.unmerged_libraries`` records through loading:
j: an unmerged partnered library declares no Properties; k: a merged partnered library's Properties
section equals its partner's. The others hold for every library: l: no element of the library's own
carries ``reserved``; m: no tag of its own has a reserved tag's name; n (partnered only): no node of its
own sits under a reserved tag of the partner.
Notes:
- A library's own elements are every element of an unpartnered library, and the elements carrying
``inLibrary`` in a partnered one (an unmerged file is stamped on loading, a merged file carries
the attribute). The partner's reserved tags in a merged file are therefore not affected.
- A merged file keeps its own Properties section, compared entry by entry with the partner's
(reason k): names, descriptions and attributes other than ``inLibrary`` and ``hedId``, which are
tool bookkeeping. The partner is loaded through the schema cache; when it cannot be loaded the
comparison is skipped, since compliance never raises.
- An unpartnered library is self-contained and may declare properties.
- Reserved names are read from reserved_tags.json through ReservedChecker (the nine structural
tags) plus its list of the ten organizational ones.
"""
self.summary.start_check(
"library_partner_rules",
"Check a library's Properties section, its use of the reserved attribute, and its tags that "
"share a reserved tag's name or sit under a reserved tag.",
)
issues = []
library = self.hed_schema.library
with_standard = self.hed_schema.with_standard
if library and with_standard:
unmerged = self.hed_schema.unmerged_libraries
if unmerged:
issues += self._check_unmerged_library_rules(unmerged)
else:
issues += self._compare_properties_with_partner(library, with_standard)
issues += self._check_library_reserved_attribute(library)
elif library:
issues += self._check_library_reserved_attribute(library)
if library:
issues += self._check_library_reserved_names(library)
self.summary.record_issues(len(issues))
return issues
def _is_own_library_entry(self, entry):
"""Return True if *entry* is declared by the library file itself rather than its standard partner."""
if not self.hed_schema.with_standard:
return True
return HedKey.InLibrary in entry.attributes
def _check_unmerged_library_rules(self, unmerged):
"""Return the reason-j and reason-l issues for the entries declared by unmerged library files."""
issues = []
for section_key in HedSectionKey:
for entry in self.hed_schema[section_key].all_entries:
library = entry.attributes.get(HedKey.InLibrary)
if library not in unmerged:
continue
if section_key == HedSectionKey.Properties:
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_PROPERTIES_DECLARED,
entry.name,
library=library,
with_standard=self.hed_schema.with_standard,
)
if HedKey.Reserved in entry.attributes:
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_RESERVED, entry.name, library=library, section=section_key.name
)
return issues
def _check_library_reserved_attribute(self, library):
"""Return the reason-l issues of a merged partnered or an unpartnered library."""
issues = []
for section_key in HedSectionKey:
for entry in self.hed_schema[section_key].all_entries:
if self._is_own_library_entry(entry) and HedKey.Reserved in entry.attributes:
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_RESERVED, entry.name, library=library, section=section_key.name
)
return issues
def _check_library_reserved_names(self, library):
"""Return the reason-m and reason-n issues: own tags with a reserved name or under a reserved tag."""
from hed.validator.reserved_checker import ReservedChecker # noqa: PLC0415 (avoids an import cycle)
# Tag names match without regard to case, so a library 'task' or 'ONSET' is the reserved tag too.
reserved_names = ReservedChecker.get_instance().reserved_names_casefold
issues = []
for entry in self.hed_schema[HedSectionKey.Tags].all_entries:
if not self._is_own_library_entry(entry):
continue
# A '#' entry's short name is its parent's, so it is judged only by its parent below.
if not entry.name.endswith("/#") and entry.short_tag_name.casefold() in reserved_names:
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_RESERVED_NAME, entry.short_tag_name, library=library
)
parent = entry.parent
if (
self.hed_schema.with_standard
and parent is not None
and not self._is_own_library_entry(parent)
and parent.short_tag_name.casefold() in reserved_names
):
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_UNDER_RESERVED,
entry.short_tag_name,
library=library,
reserved_tag=parent.short_tag_name,
)
return issues
# Standard schema versions in which reserved_tags.json expects an attribute or tag (Appendix A.6,
# SCHEMA_RESERVED_INVALID a): reserved and Inset arrive in 8.2.0, as does topLevelTagGroup on Delay and
# Duration; the ten organizational tags carry reserved from 8.3.0.
_RESERVED_FROM = "8.2.0"
_ORGANIZATIONAL_RESERVED_FROM = "8.3.0"
_TAGS_FROM = {"Inset": "8.2.0"}
_TOP_LEVEL_FROM = {"Delay": "8.2.0", "Duration": "8.2.0"}
def _compare_properties_with_partner(self, library, with_standard):
"""Return one reason-k issue per Properties entry that is missing, extra, or differs from the partner's."""
from hed.schema.hed_schema_io import _load_schema_version # noqa: PLC0415 (avoids an import cycle)
try:
partner = _load_schema_version(xml_version=with_standard)
except HedFileError:
return []
own = {entry.name: entry for entry in self.hed_schema[HedSectionKey.Properties].all_entries}
theirs = {entry.name: entry for entry in partner[HedSectionKey.Properties].all_entries}
issues = []
for name in sorted(set(own) | set(theirs)):
if name not in theirs:
difference = "is not a property of the partner"
elif name not in own:
difference = "is missing"
else:
difference = self._property_difference(own[name], theirs[name])
if difference:
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_LIBRARY_PROPERTIES_MISMATCH,
name,
library=library,
with_standard=with_standard,
difference=difference,
)
return issues
@classmethod
def _property_difference(cls, own, theirs):
"""Say how two same-named Properties entries differ, or return an empty string when they do not."""
if (own.description or "") != (theirs.description or ""):
return "has a different description"
own_attributes = {k: v for k, v in own.attributes.items() if k not in cls._PROPERTY_COMPARE_EXCLUDED}
their_attributes = {k: v for k, v in theirs.attributes.items() if k not in cls._PROPERTY_COMPARE_EXCLUDED}
if own_attributes != their_attributes:
return "has different attributes"
return ""
[docs]
def check_redundant_units(self):
"""Check that no unit is derivable from another unit of the same unit class.
From HED 8.5.0 a unit class never lists a unit that is already accepted as an SI modifier plus another
unit of the class (or as another unit's plural): 8.4.0 listed ``uV`` beside ``V``, 8.5.0 does not.
Such a unit is reported as SCHEMA_DUPLICATE_NODE. The check applies to the standard schema and to
partnered libraries whose standard version is 8.5.0 or later, and to every unpartnered library
(which defines its own units and modifiers). Earlier standard versions legitimately carry ``uV``
and are not checked.
"""
self.summary.start_check(
"redundant_units",
"Check that no unit is derivable from another unit of the same class (HED 8.5.0+).",
)
issues = []
unit_classes = self.hed_schema[HedSectionKey.UnitClasses]
unit_count = sum(len(unit_class_entry.units) for unit_class_entry in unit_classes.values())
if not self._redundant_unit_check_applies():
self.summary.record_section(HedSectionKey.Units, 0, unit_count)
self.summary.record_issues(0)
return issues
self.summary.record_section(HedSectionKey.Units, unit_count)
for unit_class_entry in unit_classes.values():
for unit_name, unit_entry in unit_class_entry.units.items():
derivations = [
self._describe_derivation(unit_name, other_entry)
for other_entry in unit_class_entry.units.values()
if other_entry is not unit_entry and unit_name in other_entry.derivative_units
]
if not derivations:
continue
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(HedSectionKey.Units))
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, unit_name)
issues += self.error_handler.format_error_with_context(
SchemaErrors.SCHEMA_REDUNDANT_UNIT,
unit_name,
unit_class_name=unit_class_entry.name,
derivations=derivations,
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
self.summary.record_issues(len(issues))
return issues
[docs]
def check_any_units_class(self):
"""Check that the anyUnits pseudo unit class, when present, lists no units and has no defaultUnits.
Specification 4.0.0: a placeholder with unitClass=anyUnits accepts a unit from every unit class of the
schema, so the class itself is empty and has no default. Violations are SCHEMA_ATTRIBUTE_INVALID.
"""
self.summary.start_check(
"any_units_class", "Check that the anyUnits pseudo unit class lists no units and has no defaultUnits."
)
issues = []
unit_class_entry = self.hed_schema[HedSectionKey.UnitClasses].get(ANY_UNITS_CLASS)
self.summary.record_section(HedSectionKey.UnitClasses, 1 if unit_class_entry else 0)
if unit_class_entry is not None:
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(HedSectionKey.UnitClasses))
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, ANY_UNITS_CLASS)
if unit_class_entry.units:
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_ANY_UNITS_HAS_UNITS,
ANY_UNITS_CLASS,
", ".join(unit_class_entry.units),
)
default_units = unit_class_entry.attributes.get(HedKey.DefaultUnits)
if default_units:
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_ANY_UNITS_HAS_DEFAULT, ANY_UNITS_CLASS, default_units
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
self.summary.record_issues(len(issues))
return issues
[docs]
def check_units_unique_across_classes(self):
"""Check that no unit string is derived by two unit classes without being listed by either.
Specification 4.0.0: a placeholder with unitClass=anyUnits resolves a unit against every class, and a
string that two classes derive (modifier + unit, plural) but neither lists would be ambiguous. A string
listed in one class and derived in another is fine: the listed unit wins (dB is the decibel, not d + B).
Two separate entries with the same name are a duplicate name, reported by check_duplicate_names; one entry
listed by two classes (possible through the JSON loader) is reported here. Violations are
SCHEMA_DUPLICATE_NODE.
"""
self.summary.start_check(
"units_unique_across_classes",
"Check that no unit string is derived by two unit classes and listed by none.",
)
issues = []
listed_in = defaultdict(set) # unit name -> classes listing it
derived_in = defaultdict(set) # form -> classes deriving it
derived_from = defaultdict(set) # form -> listed units it derives from
unit_classes = self.hed_schema[HedSectionKey.UnitClasses]
for unit_class_name, unit_class_entry in unit_classes.items():
for unit_name, unit_entry in unit_class_entry.units.items():
listed_in[unit_name].add(unit_class_name)
for form in unit_entry.derivative_units:
derived_in[form].add(unit_class_name)
derived_from[form].add(unit_name)
listed = set(listed_in)
self.summary.record_section(HedSectionKey.UnitClasses, len(unit_classes))
# A unit listed by two classes through one shared entry (the JSON loader reuses entries by name) is
# not a duplicate name, so check_duplicate_names does not see it; report the listing itself here.
# Two separate entries with the same name are left to check_duplicate_names.
unit_entry_counts = defaultdict(int)
for unit_entry in self.hed_schema[HedSectionKey.Units].all_entries:
unit_entry_counts[unit_entry.name] += 1
for unit_name in sorted(listed_in):
class_names = listed_in[unit_name]
if len(class_names) > 1 and unit_entry_counts[unit_name] == 1:
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(HedSectionKey.Units))
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, unit_name)
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_UNIT_IN_TWO_CLASSES, unit_name, ", ".join(sorted(class_names))
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
for form in sorted(derived_in):
class_names = derived_in[form]
# Forms of one unit listed under two classes are the same collision as the duplicate name itself,
# which check_duplicate_names already reports; only forms of two different units are reported here.
if len(class_names) > 1 and len(derived_from[form]) > 1 and form not in listed:
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(HedSectionKey.Units))
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, form)
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_UNIT_DERIVED_IN_TWO_CLASSES, form, ", ".join(sorted(class_names))
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
self.summary.record_issues(len(issues))
return issues
def _annotation_check_applies(self):
"""Return True when the annotation grammar applies to this schema.
The specification scopes SCHEMA_ANNOTATION_INVALID to standard schemas with versions 8.5.0 or
later and to library schemas partnered with them. An unpartnered library has no standard
version and falls outside the rule.
"""
standard_version = schema_version_for_library(self.hed_schema, "")
if standard_version is None:
return False
return Version(standard_version) >= Version("8.5.0")
def _redundant_unit_check_applies(self):
"""Return True when the redundant-unit rule applies to this schema.
The rule is keyed to the standard schema version: 8.5.0 or later for the standard schema and for
partnered libraries. An unpartnered library has no standard version and is always checked.
"""
standard_version = schema_version_for_library(self.hed_schema, "")
if standard_version is None:
return True
return Version(standard_version) >= Version("8.5.0")
@staticmethod
def _describe_derivation(unit_name, other_entry):
"""Say how *unit_name* is derived from *other_entry*, for the error message."""
for modifier in other_entry.unit_modifiers:
if unit_name.startswith(modifier.name) and unit_name[len(modifier.name) :] in other_entry.derivative_units:
base = unit_name[len(modifier.name) :]
return f"modifier '{modifier.name}' + unit '{base}'"
return f"a form of unit '{other_entry.name}'"
[docs]
def check_extras_columns(self):
"""Validate that all extras DataFrames have non-empty values in required columns.
For each extras section (Sources, Prefixes, ExternalAnnotations), checks
that every cell in the required columns defined in
``df_constants.extras_column_dict`` has a non-empty value.
Note:
Missing columns are automatically added with empty strings during
schema loading (see ``base2schema.fix_extra``), so only value
presence needs to be checked here.
"""
self.summary.start_check(
"extras_columns",
"Validate extras sections have non-empty values in required columns.",
)
self.summary.add_sub_check("non-empty cell values")
issues = []
extras = getattr(self.hed_schema, "extras", {}) or {}
for section_name, required_cols in df_constants.extras_column_dict.items():
df = extras.get(section_name)
if df is None or (isinstance(df, pd.DataFrame) and df.empty):
# Empty extras are fine — nothing to validate
continue
rows_checked = len(df)
self.summary.record_section(section_name, rows_checked)
for col in required_cols:
if col not in df.columns:
continue
mask = df[col].isna() | df[col].astype(str).str.strip().eq("")
for row_idx in mask[mask].index:
issues += ErrorHandler.format_error(
SchemaAttributeErrors.SCHEMA_MISSING_EXTRA,
section_name=section_name,
column_name=col,
row_index=row_idx,
)
self.error_handler.add_context_and_filter(issues)
self.summary.record_issues(len(issues))
return issues
[docs]
def check_annotation_attribute_values(self):
"""Validate that every ``annotation`` attribute value follows the grammar of Appendix A.1.4.2.
Each value is ``prefix:id`` optionally followed by one space and a value. A failure is a
SCHEMA_ANNOTATION_INVALID warning carrying one of four reasons:
a. ``prefix`` is not a row of the Prefixes section.
b. ``prefix:id`` is not a row of the External annotations section. A bare external term such
as ``glotto:afro1255`` falls here; it is written ``skos:exactMatch glotto:afro1255``.
c. A ``dc:source`` value names no row of the Sources section, by name or by a URL under a
row's link.
d. A value of shape ``prefix:term`` whose prefix is not a row of the Prefixes section, or a
mapping property (``skos:exactMatch``, ``skos:closeMatch``) whose value is not an external
term in prefix notation at all.
The rule is scoped as the specification scopes it: standard schemas with versions 8.5.0 or
later and library schemas partnered with them. Earlier standard schemas wrote bare terms such
as ``ncit:C25499`` and are not checked, and neither is an unpartnered library.
"""
self.summary.start_check(
"annotation_attributes",
"Validate annotation attribute values against the annotation grammar (HED 8.5.0+).",
)
self.summary.add_sub_check("prefix defined in Prefixes")
self.summary.add_sub_check("prefix:id in ExternalAnnotations")
self.summary.add_sub_check("dc:source references valid Sources entry")
self.summary.add_sub_check("external term carries a defined prefix")
issues = []
if not self._annotation_check_applies():
self.summary.record_section("annotation_entries", 0)
self.summary.record_issues(0)
return issues
# Build lookup sets from extras
extras = getattr(self.hed_schema, "extras", {}) or {}
defined_prefixes = self._get_extras_column_values(extras, df_constants.PREFIXES_KEY, df_constants.prefix)
external_pairs = self._get_external_annotation_pairs(extras)
defined_sources = self._get_extras_column_values(extras, df_constants.SOURCES_KEY, df_constants.source)
defined_links = self._get_extras_column_values(extras, df_constants.SOURCES_KEY, df_constants.link)
# Scan all entries in all sections for the "annotation" attribute
entries_checked = 0
for section_key in HedSectionKey:
self.error_handler.push_error_context(ErrorContext.SCHEMA_SECTION, str(section_key))
for entry in self.hed_schema[section_key].values():
annotation_value = entry.attributes.get("annotation")
if not annotation_value:
continue
entries_checked += 1
self.error_handler.push_error_context(ErrorContext.SCHEMA_TAG, entry.name)
# Annotation values can be comma-separated (multiple annotations)
for single_annotation in annotation_value.split(","):
single_annotation = single_annotation.strip()
if single_annotation:
issues += self._validate_annotation_value(
entry,
single_annotation,
defined_prefixes,
external_pairs,
defined_sources,
defined_links,
)
self.error_handler.pop_error_context()
self.error_handler.pop_error_context()
self.summary.record_section("annotation_entries", entries_checked)
self.summary.record_issues(len(issues))
return issues
# -----------------------------------------------------------------------
# Private helpers — extras / annotation validation
# -----------------------------------------------------------------------
@staticmethod
def _get_extras_column_values(extras, section_key, column_name):
"""Return the set of values in a column of an extras DataFrame.
Parameters:
extras (dict): The schema extras dictionary.
section_key (str): Key into the extras dict (e.g. "Prefixes").
column_name (str): The column whose values to collect.
Returns:
set: The set of non-empty string values in that column.
"""
df = extras.get(section_key)
if df is None or not isinstance(df, pd.DataFrame) or df.empty:
return set()
if column_name not in df.columns:
return set()
return {str(v).strip() for v in df[column_name] if pd.notna(v) and str(v).strip()}
@staticmethod
def _get_external_annotation_pairs(extras):
"""Return a set of (prefix, id) tuples from the ExternalAnnotations DataFrame.
Parameters:
extras (dict): The schema extras dictionary.
Returns:
set: Set of (prefix_str, id_str) tuples.
"""
df = extras.get(df_constants.EXTERNAL_ANNOTATION_KEY)
if df is None or not isinstance(df, pd.DataFrame) or df.empty:
return set()
pairs = set()
if df_constants.prefix in df.columns and df_constants.id in df.columns:
for _, row in df.iterrows():
p = str(row[df_constants.prefix]).strip() if pd.notna(row[df_constants.prefix]) else ""
i = str(row[df_constants.id]).strip() if pd.notna(row[df_constants.id]) else ""
if p and i:
pairs.add((p, i))
return pairs
# Mapping properties whose value must itself be an external term in prefix notation.
_MAPPING_PROPERTIES = frozenset({("skos:", "exactMatch"), ("skos:", "closeMatch")})
# A URL token inside free dc:source text, e.g. "Adapted from https://fooddb.example.org/apple".
_URL_TOKEN_RE = re.compile(r"https?://\S+", re.IGNORECASE)
@staticmethod
def _normalize_link(url):
"""Return a URL without its scheme or its trailing slash, for Sources link comparison.
Parameters:
url (str): A URL, with or without an http or https scheme.
Returns:
str: The URL lowercased, without the scheme and without any trailing slashes.
"""
without_scheme = re.sub(r"^https?://", "", url.strip(), flags=re.IGNORECASE)
return without_scheme.rstrip("/").lower()
@staticmethod
def _is_prefix_notation(text):
"""Return True if text has the shape 'prefix:term' rather than free text or a bare URL.
Both parts must be there: a prefix before the colon and a non-empty term after it, all in the
first whitespace-delimited token. A bare ``foodonto:`` names no term and is not prefix
notation, and neither is free text whose first word happens to end in a colon.
Parameters:
text (str): The value to inspect.
Returns:
bool: True if the first token is 'prefix:term' with both parts non-empty.
"""
if not text or not text.strip():
return False
first_token = text.split()[0]
if first_token.lower().startswith(("http://", "https://")):
return False
colon_pos = first_token.find(":")
if colon_pos < 1:
return False
return bool(first_token[colon_pos + 1 :])
@staticmethod
def _url_is_under_link(normalized_token, normalized_link):
"""Return True if a normalized URL is a Sources link itself or a path under it.
The match stops at a path boundary, so a link such as ``fooddb.example.org`` names the row for
``fooddb.example.org/apple`` but not for the unrelated host ``fooddb.example.org.evil``.
Parameters:
normalized_token (str): A URL from the text, passed through :meth:`_normalize_link`.
normalized_link (str): A Sources row link, passed through :meth:`_normalize_link`.
Returns:
bool: True if the URL belongs to that row.
"""
if normalized_token == normalized_link:
return True
return normalized_token.startswith(normalized_link) and normalized_token[len(normalized_link)] in "/?#"
@classmethod
def _source_text_names_row(cls, text, defined_sources, defined_links):
"""Return True if a dc:source text names a Sources row (rule R1 of the annotation rules).
The text names a row when it begins with the row's source name, or when it contains a URL
that is the row's link or a path under it, compared without the scheme and without a trailing
slash.
Parameters:
text (str): The text following 'dc:source'.
defined_sources (set): The source names from the Sources section.
defined_links (set): The links from the Sources section.
Returns:
bool: True if the text names a row.
"""
if not text:
return False
if any(source and text.startswith(source) for source in defined_sources):
return True
normalized_links = {cls._normalize_link(link) for link in defined_links if link}
normalized_links.discard("")
if not normalized_links:
return False
for token in cls._URL_TOKEN_RE.findall(text):
normalized_token = cls._normalize_link(token.rstrip(".),"))
if any(cls._url_is_under_link(normalized_token, link) for link in normalized_links):
return True
return False
def _annotation_issue(self, tag_name, annotation_value, reason, **kwargs):
"""Report one SCHEMA_ANNOTATION_INVALID reason against an annotation value."""
return self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_ANNOTATION_INVALID,
tag_name,
annotation_value=annotation_value,
reason=reason,
**kwargs,
)
def _validate_annotation_value(
self, entry, annotation_value, defined_prefixes, external_pairs, defined_sources, defined_links
):
"""Validate a single annotation attribute value.
Parameters:
entry: The schema entry with the annotation attribute.
annotation_value (str): The annotation value string.
defined_prefixes (set): Valid prefixes from the Prefixes section.
external_pairs (set): Valid (prefix, id) pairs from ExternalAnnotations.
defined_sources (set): Valid source names from the Sources section.
defined_links (set): Links of the Sources rows.
Returns:
list: A list of issue dicts.
"""
issues = []
tag_name = entry.name
# Parse prefix:id from the annotation value
# Expected format: "prefix:id rest_of_text" e.g. "dc:source Beniczky ea 2017 Table 2."
colon_pos = annotation_value.find(":")
if colon_pos < 1:
# No colon found - cannot parse prefix:id
issues += self._annotation_issue(tag_name, annotation_value, "a", prefix="(none)")
return issues
ann_prefix = annotation_value[: colon_pos + 1] # e.g. "dc:"
remainder = annotation_value[colon_pos + 1 :] # e.g. "source Beniczky ea 2017 Table 2."
# Split remainder into id and rest - id is the first whitespace-delimited token
parts = remainder.split(None, 1) # split on whitespace, max 1 split
ann_id = parts[0] if parts else remainder # e.g. "source"
rest_text = parts[1].strip() if len(parts) > 1 else "" # e.g. "Beniczky ea 2017 Table 2."
# Reason a: the prefix must be in Prefixes
if ann_prefix not in defined_prefixes:
issues += self._annotation_issue(tag_name, annotation_value, "a", prefix=ann_prefix)
# Reason b: prefix:id must be in External annotations
if (ann_prefix, ann_id) not in external_pairs:
issues += self._annotation_issue(tag_name, annotation_value, "b", prefix=ann_prefix, annotation_id=ann_id)
# Reason c: a dc:source value must name a Sources row, by name or by a URL under its link
if ann_prefix == "dc:" and ann_id == "source":
if not self._source_text_names_row(rest_text, defined_sources, defined_links):
issues += self._annotation_issue(tag_name, annotation_value, "c", source_text=rest_text)
# Reason d: a value that is an external term must carry a prefix from the Prefixes section
if self._is_prefix_notation(rest_text):
value_prefix = rest_text[: rest_text.find(":") + 1]
if value_prefix not in defined_prefixes:
issues += self._annotation_issue(tag_name, annotation_value, "d", prefix=value_prefix, term=rest_text)
elif (ann_prefix, ann_id) in self._MAPPING_PROPERTIES:
# A mapping property's value must be an external term in prefix notation at all.
issues += self._annotation_issue(tag_name, annotation_value, "d", term=rest_text)
for issue in issues:
issue["severity"] = ErrorSeverity.WARNING
return issues
# -----------------------------------------------------------------------
# Private helpers — attribute validation
# -----------------------------------------------------------------------
def _check_entry_attributes(self, entry):
"""Run domain, range, and semantic checks on a single schema entry."""
issues = []
# 1. Domain check — report unknown attributes
issues += self._check_unknown_attributes(entry)
# 2–4. Per-attribute range + semantic + deprecated checks
for attribute_name in entry.attributes:
validators = self._build_validators(attribute_name)
issues += self._run_validators(entry, attribute_name, validators)
return issues
def _check_unknown_attributes(self, entry):
"""Report attributes that are not valid for this entry's section."""
issues = []
if entry._unknown_attributes:
for attr in entry._unknown_attributes:
issues += self.error_handler.format_error_with_context(
SchemaAttributeErrors.SCHEMA_ATTRIBUTE_INVALID,
attr,
source_tag=entry.name,
)
return issues
def _build_validators(self, attribute_name):
"""Assemble the validator list for *attribute_name*.
Combines (in order):
- deprecated-attribute check (always)
- range validators (from the attribute's definition)
- semantic validators (hard-coded extras for specific attributes)
- hedId validator (if attribute is hedId)
"""
validators = [attribute_validators.attribute_is_deprecated]
# Range validators — look up the attribute definition to find its range
attr_entry = self.hed_schema.get_tag_entry(attribute_name, HedSectionKey.Attributes)
if attr_entry:
for range_key, range_funcs in self._range_validators.items():
if attr_entry.has_attribute(range_key):
validators.extend(range_funcs)
# Semantic validators
validators.extend(self._semantic_validators.get(attribute_name, []))
# HedID validator
if attribute_name == HedKey.HedID:
validators.append(self._id_validator.verify_tag_id)
return validators
def _run_validators(self, entry, attribute_name, validators):
"""Run a list of validators for one attribute on one entry."""
issues = []
for validator in validators:
self.error_handler.push_error_context(ErrorContext.SCHEMA_ATTRIBUTE, attribute_name)
new_issues = validator(self.hed_schema, entry, attribute_name)
self.error_handler.add_context_and_filter(new_issues)
issues += new_issues
self.error_handler.pop_error_context()
return issues