import csv
import io
import json
import numbers
import pandas as pd
from hdmf.common import MeaningsTable
from hed.models import Sidecar
from hed.schema import HedSchema, HedSchemaGroup
from pynwb.core import DynamicTable, VectorData
from pynwb.event import DurationVectorData, EventsTable, TimestampVectorData
from ndx_hed import HedDurationVector, HedLabMetaData, HedTags, HedValueVector
from ndx_hed.hed_tags import HED_TEMPLATE_TYPES
# Sidecar key under which the HedLabMetaData definitions are exported. In BIDS, definitions live in
# a sidecar entry that names no actual column; "definitions" is the conventional name for it.
DEFINITIONS_KEY = "definitions"
[docs]
def read_bids_tsv(path) -> pd.DataFrame:
"""
Reads a BIDS TSV file into a DataFrame the way hedtools reads it for validation.
Every cell is kept as text exactly as written: a double quote is an ordinary character, a line is
split on every tab, and only an empty cell and ``n/a`` are missing (NaN); ``NA``, ``null``, or ``nan``
are values. Column names are the header cells exactly as written too: the header line is read as a
data row, so pandas does not rename a repeated name (``HED``, ``HED.1``) or a blank one (``Unnamed: 1``),
and a padded, repeated, or blank name reaches ``get_events_table`` and validation as it is in the file
(HED specification 3.2.10). ``pd.read_csv(path, sep="\\t")`` with its defaults removes quotes, reads
``NA`` as missing, and renames repeated and blank names, so a file converted with it is not the file
hedtools validates.
Args:
path (str or path-like or file-like): The TSV file, UTF-8 (a byte order mark is removed).
Returns:
pd.DataFrame: The file's columns, labeled by the header cells, every value a string or NaN.
Raises:
pandas.errors.EmptyDataError: If the file has no header line.
"""
if hasattr(path, "read"):
contents = path.read()
if isinstance(contents, bytes):
contents = contents.decode("utf-8")
else:
with open(path, encoding="utf-8", newline="") as file:
contents = file.read()
if contents.startswith(chr(0xFEFF)): # a UTF-8 byte order mark
contents = contents[1:]
options = {"sep": "\t", "dtype": str, "keep_default_na": False, "quoting": csv.QUOTE_NONE}
header = pd.read_csv(io.StringIO(contents), header=None, nrows=1, **options)
frame = pd.read_csv(io.StringIO(contents), na_values=["", "n/a"], index_col=False, **options)
if len(header) and len(header.columns) == len(frame.columns):
frame.columns = list(header.iloc[0])
return frame
[docs]
def get_categorical_meanings(target_column: VectorData, column_info: dict) -> MeaningsTable:
"""
Converts a categorical column info dict to a MeaningsTable annotating a target column.
As of PyNWB 4.0.0, a MeaningsTable is bound to the VectorData column it annotates and its
name is derived automatically as "{target_column.name}_meanings".
Args:
target_column (VectorData): The column object this MeaningsTable annotates. Must already
be a column of the DynamicTable that the MeaningsTable will be added to.
column_info (dict): The column info dictionary from the sidecar (Levels and/or HED).
Returns:
MeaningsTable: The constructed MeaningsTable object (name "{target_column.name}_meanings").
"""
column_name = target_column.name
description = column_info.get("Description", f"Meanings for {column_name}")
meanings_tab = MeaningsTable(target=target_column, description=description)
levels = column_info.get("Levels", {}) # Default to empty dict
# Only a dict-valued HED provides per-category annotations. A string HED is a column-wide value
# annotation (handled elsewhere as a HedValueVector), not categorical, so ignore it here.
hed_info = column_info.get("HED", None)
if not isinstance(hed_info, dict):
hed_info = None
# The categories are the Levels keys followed by the HED keys that Levels does not list, so a value
# annotated only under HED keeps its annotation. Keys are matched exactly, as validation matches them.
values = list(levels.keys())
if hed_info is not None:
values += [value for value in hed_info if value not in levels]
hed_data = []
for value in values:
meanings_tab.add_row(value=value, meaning=levels.get(value, f"Description for {value}"))
if hed_info is not None:
# A value with no HED entry gets an empty cell (no annotation), not "n/a": an explicit "n/a"
# entry is a key under HED in the sidecar, an absent one is not.
hed_data.append(hed_info.get(value, ""))
if hed_info is not None:
meanings_tab.add_column(
name="HED", description=f"HED tags for {column_name} categories", col_cls=HedTags, data=hed_data
)
return meanings_tab
[docs]
def get_events_table(name: str, description: str, df: pd.DataFrame, meanings: dict) -> "EventsTable":
"""
Converts a pandas DataFrame and meanings dictionary to an EventsTable.
Parameters:
name (str): The name of the EventsTable.
description (str): The description of the EventsTable.
df (pd.DataFrame): The DataFrame containing event data.
meanings (dict): The meanings dictionary with keys "categorical" and "value". The
"categorical" values are raw sidecar column-info dicts (see extract_meanings).
Returns:
EventsTable: The constructed EventsTable object. Categorical columns are stored as plain
VectorData columns, each annotated by a MeaningsTable attached to the table. The onset
column becomes the TimestampVectorData "timestamp"; the duration column becomes a
HedDurationVector when meanings gives it a value template, a DurationVectorData otherwise.
Raises:
ValueError: If the DataFrame repeats a column name, if meanings gives the onset column any
HED or Levels, or if meanings makes the duration column categorical.
"""
# A repeated column name (two blank names included) cannot become an NWB table, whose column names are
# unique, and the specification makes it TABULAR_INVALID; refuse it here rather than build a broken table.
repeated = list(dict.fromkeys(column for column in df.columns[df.columns.duplicated()]))
if repeated:
raise ValueError(
f"Table '{name}' cannot be converted to an EventsTable: the column name(s) {repeated} occur more than once."
)
# The onset column carries no HED, and duration is a value column: it can carry one value template (it
# becomes a HedDurationVector) but cannot be categorical. Refuse the rest rather than drop it silently.
if "onset" in meanings["categorical"] or "onset" in meanings["value"]:
raise ValueError(
f"Table '{name}' cannot be converted to an EventsTable: the sidecar gives the onset column HED or "
"Levels, and an onset column cannot carry HED."
)
if "duration" in meanings["categorical"]:
raise ValueError(
f"Table '{name}' cannot be converted to an EventsTable: the sidecar makes the duration column "
"categorical (Levels or a HED dict); duration can only carry a value template such as 'Duration/# s'."
)
columns = []
# Replace "n/a" with NaN in onset and duration columns directly in DataFrame and make them numeric;
# any other non-numeric value raises rather than becoming NaN
for time_col in ("onset", "duration"):
if time_col in df.columns:
df[time_col] = pd.to_numeric(df[time_col].mask(df[time_col].isin(["n/a", "N/A", "na", "NA"])))
# Add columns from the DataFrame
for col_name in df.columns:
col_data = df[col_name].tolist()
if col_name == "onset":
columns.append(TimestampVectorData(name="timestamp", description="Onset times of events", data=col_data))
elif col_name == "duration" and col_name in meanings["value"]:
columns.append(
HedDurationVector(
name="duration", description="Duration of events", data=col_data, hed=meanings["value"][col_name]
)
)
elif col_name == "duration":
columns.append(DurationVectorData(name="duration", description="Duration of events", data=col_data))
elif col_name in meanings["categorical"]:
# A categorical column is a plain VectorData column; its MeaningsTable is attached to
# the table after it is built (see below).
columns.append(VectorData(name=col_name, description=f"Categorical column {col_name}", data=col_data))
elif col_name in meanings["value"]:
columns.append(
HedValueVector(
name=col_name,
description=f"Value column {col_name}",
data=col_data,
hed=meanings["value"][col_name],
)
)
elif col_name == "HED":
columns.append(HedTags(name="HED", description="HED tags for events", data=col_data))
else:
columns.append(VectorData(name=col_name, description=f"Value column {col_name}", data=col_data))
events_tab = EventsTable(name=name, description=description, columns=columns)
# Attach a MeaningsTable to each categorical column now that the columns exist in the table.
for col_name, column_info in meanings["categorical"].items():
if col_name in events_tab:
meanings_tab = get_categorical_meanings(events_tab[col_name], column_info)
events_tab.add_meanings_table(meanings_tab)
return events_tab
def _get_meanings_table(table: DynamicTable, col_name: str) -> MeaningsTable | None:
"""
Returns the MeaningsTable annotating a column of a table, or None if the column has none.
Prefers the public DynamicTable.get_meanings_for_column() API (it raises KeyError when the
column has no MeaningsTable); falls back to the meanings_tables dict if the API is unavailable.
Args:
table (DynamicTable): The table owning the column.
col_name (str): The name of the column.
Returns:
MeaningsTable or None: The MeaningsTable annotating col_name, or None if there is none.
"""
getter = getattr(table, "get_meanings_for_column", None)
if getter is not None:
try:
return getter(col_name)
except KeyError:
return None
return table.meanings_tables.get(f"{col_name}_meanings")
def _is_missing(value) -> bool:
"""
Returns True if a HED value is missing (None, the empty string, or a NaN of any float width).
The NaN test is self-inequality on any real number, as hedtools' ``is_missing`` does: numpy's
float32 and float16 do not subclass Python's float (only float64 does), so an isinstance check
would treat those NaNs as present, and ``math.isnan`` converts its argument to float, so an integer
too large for a float would raise OverflowError. NaN is the one number that is not equal to itself.
"""
if value is None:
return True
if isinstance(value, str):
return value == ""
if isinstance(value, numbers.Real):
return bool(value != value)
return False
[docs]
def get_levels_and_hed(meanings_table: MeaningsTable) -> tuple:
"""
Extracts the BIDS "Levels" and "HED" dictionaries from a MeaningsTable without using pandas.
Args:
meanings_table (MeaningsTable): The MeaningsTable annotating a categorical column. Its
"value" and "meaning" columns supply the levels; an optional HedTags column named
"HED" supplies the per-value HED annotations.
Returns:
tuple: A tuple containing:
- dict: Levels -- maps each value to its meaning (empty string if there is no meaning).
- dict: HED -- maps each value to its HED string, omitting missing (None, NaN, or
empty) annotations. Empty if the MeaningsTable has no HED column. An explicit ``n/a``
is kept (the sidecar annotates the value as having no HED); a value with an empty cell,
which ``get_categorical_meanings`` writes for a value only ``Levels`` lists, is not a key
under HED, so the rule for keys under HED (specification 3.2.9.2) does not apply to it.
"""
# Each column is sliced once rather than iterated, so a file-backed column costs one read.
values = list(meanings_table["value"].data[:])
meanings = list(meanings_table["meaning"].data[:]) if "meaning" in meanings_table.colnames else []
hed_strings = list(meanings_table["HED"].data[:]) if "HED" in meanings_table.colnames else []
levels = {}
hed_dict = {}
for index, value in enumerate(values):
levels[value] = meanings[index] if index < len(meanings) else ""
if index < len(hed_strings) and not _is_missing(hed_strings[index]):
hed_dict[value] = hed_strings[index]
return levels, hed_dict
[docs]
def get_json_hed_dict(table: DynamicTable, hed_metadata: HedLabMetaData = None) -> dict:
"""
Builds the BIDS-style JSON sidecar dictionary of a DynamicTable directly from its columns.
This is the metadata half of get_bids_tabular(). It reads the column objects and the
MeaningsTable annotating each categorical column, so it needs neither a dataframe nor pandas.
Each entry maps a column name to a BIDS sidecar column-info dict, which may contain:
- "Description": the column's description, if it has one.
- "HED": for a HedValueVector or HedDurationVector, its value template (a string containing a single "#");
for a categorical column, a dict mapping each value to its HED string.
- "Levels": for a categorical column, a dict mapping each value to its meaning.
Columns contribute as follows:
- HedTags (the "HED" column): omitted entirely. Its cells are HED strings, so the column
is self-describing, and "HED" is a reserved sidecar key that must not name an entry.
- HedValueVector and HedDurationVector: the HED template, unless the template is empty or "n/a".
- A plain TimestampVectorData/DurationVectorData: description only. The entry keeps the
column's own name, so an EventsTable "timestamp" column is keyed "timestamp" even though
get_bids_tabular() renames it to "onset" in the dataframe.
- Any other column: the Levels and HED of the MeaningsTable annotating it, if any.
Columns with no metadata are omitted, so the result is empty for a table with no descriptions,
no MeaningsTables, and no HED template columns.
In NWB, HED definitions live in a HedLabMetaData object, which is the only place extra
definitions may come from. When one is given, its definitions are exported under the
"definitions" key as ``{"HED": {"defList": <definitions>}}`` -- a sidecar entry that names no
column, which is how BIDS carries definitions. The resulting sidecar is then self-contained:
the ``Def/`` references in the table can be resolved without supplying the definitions
separately. If the HedLabMetaData holds no definitions, no entry is added.
Parameters:
table (DynamicTable): The table to extract the sidecar metadata from. As with
get_bids_tabular(), this is not meant for a MeaningsTable, whose HED is consumed while
assembling the table whose column it annotates.
hed_metadata (HedLabMetaData, optional): The HED lab metadata supplying the definitions.
If None (the default), the sidecar carries no definitions.
Returns:
dict: The JSON sidecar data with column metadata, levels, HED annotations, and definitions.
Raises:
ValueError: If hed_metadata is given but is not a HedLabMetaData instance.
ValueError: If the definitions would overwrite the entry of a column named "definitions".
"""
if hed_metadata is not None and not isinstance(hed_metadata, HedLabMetaData):
raise ValueError(
"The hed_metadata must be a HedLabMetaData instance -- in NWB it is the only source of "
f"extra HED definitions, but {type(hed_metadata).__name__} was given."
)
json_data = {}
for col_name in table.colnames:
column = table[col_name]
column_info = {}
# Add description if available
if hasattr(column, "description") and column.description:
column_info["Description"] = column.description
# Handle different column types. A HedDurationVector is a DurationVectorData, so it is tested first.
if isinstance(column, HED_TEMPLATE_TYPES):
if column.hed != "" and column.hed != "n/a":
column_info["HED"] = column.hed
elif isinstance(column, (TimestampVectorData, DurationVectorData)):
# A plain BIDS onset or duration column carries no HED metadata.
pass
elif isinstance(column, HedTags):
# The HED column is self-describing (its cells are HED strings). "HED" is a reserved
# sidecar key and must NOT appear as a sidecar metadata entry, so emit nothing for it.
continue
else:
# A categorical column is a plain VectorData annotated by a MeaningsTable.
meanings_table = _get_meanings_table(table, col_name)
if meanings_table is not None:
levels, hed_dict = get_levels_and_hed(meanings_table)
if levels:
column_info["Levels"] = levels
if hed_dict:
column_info["HED"] = hed_dict
# Add column info to JSON if it has any metadata
if column_info:
json_data[col_name] = column_info
# Export the definitions from the HedLabMetaData -- the only source of extra HED definitions
definitions = hed_metadata.definitions if hed_metadata is not None else None
if definitions:
if DEFINITIONS_KEY in json_data:
raise ValueError(
f"Table '{table.name}' has a column named '{DEFINITIONS_KEY}', which collides with the "
f"sidecar key used to export the HED definitions. Rename the column to export definitions."
)
json_data[DEFINITIONS_KEY] = {"HED": {"defList": definitions}}
return json_data
def _bids_column_name(table: DynamicTable, name: str) -> str:
"""Return the BIDS name of a table column: a TimestampVectorData named ``timestamp`` is ``onset``."""
return "onset" if name == "timestamp" and isinstance(table[name], TimestampVectorData) else name
def _repeated_bids_column_names(table: DynamicTable) -> list[str]:
"""Return the BIDS column names that occur more than once in a table, in column order.
A repeated name is TABULAR_INVALID (HED specification 3.2.10); validation reports it through hedtools,
and ``get_bids_dataframe`` refuses to build a dataframe with one.
hdmf keeps a table's column names unique, so the only repeat is an ``onset`` column next to the
TimestampVectorData ``timestamp`` that is exported as ``onset``.
"""
names = [_bids_column_name(table, name) for name in table.colnames]
return list(dict.fromkeys(name for name in names if names.count(name) > 1))
[docs]
def get_bids_dataframe(table: DynamicTable) -> pd.DataFrame:
"""
Converts the data of a DynamicTable to a BIDS-style dataframe, the TSV half of get_bids_tabular().
This reads every column of the table. A ``TimestampVectorData`` column is renamed to ``onset``
so that downstream BIDS-HED validation treats the table as a timeline (temporal) file. Missing
cells (NaN, None, the empty string) become the text ``n/a``, as in a BIDS TSV file, so every
column of the returned dataframe has dtype object.
Parameters:
table (DynamicTable): The table to convert.
Returns:
pd.DataFrame: The table data with BIDS column names (onset, duration, etc.)
Raises:
ValueError: If a BIDS column name would occur twice, that is the table has a column named
``onset`` as well as a TimestampVectorData named ``timestamp`` (TABULAR_INVALID, HED
specification 3.2.10).
"""
repeated = _repeated_bids_column_names(table)
if repeated:
raise ValueError(
f"Table '{table.name}' cannot be converted to a BIDS dataframe: the column name(s) {repeated} would "
"occur more than once (a TimestampVectorData named 'timestamp' is exported as 'onset')."
)
df = table.to_dataframe()
# Rename the timestamp column back to onset so the table reads as a BIDS timeline file. The
# "timestamp" column must itself be a TimestampVectorData -- an unrelated column that merely
# happens to be named "timestamp" is left alone.
if "timestamp" in table.colnames and isinstance(table["timestamp"], TimestampVectorData):
df = df.rename(columns={"timestamp": "onset"})
# hedtools reads BIDS TSV files, where a missing cell is the text "n/a". to_dataframe() holds NaN,
# None, or "" instead; hedtools would turn NaN into the text "nan" ("Delay/nan s") and substitute ""
# into a value template ("Delay/ s"). hedtools converts every column to text on input, so the
# numeric dtype is not needed downstream.
return df.astype(object).where(df.notna() & (df != ""), "n/a")
[docs]
def get_bids_tabular(table: DynamicTable, hed_metadata: HedLabMetaData = None) -> tuple:
"""
Converts a DynamicTable to a BIDS-style tabular representation (DataFrame and JSON sidecar).
Works for any DynamicTable (an EventsTable or a plain DynamicTable). It is not meant for a
MeaningsTable, whose HED is consumed while assembling the table whose column it annotates. The
dataframe is get_bids_dataframe(table) and the sidecar is get_json_hed_dict(table, hed_metadata);
call those directly to get only one half (the sidecar half does not read the table's data).
Parameters:
table (DynamicTable): The table to convert.
hed_metadata (HedLabMetaData, optional): The HED lab metadata supplying the definitions,
passed through to get_json_hed_dict(). If None (the default), the sidecar carries no
definitions and any ``Def/`` references in the table cannot be resolved from it alone.
Returns:
tuple: A tuple containing:
- pd.DataFrame: The table data with BIDS column names (onset, duration, etc.)
- dict: The JSON sidecar data as returned by get_json_hed_dict().
"""
return get_bids_dataframe(table), get_json_hed_dict(table, hed_metadata)