"""
HedValidator class for validating HED tags in NWB DynamicTable objects.
"""
import io
import json
import math
from typing import List, Dict, Any, Optional
from pynwb import NWBFile
from pynwb.core import DynamicTable
from pynwb.event import EventsTable
from hdmf.common import MeaningsTable
from hed.errors import ErrorHandler, ErrorContext, HedExceptions, HedFileError
from hed.errors.error_reporter import check_for_any_errors
from hed.models import HedString, TabularInput, Sidecar
from ..hed_lab_metadata import HedLabMetaData
from ..hed_tags import HedTags, HedValueVector
from .bids2nwb import get_bids_tabular
[docs]
class HedNWBValidator:
"""
A validator class for HED tags in NWB DynamicTable objects.
This class provides methods to validate HED tags in various NWB data structures
using HED schema information stored in HedLabMetaData.
"""
[docs]
def __init__(self, hed_metadata: HedLabMetaData):
"""
Initialize the HedNWBValidator with HED metadata.
Parameters:
hed_metadata (HedLabMetaData): The HED lab metadata containing schema information.
Must be a valid HedLabMetaData instance with a loaded
HED schema. If the HedLabMetaData was constructed successfully,
it is guaranteed to have a valid schema.
Raises:
ValueError: If hed_metadata is not an instance of HedLabMetaData
Notes:
HedLabMetaData validates the schema during its own construction, so if a
HedLabMetaData instance exists, it is guaranteed to have a valid HED schema
and version. No additional validation is needed here.
"""
if not isinstance(hed_metadata, HedLabMetaData):
raise ValueError("hed_metadata must be an instance of HedLabMetaData")
self.hed_schema = hed_metadata.get_hed_schema()
self.def_dict = hed_metadata.get_definition_dict()
[docs]
def validate_table(self, table: DynamicTable, error_handler: Optional[ErrorHandler] = None) -> List[Dict[str, Any]]:
"""
Validates all HedTags columns in a DynamicTable using the provided HED schema metadata.
Parameters:
table (DynamicTable): The dynamic table to validate
error_handler (ErrorHandler, optional): An ErrorHandler instance for collecting errors.
If None, a new instance will be created.
Returns:
List[Dict[str, Any]]: A consolidated list of validation issues from all HedTags columns
"""
if table is None or not isinstance(table, DynamicTable):
raise ValueError("The provided table is not a valid DynamicTable instance.")
if error_handler is None:
error_handler = ErrorHandler(check_for_warnings=False)
issues = []
# Contexts are popped in finally blocks so that an exception raised while validating a column
# does not leave a stale context on a caller-provided error handler.
error_handler.push_error_context(ErrorContext.TABLE_NAME, table.name)
try:
for col in table.columns:
if isinstance(col, HedTags):
validate_column = self.validate_vector
elif isinstance(col, HedValueVector):
validate_column = self.validate_value_vector
else:
continue
error_handler.push_error_context(ErrorContext.COLUMN, col.name)
try:
issues += validate_column(col, error_handler)
finally:
error_handler.pop_error_context()
finally:
error_handler.pop_error_context()
return issues
[docs]
def validate_vector(self, hed_tags: HedTags, error_handler: Optional[ErrorHandler] = None) -> List[Dict[str, Any]]:
"""
Validates a HedTags column using the provided HED schema metadata.
Parameters:
hed_tags (HedTags): The HedTags column to validate
error_handler (ErrorHandler, optional): An ErrorHandler instance for collecting errors.
If None, a new instance will be created.
Returns:
List[Dict[str, Any]]: A list of validation issues found in the HedTags column
Notes:
An annotation that has already been validated in this column and found to have no issues
is not validated again on the rows that repeat it. Such a row contributes nothing to the
result, so the issues returned are the same as if every row were validated. A row whose
annotation does have issues is still validated, so that each affected row is reported.
"""
if hed_tags is None or not isinstance(hed_tags, HedTags):
raise ValueError("The provided hed_tags is not a valid HedTags instance.")
if error_handler is None:
error_handler = ErrorHandler(check_for_warnings=False)
issues = []
validated_without_issues = set()
# Slice once: on a file-backed column, iterating the dataset directly is one HDF5 read per row.
# For an in-memory list the slice is a shallow copy of references (about 4 ms and 8 MB per
# million rows, the cost of validating a few rows); for a numpy array it is a view.
for index, tag in enumerate(hed_tags.data[:]):
if tag is None or tag == "" or tag == "n/a" or tag in validated_without_issues:
continue
error_handler.push_error_context(ErrorContext.ROW, index)
try:
hed_obj = HedString(tag, self.hed_schema, def_dict=self.def_dict)
row_issues = hed_obj.validate(allow_placeholders=False, error_handler=error_handler)
finally:
error_handler.pop_error_context()
issues += row_issues
if not row_issues:
validated_without_issues.add(tag)
return issues
[docs]
def validate_value_vector(
self, hed_values: HedValueVector, error_handler: Optional[ErrorHandler] = None
) -> List[Dict[str, Any]]:
"""
Validates a HedValueVector column using the provided HED schema metadata.
Parameters:
hed_values (HedValueVector): The HedValueVector column to validate
error_handler (ErrorHandler, optional): An ErrorHandler instance for collecting errors.
If None, a new instance will be created.
Returns:
List[Dict[str, Any]]: A list of validation issues found in the HedValueVector column
Notes:
As in validate_vector, a substituted annotation that has already been validated in this
column and found to have no issues is not validated again on the rows that repeat it.
"""
if hed_values is None or not isinstance(hed_values, HedValueVector) or hed_values.hed is None:
raise ValueError("The provided hed_values is not a valid HedValueVector instance.")
if error_handler is None:
error_handler = ErrorHandler(check_for_warnings=False)
issues = []
# Validate the HED template first
hed_template = HedString(hed_values.hed, self.hed_schema, def_dict=self.def_dict)
issues += hed_template.validate(allow_placeholders=True, error_handler=error_handler)
if check_for_any_errors(issues):
return issues
validated_without_issues = set()
# Slice once: on a file-backed column, iterating the dataset directly is one HDF5 read per row.
# For an in-memory list the slice is a shallow copy of references (about 4 ms and 8 MB per
# million rows, the cost of validating a few rows); for a numpy array it is a view.
for index, tag in enumerate(hed_values.data[:]):
if tag is None or tag == "" or tag == "n/a" or (isinstance(tag, float) and math.isnan(tag)):
continue
# Substitute the tag value into the template in place of #
eval_tag = hed_values.hed.replace("#", str(tag))
if eval_tag in validated_without_issues:
continue
error_handler.push_error_context(ErrorContext.ROW, index)
try:
hed_obj = HedString(eval_tag, self.hed_schema, def_dict=self.def_dict)
row_issues = hed_obj.validate(allow_placeholders=False, error_handler=error_handler)
finally:
error_handler.pop_error_context()
issues += row_issues
if not row_issues:
validated_without_issues.add(eval_tag)
return issues
[docs]
def validate_events(
self, events: EventsTable, error_handler: Optional[ErrorHandler] = None
) -> List[Dict[str, Any]]:
"""
Validates HED tags in an EventsTable by converting it to BIDS format and validating the events.
This function extracts the BIDS-formatted DataFrame and JSON sidecar from the EventsTable
using get_bids_tabular(), then validates the HED tags contained within using the provided
HED schema metadata.
Parameters:
events (EventsTable): The EventsTable to validate containing HED tags
error_handler (ErrorHandler, optional): An ErrorHandler instance for collecting errors.
If None, a new instance will be created.
Returns:
List[Dict[str, Any]]: A list of validation issues found in the EventsTable HED tags
Raises:
ValueError: If the EventsTable is invalid or cannot be converted to BIDS format
Notes:
This function uses get_bids_tabular() to extract BIDS-formatted data from the EventsTable,
then applies HED validation to the extracted event annotations. The validation follows
BIDS-HED standards for event annotation validation. If the table's sidecar-level HED (a
HedValueVector template or a MeaningsTable HED string) has any error, validation of the
table stops there and only the sidecar issues are returned (see _validate_assembled).
"""
if events is None or not isinstance(events, EventsTable):
raise ValueError("The provided events is not a valid EventsTable instance.")
if error_handler is None:
error_handler = ErrorHandler(check_for_warnings=False)
return self._validate_assembled(events, error_handler)
def _validate_assembled(self, table: DynamicTable, error_handler: ErrorHandler) -> List[Dict[str, Any]]:
"""
Assembled (BIDS-style) validation of a DynamicTable.
The table is converted to a BIDS-format dataframe + JSON sidecar with get_bids_tabular().
The sidecar (column metadata: value templates and categorical Levels/HED) is validated first
with Sidecar.validate(). If it has any error, validation of the table stops and the sidecar
issues are returned: a sidecar-level error is repeated on every row that uses the template or
the categorical value, and the row annotations assembled from it cannot be trusted, so the
assembled-table step would only add noise. Sidecar issues name the column and, for categorical
HED, the value (``ec_sidecarColumnName``, ``ec_sidecarKeyName``); they have no row. Warnings
alone do not stop validation.
Otherwise the assembled per-row annotations are validated with TabularInput.validate(). Both
steps are needed: TabularInput.validate() does NOT re-run the sidecar's brace-structure /
column-reference checks (self, nested, invalid, or malformed ``{column}`` references), so a
malformed sidecar validated only through the assembled table would be missed and surface as
misleading downstream errors (e.g. a stray ``{`` reported as CHARACTER_INVALID).
Assembly combines, for each row, the row's direct HED column, its categorical HED (from
attached MeaningsTables), and its value-template HED into a single annotation. If the assembled
dataframe has an ``onset`` column, TabularInput performs temporal (timeline) validation;
otherwise it performs non-temporal (per-row) validation.
Every issue carries the table name in the TABLE_NAME error context (``ec_table_name``).
Parameters:
table (DynamicTable): The table to validate.
error_handler (ErrorHandler): The error handler collecting issues.
Returns:
List[Dict[str, Any]]: Validation issues for the table.
"""
error_handler.push_error_context(ErrorContext.TABLE_NAME, table.name)
try:
return self._validate_assembled_in_context(table, error_handler)
finally:
error_handler.pop_error_context()
def _validate_assembled_in_context(self, table: DynamicTable, error_handler: ErrorHandler) -> List[Dict[str, Any]]:
"""Body of _validate_assembled, run with the TABLE_NAME context already pushed."""
df, json_data = get_bids_tabular(table)
# No sidecar metadata: validate the assembled table on its own (e.g. only a direct HED column).
if not json_data:
tab_input = TabularInput(file=df, name=table.name)
return tab_input.validate(self.hed_schema, extra_def_dicts=self.def_dict, error_handler=error_handler)
# Step 1: validate the sidecar metadata explicitly. Only Sidecar.validate() performs the
# brace-structure / column-reference checks; it also validates the HED of every categorical
# level even those not present in the data.
sidecar = Sidecar(io.StringIO(json.dumps(json_data)), name=table.name)
sidecar_issues = sidecar.validate(self.hed_schema, extra_def_dicts=self.def_dict, error_handler=error_handler)
for issue in sidecar_issues:
if not issue.get("ec_filename"): # Sidecar.validate() leaves it empty
issue["ec_filename"] = table.name
# Any sidecar error (a bad value template or a bad categorical HED string) would be repeated on
# every row that uses it, and the row annotations assembled from it are not trustworthy, so
# stop here. The sidecar issues already name the column and the categorical value.
if check_for_any_errors(sidecar_issues):
return sidecar_issues
# Step 2: validate the assembled table. This carries row context (ec_column / ec_row) and
# performs temporal (timeline) validation when an ``onset`` column is present. Sidecar
# warnings for values that occur in the data are re-reported here with that context.
tab_input = TabularInput(file=df, sidecar=sidecar, name=table.name)
issues = tab_input.validate(self.hed_schema, extra_def_dicts=self.def_dict, error_handler=error_handler)
# TabularInput only sees categorical values that occur in the data, so add the sidecar
# warnings for categorical levels that never appear (otherwise they would be missed).
issues += self._unused_categorical_level_issues(sidecar_issues, df, json_data)
return issues
@staticmethod
def _unused_categorical_level_issues(sidecar_issues, df, json_data):
"""Return the sidecar issues for categorical levels that do not occur in the data.
TabularInput validates only values present in the data, so a bad HED annotation on a
categorical level that is never used would be missed. Those sidecar issues are added back.
Value-column (template) and data-column errors are excluded because TabularInput reports them.
"""
extra = []
for issue in sidecar_issues:
col = issue.get("ec_sidecarColumnName")
key = issue.get("ec_sidecarKeyName")
if not col or key is None or col not in df.columns:
continue
if "Levels" not in json_data.get(col, {}): # only categorical columns have levels
continue
present = {str(v) for v in df[col].tolist()}
if str(key) not in present:
extra.append(issue)
return extra
def _check_meanings_table_rules(self, meanings_table: MeaningsTable) -> None:
"""
Enforce structural rules on a MeaningsTable.
A MeaningsTable carries categorical (per-value) HED in a HedTags column. A HedValueVector
(a value template with a ``#`` placeholder) has no meaning for per-value categorical
annotation and is not allowed in a MeaningsTable.
Raises:
ValueError: If the MeaningsTable contains a HedValueVector column.
"""
for col in meanings_table.columns:
if isinstance(col, HedValueVector):
raise ValueError(
f"HedValueVector column '{col.name}' is not allowed in MeaningsTable "
f"'{meanings_table.name}'; categorical HED must be stored in a HedTags column."
)
[docs]
def validate_file(self, nwbfile: NWBFile, error_handler: Optional[ErrorHandler] = None) -> List[Dict[str, Any]]:
"""
Validates all HED tags in an NWB file by iterating through all DynamicTable objects.
This method first checks that HedLabMetaData is defined in the NWB file and that its schema
version matches the validator's. Then it validates every DynamicTable **except MeaningsTable**
using assembled (BIDS-style) validation (see _validate_assembled): the table is converted to a
dataframe + sidecar and validated with TabularInput, which performs temporal (timeline)
validation when the table has an ``onset`` column and non-temporal validation otherwise. A
MeaningsTable is not validated on its own -- its categorical HED is validated as part of the
table whose column it annotates -- but it is checked against the structural rule that it must
not contain a HedValueVector column (a violation raises ValueError).
For each table, the sidecar-level HED (HedValueVector templates and MeaningsTable HED strings)
is validated first; if it has any error, validation of that table stops and only its sidecar
issues are returned, since such an error would repeat on every row that uses it. Every issue
carries its table's name in ``ec_table_name``.
Parameters:
nwbfile (NWBFile): The NWB file to validate
error_handler (ErrorHandler, optional): An ErrorHandler instance for collecting errors.
If None, a new instance will be created.
Returns:
List[Dict[str, Any]]: A consolidated list of validation issues from all tables in the file
Raises:
ValueError: If nwbfile is not a valid NWBFile instance
ValueError: If a MeaningsTable contains a HedValueVector column
HedFileError: If HedLabMetaData is missing or invalid in the NWB file
HedFileError: If the HED schema version in the NWB file does not match the validator's schema version
"""
if nwbfile is None or not isinstance(nwbfile, NWBFile):
raise ValueError("The provided nwbfile is not a valid NWBFile instance.")
# Check if HedLabMetaData is defined in the file and matches the validator's schema version
hed_metadata = nwbfile.lab_meta_data.get("hed_schema")
if hed_metadata is None or not isinstance(hed_metadata, HedLabMetaData):
raise HedFileError(
HedExceptions.SCHEMA_INVALID, f"NWB file {nwbfile.identifier} does not have a valid HED schema", ""
)
if hed_metadata.get_hed_schema_version() != self.hed_schema.version:
raise HedFileError(
HedExceptions.SCHEMA_VERSION_INVALID,
f"HED schema version in NWB file ({hed_metadata.get_hed_schema_version()})"
+ " does not match validator schema version"
+ f"({self.hed_schema.version})",
"",
)
if error_handler is None:
error_handler = ErrorHandler(check_for_warnings=False)
issues = []
error_handler.push_error_context(ErrorContext.FILE_NAME, nwbfile.identifier)
# Validate every DynamicTable with assembled (BIDS-style) validation, except MeaningsTables.
# A MeaningsTable is a lookup consumed during the assembly of the table whose column it
# annotates, so it is not validated on its own; it is only checked against the structural
# rule that it must not contain a HedValueVector column.
for obj in nwbfile.all_children():
if not isinstance(obj, DynamicTable):
continue
if isinstance(obj, MeaningsTable):
self._check_meanings_table_rules(obj) # raises ValueError on a disallowed column
continue
issues.extend(self._validate_assembled(obj, error_handler))
error_handler.pop_error_context()
return issues