Source code for hed.models.base_input
"""
Superclass representing a basic columnar file.
"""
import csv
import io
import os
import warnings
import openpyxl
import pandas as pd
from hed.errors.exceptions import HedExceptions, HedFileError
from hed.models.column_mapper import ColumnMapper
from hed.models.column_metadata import ColumnMetadata
from hed.models.column_source import check_column_names, column_name_text, distinct_values
from hed.models.definition_dict import DefinitionDict
from hed.models.df_util import _handle_curly_braces_refs, filter_series_by_onset
[docs]
class BaseInput:
"""Superclass representing a basic columnar file."""
TEXT_EXTENSION = [".tsv", ".txt"]
EXCEL_EXTENSION = [".xlsx"]
[docs]
def __init__(
self,
file,
file_type=None,
worksheet_name=None,
has_column_names=True,
mapper=None,
name=None,
allow_blank_names=None,
):
"""Constructor for the BaseInput class.
Parameters:
file (str or file-like or pd.Dataframe): An xlsx/tsv file to open. A DataFrame's column labels are
taken literally, except that each is converted to text as an Excel header cell is (``1`` or
``1.0`` becomes ``"1"``; None or NaN becomes ``""``, a blank name). A DataFrame already read with
pandas has lost its original header (pandas renames an empty name to ``Unnamed: N`` and suffixes
a repeated one, ``trial.1``), so its blank and repeated names cannot be found. Pass the file path
or a text stream to have them checked.
file_type (str or None): ".xlsx" (Excel), ".tsv" or ".txt" (tab-separated text).
Derived from file if file is a filename. Ignored if pandas dataframe.
worksheet_name (str or None): Name of Excel workbook worksheet name to use.
(Not applicable to tsv files.)
has_column_names (bool): True if file has column names.
This value is ignored if you pass in a pandas dataframe.
mapper (ColumnMapper or None): Indicates which columns have HED tags.
See SpreadsheetInput or TabularInput for examples of how to use built-in a ColumnMapper.
name (str or None): Optional field for how this file will report errors.
allow_blank_names: Deprecated and ignored (removed in hedtools 2.0.0). A blank column name no
longer raises; ``validate`` reports it as TABULAR_INVALID.
Raises:
HedFileError: For various issues.
Notes: Reasons for raising HedFileError include:
- file is blank.
- An invalid dataframe was passed with size 0.
- An invalid extension was provided.
- Cannot open the indicated file.
- The specified worksheet name does not exist.
- If the sidecar file or tabular file had invalid format and could not be read.
Column names are kept exactly as read. A padded, blank or repeated column name does not raise: it
is recorded in ``padded_columns``, ``blank_columns`` or ``duplicate_columns`` and ``validate``
reports it as TABULAR_INVALID.
"""
if allow_blank_names is not None:
warnings.warn(
"The allow_blank_names= parameter of BaseInput is deprecated, ignored, and will be removed in "
"hedtools 2.0.0; a blank column name is reported as TABULAR_INVALID by validate.",
DeprecationWarning,
stacklevel=2,
)
if mapper is None:
mapper = ColumnMapper()
self._mapper = mapper
self._has_column_names = has_column_names
self._name = name
# This is the loaded workbook if we loaded originally from an Excel file.
self._loaded_workbook = None
self._worksheet_name = worksheet_name
self._dataframe = None
self._padded_columns = []
self._blank_columns = []
self._duplicate_columns = []
input_type = file_type
if isinstance(file, str):
if file_type is None:
_, input_type = os.path.splitext(file)
if self.name is None:
self._name = file
self._open_dataframe_file(file, has_column_names, input_type)
self.reset_mapper(mapper)
[docs]
def reset_mapper(self, new_mapper):
"""Set mapper to a different view of the file.
Parameters:
new_mapper (ColumnMapper): A column mapper to be associated with this base input.
"""
self._mapper = new_mapper
if not self._mapper:
self._mapper = ColumnMapper()
if self._dataframe is not None and self._has_column_names:
columns = self._dataframe.columns
self._mapper.set_column_map(columns)
@property
def dataframe(self):
"""The underlying dataframe."""
return self._dataframe
@property
def dataframe_a(self) -> pd.DataFrame:
"""Return the assembled dataframe Probably a placeholder name.
Returns:
pd.Dataframe: the assembled dataframe
"""
return self.assemble()
@property
def series_a(self) -> pd.Series:
"""Return the assembled dataframe as a series.
Returns:
pd.Series: the assembled dataframe with columns merged.
"""
return self.combine_dataframe(self.assemble())
@property
def series_filtered(self) -> pd.Series | None:
"""Return the assembled dataframe as a series, with rows that have the same onset combined.
Returns:
Union[pd.Series, None]: the assembled dataframe with columns merged, and the rows filtered together.
"""
if self.onsets is not None:
return filter_series_by_onset(self.series_a, self.onsets)
return None
@property
def onsets(self):
"""Return the onset column if it exists."""
if "onset" in self.columns:
return self._dataframe["onset"]
return None
@property
def needs_sorting(self) -> bool:
"""Return True if this both has an onset column, and it needs sorting."""
onsets = self.onsets
if onsets is not None:
onsets = pd.to_numeric(self.dataframe["onset"], errors="coerce")
return not onsets.is_monotonic_increasing
else:
return False
@property
def name(self) -> str:
"""Name of the data."""
return self._name
@property
def has_column_names(self) -> bool:
"""True if dataframe has column names."""
return self._has_column_names
@property
def duplicate_columns(self) -> list[str]:
"""The column names that appear more than once in the header, each once, in order of its second appearance.
Names are compared exactly; padded names (``padded_columns``) and blank names (``blank_columns``) are not
counted. Read from the header itself, so a repeated name is listed under its own name, not the suffixed
name the loaded dataframe gives its second column. Empty when the file has no column names.
"""
return self._duplicate_columns
@property
def padded_columns(self) -> list[str]:
"""The column names that begin or end with whitespace (blank, tab, LF, CR), each once, in header order.
A name made only of whitespace is padded, not blank. Empty when the file has no column names.
"""
return self._padded_columns
@property
def blank_columns(self) -> list[int]:
"""The 0-based positions of the blank column names (empty; a None or NaN label counts as empty), in header order.
A first line that ends with a tab has a blank last name. Empty when the file has no column names.
"""
return self._blank_columns
@property
def loaded_workbook(self):
"""The underlying loaded workbooks."""
return self._loaded_workbook
@property
def worksheet_name(self):
"""The worksheet name."""
return self._worksheet_name
[docs]
def convert_to_form(self, hed_schema, tag_form):
"""Convert all tags in underlying dataframe to the specified form.
Parameters:
hed_schema (HedSchema): The schema to use to convert tags.
tag_form (str): HedTag property to convert tags to.
Most cases should use convert_to_short or convert_to_long below.
"""
from hed.models.df_util import convert_to_form
convert_to_form(self._dataframe, hed_schema, tag_form, self._mapper.get_tag_columns())
[docs]
def convert_to_short(self, hed_schema):
"""Convert all tags in underlying dataframe to short form.
Parameters:
hed_schema (HedSchema): The schema to use to convert tags.
"""
self.convert_to_form(hed_schema, "short_tag")
[docs]
def convert_to_long(self, hed_schema):
"""Convert all tags in underlying dataframe to long form.
Parameters:
hed_schema (HedSchema or None): The schema to use to convert tags.
"""
self.convert_to_form(hed_schema, "long_tag")
[docs]
def shrink_defs(self, hed_schema):
"""Shrinks any def-expand found in the underlying dataframe.
Parameters:
hed_schema (HedSchema or None): The schema to use to identify defs.
"""
from df_util import shrink_defs
shrink_defs(self._dataframe, hed_schema=hed_schema, columns=self._mapper.get_tag_columns())
[docs]
def expand_defs(self, hed_schema, def_dict):
"""Shrinks any def-expand found in the underlying dataframe.
Parameters:
hed_schema (HedSchema or None): The schema to use to identify defs.
def_dict (DefinitionDict): The definitions to expand.
"""
from df_util import expand_defs
expand_defs(self._dataframe, hed_schema=hed_schema, def_dict=def_dict, columns=self._mapper.get_tag_columns())
[docs]
def to_excel(self, file):
"""Output to an Excel file.
Parameters:
file (str or file-like): Location to save this base input.
Raises:
ValueError: If empty file object was passed.
OSError: If the file cannot be opened.
"""
if not file:
raise ValueError("Empty file name or object passed in to BaseInput.save.")
dataframe = self._dataframe
if self._loaded_workbook:
old_worksheet = self.get_worksheet(self._worksheet_name)
# Excel spreadsheets are 1 based, then add another 1 for column names if present
adj_row_for_col_names = 1
if self._has_column_names:
adj_row_for_col_names += 1
adj_for_one_based_cols = 1
for row_number, text_file_row in dataframe.iterrows():
for column_number, _column_text in enumerate(text_file_row):
cell_value = dataframe.iloc[row_number, column_number]
old_worksheet.cell(
row_number + adj_row_for_col_names, column_number + adj_for_one_based_cols
).value = cell_value
self._loaded_workbook.save(file)
else:
dataframe.to_excel(file, header=self._has_column_names)
[docs]
def to_csv(self, file=None) -> str | None:
"""Write to file or return as a string.
Parameters:
file (str, file-like, or None): Location to save this file. If None, return as string.
Returns:
Union[str, None]: None if file is given or the contents as a str if file is None.
Raises:
OSError: If the file cannot be opened.
csv.Error: If a column name or cell holds a tab or line break, which a TSV file cannot hold.
Notes:
Double quotes are written as ordinary characters, as the file is read (see ``_load_text_file``), so a
file written here reads back with the same column names and cells.
"""
dataframe = self._dataframe
csv_string_if_filename_none = dataframe.to_csv(
file,
sep="\t",
index=False,
header=self._has_column_names,
lineterminator="\n",
quoting=csv.QUOTE_NONE,
)
return csv_string_if_filename_none
@property
def columns(self) -> list[str]:
"""Returns a list of the column names.
Empty if no column names. The names are the header cells exactly as read, as text (an empty cell is
``""``; an Excel cell or DataFrame label ``1`` is ``"1"``);
a name repeated in the header is suffixed in the second column it names (``trial.1``), as pandas does,
and is listed in ``duplicate_columns``.
Returns:
list: The column names.
"""
columns = []
if self._dataframe is not None and self._has_column_names:
columns = list(self._dataframe.columns)
return columns
[docs]
def column_metadata(self) -> dict[int, "ColumnMetadata"]:
"""Return the metadata for each column.
Returns:
dict[int, ColumnMetadata]: Number/ColumnMetadata pairs.
"""
if self._mapper:
return self._mapper._final_column_map
return {}
[docs]
def set_cell(self, row_number, column_number, new_string_obj, tag_form="short_tag"):
"""Replace the specified cell with transformed text.
Parameters:
row_number (int): The row number of the spreadsheet to set.
column_number (int): The column number of the spreadsheet to set.
new_string_obj (HedString): Object with text to put in the given cell.
tag_form (str): Version of the tags (short_tag, long_tag, base_tag, etc.)
Notes:
Any attribute of a HedTag that returns a string is a valid value of tag_form.
Raises:
ValueError: If there is not a loaded dataframe.
KeyError: If the indicated row/column does not exist.
AttributeError: If the indicated tag_form is not an attribute of HedTag.
"""
if self._dataframe is None:
raise ValueError("No data frame loaded")
new_text = new_string_obj.get_as_form(tag_form)
self._dataframe.iloc[row_number, column_number] = new_text
[docs]
def get_worksheet(self, worksheet_name=None) -> openpyxl.workbook.Workbook | None:
"""Get the requested worksheet.
Parameters:
worksheet_name (str or None): The name of the requested worksheet by name or the first one if None.
Returns:
Union[openpyxl.workbook.Workbook, None]: The workbook request.
Notes:
If None, returns the first worksheet.
Raises:
KeyError: If the specified worksheet name does not exist.
"""
if worksheet_name and self._loaded_workbook:
# return self._loaded_workbook.get_sheet_by_name(worksheet_name)
return self._loaded_workbook[worksheet_name]
elif self._loaded_workbook:
return self._loaded_workbook.worksheets[0]
else:
return None
@staticmethod
def _get_dataframe_from_worksheet(worksheet, has_headers) -> pd.DataFrame:
"""Create a dataframe from the worksheet.
Parameters:
worksheet (Worksheet): The loaded worksheet to convert.
has_headers (bool): True if this worksheet has column headers.
Returns:
pd.DataFrame: The converted data frame.
"""
if has_headers:
data = worksheet.values
# first row is columns
cols = next(data)
data = list(data)
return pd.DataFrame(data, columns=cols, dtype=str)
else:
return pd.DataFrame(worksheet.values, dtype=str)
[docs]
def validate(
self, hed_schema, extra_def_dicts=None, name=None, error_handler=None, validate_sidecar=True, row_offset=None
) -> list[dict]:
"""Validate this file with a SpreadsheetValidator and return all issues.
The validator runs five stages and stops at the first stage that finds an error: the column names
(a padded, blank or repeated name is TABULAR_INVALID), the sidecar, each column's own values, the HED column, and the
assembled rows. See
:class:`~hed.validator.spreadsheet_validator.SpreadsheetValidator`.
Parameters:
hed_schema (HedSchema): The schema to use for validation.
extra_def_dicts (list of DefDict or DefDict): Definitions in addition to the sidecar's.
name (str or None): The name to report errors from this file as. None (the default) uses
this input's own name; an empty string suppresses the FILE_NAME context even when the
input has a name, so a caller can manage the location context itself.
error_handler (ErrorHandler): Error context to use. Creates a new one if None.
validate_sidecar (bool): If False, the sidecar stage is skipped because the caller has
validated the sidecar already.
row_offset (int or None): Added to the 0-based row index in every reported row. None means
1, plus 1 more when the file has a header line, so rows are 1-based file lines.
Returns:
list[dict]: A list of issues for a HED string.
"""
from hed.validator.spreadsheet_validator import SpreadsheetValidator
if name is None:
name = self.name
tab_validator = SpreadsheetValidator(hed_schema)
validation_issues = tab_validator.validate(
self,
sidecar=self.get_sidecar(),
extra_def_dicts=extra_def_dicts,
name=name,
error_handler=error_handler,
row_offset=row_offset,
validate_sidecar=validate_sidecar,
)
return validation_issues
# ------------------------------------------------------------------------------------------
# ColumnSource: what the tabular validator asks of a table
[docs]
def column_names(self) -> list:
"""Return the column names in table order, or the column numbers if the file has no names."""
if self._dataframe is None:
return []
return list(self._dataframe.columns)
[docs]
def distinct_values(self, column_name) -> dict[str, list[int]]:
"""Return the distinct non-missing values of one column, as text, mapped to the 0-based rows holding them.
Parameters:
column_name (str or int): The column, as it appears in ``column_names``.
Returns:
dict[str, list[int]]: Distinct text -> rows. Empty if the column does not exist.
"""
if self._dataframe is None or column_name not in self._dataframe.columns:
return {}
return distinct_values(self._dataframe[column_name])
[docs]
def as_base_input(self):
"""Return self: a BaseInput can always assemble its rows."""
return self
[docs]
def get_sidecar(self):
"""Return the sidecar this file's mapper was built from, or None."""
if self._mapper:
return self._mapper._sidecar
return None
@staticmethod
def _dataframe_has_names(dataframe) -> bool:
for column in dataframe.columns:
if isinstance(column, str):
return True
return False
@staticmethod
def _unique_names(names) -> list:
"""Return the names with each repeat of a non-blank string name suffixed ``.1``, ``.2``, ... as pandas
reads a header. Blank names and labels that are not strings are left as they are."""
taken = set()
unique = []
for name in names:
new_name = name
suffix = 0
while isinstance(name, str) and name and new_name in taken:
suffix += 1
new_name = f"{name}.{suffix}"
taken.add(new_name)
unique.append(new_name)
return unique
def _set_column_names(self, names, header_cells=None):
"""Name the dataframe's columns as read, as text, and record the padded, blank and repeated names.
Each name is converted to text with ``column_name_text`` (an Excel cell or DataFrame label ``1`` or
``1.0`` becomes ``"1"``, None or NaN becomes ``""``) and is otherwise used exactly as read.
Parameters:
names (list): The dataframe's column labels as loaded.
header_cells (list or None): The cells of a text file's header record, used instead of names
because pandas has already renamed a blank cell (``Unnamed: N``) and suffixed a repeat in names.
None uses names as they are (Excel and DataFrame input).
"""
names = list(names)
if header_cells is not None and len(header_cells) == len(names):
names = list(header_cells)
names = [column_name_text(name) for name in names]
self._padded_columns, self._blank_columns, self._duplicate_columns = check_column_names(names)
self._dataframe.columns = self._unique_names(names)
@staticmethod
def _read_header_cells(file):
"""Return the cells of a text file's header record and the file to hand to pandas.
The header is read by pandas as a data row (``header=None``), so its cells are parsed exactly as the
dataframe's header is (blank lines skipped, BOM removed, a double quote kept as an ordinary character)
but are not yet suffixed. A file-like object is returned to where it was; one that cannot seek or
report its position is read into memory first.
Parameters:
file (str, os.PathLike or file-like): Path to the text file or a binary or text file-like object.
Returns:
tuple: The header cells (list[str]) and the file or in-memory copy for pandas.
"""
if isinstance(file, (str, os.PathLike)):
return BaseInput._first_record_cells(file), file
position = None
if hasattr(file, "seekable") and file.seekable():
try:
position = file.tell()
except OSError:
pass # A text file advanced with next() cannot tell its position, but can still be read.
if position is None:
contents = file.read()
file = io.BytesIO(contents) if isinstance(contents, bytes) else io.StringIO(contents)
position = 0
cells = BaseInput._first_record_cells(file)
file.seek(position)
return cells, file
@staticmethod
def _first_record_cells(file) -> list[str]:
"""Return the cells of the first record pandas reads from file, or [] if there is none."""
try:
first = pd.read_csv(
file,
delimiter="\t",
header=None,
nrows=1,
dtype=str,
keep_default_na=False,
skip_blank_lines=True,
quoting=csv.QUOTE_NONE,
)
except pd.errors.EmptyDataError:
return []
return list(first.iloc[0]) if len(first) else []
[docs]
def assemble(self, mapper=None, skip_curly_braces=False) -> pd.DataFrame:
"""Assembles the HED strings.
``dataframe_a``, ``series_a`` and ``series_filtered`` assemble through this method, so the rule below
holds for them too.
Parameters:
mapper (ColumnMapper or None): Generally pass none here unless you want special behavior.
skip_curly_braces (bool): If True, don't plug in curly brace values into columns.
Returns:
pd.Dataframe: The assembled dataframe.
Raises:
HedFileError: BAD_COLUMN_NAMES if a column name is padded, blank or repeated (``padded_columns``,
``blank_columns``, ``duplicate_columns``). Such a table cannot be assembled correctly (a padded
``HED`` column, for example, would be ignored), so it is refused rather than assembled wrongly.
``validate`` reports these names as TABULAR_INVALID and never reaches assembly for them.
"""
self._check_names_for_assembly()
if mapper is None:
mapper = self._mapper
all_columns = self._handle_transforms(mapper)
if skip_curly_braces:
return all_columns
transformers, _ = mapper.get_transformers()
refs = self.get_column_refs()
column_names = list(transformers)
return _handle_curly_braces_refs(all_columns, refs, column_names)
def _check_names_for_assembly(self):
"""Raise BAD_COLUMN_NAMES if a column name is padded, blank or repeated; see ``assemble``."""
if not (self._padded_columns or self._blank_columns or self._duplicate_columns):
return
problems = []
if self._padded_columns:
problems.append(f"padded {self._padded_columns!r}")
if self._blank_columns:
problems.append(f"blank at columns {[position + 1 for position in self._blank_columns]!r}")
if self._duplicate_columns:
problems.append(f"repeated {self._duplicate_columns!r}")
raise HedFileError(
HedExceptions.BAD_COLUMN_NAMES,
f"Cannot assemble rows: column names are {'; '.join(problems)}. "
"Validate the file for the TABULAR_INVALID details.",
self.name,
)
def _handle_transforms(self, mapper) -> pd.DataFrame:
"""Apply transformations to the dataframe using the provided mapper.
Parameters:
mapper: The column mapper object containing transformation functions.
Returns:
pd.DataFrame: The transformed dataframe with all transformations applied.
Notes:
- Handles categorical column conversions before and after transformations
- Returns original dataframe if no transformers are defined
- Categorical columns are temporarily converted to 'category' type for processing
then converted back to 'str' type after transformation
"""
transformers, need_categorical = mapper.get_transformers()
if transformers:
all_columns = self._dataframe
if need_categorical:
all_columns[need_categorical] = all_columns[need_categorical].astype("category")
all_columns = all_columns.transform(transformers)
if need_categorical:
all_columns[need_categorical] = all_columns[need_categorical].astype("str")
else:
all_columns = self._dataframe
return all_columns
[docs]
@staticmethod
def combine_dataframe(dataframe) -> pd.Series:
"""Combine all columns in the given dataframe into a single HED string series,
skipping empty columns and columns with empty strings.
Parameters:
dataframe (pd.Dataframe): The dataframe to combine
Returns:
pd.Series: The assembled series.
"""
dataframe = dataframe.apply(lambda x: ", ".join(filter(lambda e: bool(e) and e != "n/a", map(str, x))), axis=1)
return dataframe
[docs]
def get_def_dict(self, hed_schema, extra_def_dicts=None) -> "DefinitionDict":
"""Return the definition dict for this file.
Note: Baseclass implementation returns just extra_def_dicts.
Parameters:
hed_schema (HedSchema): Identifies tags to find definitions(if needed).
extra_def_dicts (list, DefinitionDict, or None): Extra dicts to add to the list.
Returns:
DefinitionDict: A single definition dict representing all the data(and extra def dicts).
"""
from hed.models.definition_dict import DefinitionDict
return DefinitionDict(extra_def_dicts, hed_schema)
[docs]
def get_column_refs(self) -> list:
"""Return a list of column refs for this file.
Default implementation returns empty list.
Returns:
list: A list of unique column refs found.
"""
return []
def _open_dataframe_file(self, file, has_column_names, input_type):
"""Load data from various file types into the internal DataFrame.
This method handles loading data from different file formats including Excel files,
text files (TSV/CSV), and existing pandas DataFrames. It sets the _dataframe property
and handles appropriate type conversions and error handling for each file type.
Parameters:
file (str, file-like, or pd.DataFrame): The input data source.
- str: File path to load from
- file-like: File object to read from
- pd.DataFrame: Existing DataFrame to use directly
has_column_names (bool): Whether the file contains column headers.
Used to determine pandas header parameter for text files.
input_type (str): File extension indicating the file type.
Supported types: '.xlsx' (Excel), '.tsv', '.txt' (tab-separated text).
Raises:
HedFileError:
- If file is empty or None (FILE_NOT_FOUND)
- If unsupported file extension provided (INVALID_EXTENSION)
- If file loading fails due to format issues (INVALID_FILE_FORMAT)
Notes:
- For DataFrame input: Converts to string type and auto-detects column names
- For Excel files: Loads workbook and converts specified worksheet to DataFrame
- For text files: Uses pandas read_csv with tab delimiter and handles empty files
- All loaded data is converted to string type for consistency
- Missing cells (empty, "n/a", or a DataFrame NaN or None) are stored as "n/a"; any other cell
text, including "nan" or "NULL", is a value
"""
pandas_header = 0 if has_column_names else None
# If file is already a DataFrame
if isinstance(file, pd.DataFrame):
# A missing cell (NaN or None) becomes "n/a", as in a text file. The object step is what makes
# this the same on every pandas: pandas 2 turns a NaN into the text "nan" under astype(str), pandas 3
# keeps it missing.
self._dataframe = file.astype(object).fillna("n/a").astype(str)
self._has_column_names = self._dataframe_has_names(self._dataframe)
if self._has_column_names:
self._set_column_names(self._dataframe.columns)
return
# Check for empty file or None
if not file:
raise HedFileError(HedExceptions.FILE_NOT_FOUND, "Empty file specification passed to BaseInput.", file)
# Handle Excel file input
if input_type in self.EXCEL_EXTENSION:
self._load_excel_file(file, has_column_names)
return
# Handle unsupported file extensions
if input_type not in self.TEXT_EXTENSION:
raise HedFileError(HedExceptions.INVALID_EXTENSION, "Unsupported file extension for text files.", self.name)
# Handle text file input (CSV/TSV)
self._load_text_file(file, pandas_header)
def _load_excel_file(self, file, has_column_names):
"""Load an Excel file into a pandas DataFrame.
This method loads an Excel workbook using openpyxl, retrieves the specified
worksheet (or the first one if none specified), and converts it to a pandas
DataFrame. The loaded workbook is stored for potential later use in saving.
Parameters:
file (str or file-like): Path to the Excel file or file-like object to load.
Must be a valid Excel file format (.xlsx).
has_column_names (bool): Whether the first row of the worksheet contains
column headers that should be used as DataFrame column names.
Raises:
HedFileError: If loading fails due to file format issues, missing file,
corrupted Excel file, or any other openpyxl-related errors.
The original exception is chained for debugging purposes.
Notes:
- Uses openpyxl library for Excel file handling
- Stores the loaded workbook in self._loaded_workbook for later use
- Retrieves worksheet using self._worksheet_name (or first sheet if None)
- Converts worksheet data to DataFrame using _get_dataframe_from_worksheet
- All data is converted to string type for consistency
"""
try:
self._loaded_workbook = openpyxl.load_workbook(file)
loaded_worksheet = self.get_worksheet(self._worksheet_name)
self._dataframe = self._get_dataframe_from_worksheet(loaded_worksheet, has_column_names)
if has_column_names:
self._set_column_names(self._dataframe.columns)
except Exception as e:
raise HedFileError(
HedExceptions.INVALID_FILE_FORMAT, f"Failed to load Excel file: {str(e)}", self.name
) from e
def _load_text_file(self, file, pandas_header):
"""Load a text file (TSV/CSV) into a pandas DataFrame.
This method handles loading tab-separated value files and other text-based
formats using pandas read_csv. It includes special handling for empty files,
proper NaN value replacement, and comprehensive error handling.
Parameters:
file (str or file-like): Path to the text file or file-like object to load.
Can be any format supported by pandas read_csv with tab delimiter.
pandas_header (int or None): Row number to use as column headers.
- 0: First row contains headers
- None: No header row, generate default column names
Raises:
HedFileError: If loading fails due to file format issues, encoding problems,
or any other pandas-related errors. The original exception is chained
for debugging purposes.
Notes:
- Uses tab delimiter for parsing (appropriate for .tsv files)
- Handles empty files by creating an empty DataFrame
- Converts all data to string type for consistency
- Replaces NaN values with "n/a" for consistent handling
- Skips blank lines during parsing
- Treats only an empty cell and "n/a" as missing (na_values ("", "n/a"), keep_default_na=False), so
"NaN", "NULL" and the other pandas default NA strings are values
- Handles pandas.errors.EmptyDataError for files with no data
- With a header, the header record is read first so that blank and repeated names are known before
pandas renames them (``blank_columns``, ``duplicate_columns``); the column names are the header
cells exactly as read
- Double quotes are ordinary characters, as in hed-javascript and the BIDS validator: a line is split
on every tab, a quoted cell keeps its quotes (a quote in a HED string is CHARACTER_INVALID), and a
quoted cell holding a tab is two cells. A row with more cells than the header loses the extra
cells (pandas warns); the BIDS validator reports such a row
"""
if isinstance(file, str) and os.path.exists(file) and os.path.getsize(file) == 0:
self._dataframe = pd.DataFrame() # Handle empty file
return
try:
header_cells = None
if pandas_header is not None:
header_cells, file = self._read_header_cells(file)
self._dataframe = pd.read_csv(
file,
delimiter="\t",
header=pandas_header,
skip_blank_lines=True,
dtype=str,
keep_default_na=False,
na_values=("", "n/a"),
quoting=csv.QUOTE_NONE,
index_col=False,
)
# Only an empty cell and n/a are missing (BIDS). Every other cell text, including nan, NaN, NA,
# None and NULL, which pandas would otherwise read as missing, is a value and is validated as one,
# as it is when a DataFrame is passed in.
self._dataframe = self._dataframe.fillna("n/a")
if header_cells is not None:
self._set_column_names(self._dataframe.columns, header_cells)
except pd.errors.EmptyDataError:
self._dataframe = pd.DataFrame() # Handle case where file has no data
except Exception as e:
raise HedFileError(
HedExceptions.INVALID_FILE_FORMAT, f"Failed to load text file: {str(e)}", self.name
) from e