From 97640815baa582d942582ebee87ec6af71714efc Mon Sep 17 00:00:00 2001 From: Daniel McCloy Date: Wed, 12 Aug 2026 17:40:38 -0500 Subject: [PATCH] groundwork --- mne/_fiff/_digitization.py | 4 ++-- mne/_fiff/meas_info.py | 4 ++-- mne/_fiff/proc_history.py | 4 ++-- mne/_fiff/proj.py | 4 ++-- mne/_fiff/write.py | 26 ++++++++++++-------------- mne/annotations.py | 20 ++++++++++---------- mne/cov.py | 4 ++-- 7 files changed, 32 insertions(+), 34 deletions(-) diff --git a/mne/_fiff/_digitization.py b/mne/_fiff/_digitization.py index 9664a4f470e..36e2511d7ec 100644 --- a/mne/_fiff/_digitization.py +++ b/mne/_fiff/_digitization.py @@ -12,7 +12,7 @@ from .constants import FIFF, _coord_frame_named from .tag import read_tag from .tree import dir_tree_find -from .write import _safe_name_list, start_and_end_file, write_dig_points +from .write import _safe_read_name_list, start_and_end_file, write_dig_points _dig_kind_dict = { "cardinal": FIFF.FIFFV_POINT_CARDINAL, @@ -191,7 +191,7 @@ def _read_dig_fif(fid, meas_info, *, return_ch_names=False): coord_frame = _coord_frame_named.get(coord_frame, coord_frame) elif kind == FIFF.FIFF_MNE_CH_NAME_LIST: tag = read_tag(fid, pos) - ch_names = _safe_name_list(tag.data, "read", "ch_names") + ch_names = _safe_read_name_list(tag.data) for d in dig: d["coord_frame"] = coord_frame out = _format_dig_points(dig) diff --git a/mne/_fiff/meas_info.py b/mne/_fiff/meas_info.py index 176091c9a0b..d6742531756 100644 --- a/mne/_fiff/meas_info.py +++ b/mne/_fiff/meas_info.py @@ -85,7 +85,7 @@ from .tree import dir_tree_find from .write import ( DATE_NONE, - _safe_name_list, + _safe_read_name_list, end_block, start_and_end_file, start_block, @@ -2382,7 +2382,7 @@ def _read_bad_channels(fid, node, ch_names_mapping): for node in nodes: tag = find_tag(fid, node, FIFF.FIFF_MNE_CH_NAME_LIST) if tag is not None and tag.data is not None: - bads = _safe_name_list(tag.data, "read", "bads") + bads = _safe_read_name_list(tag.data) bads[:] = _rename_list(bads, ch_names_mapping) return bads diff --git a/mne/_fiff/proc_history.py b/mne/_fiff/proc_history.py index 2b2c753feb9..4987093eb54 100644 --- a/mne/_fiff/proc_history.py +++ b/mne/_fiff/proc_history.py @@ -13,7 +13,7 @@ from .tag import _float_item, _int_item, find_tag from .tree import dir_tree_find from .write import ( - _safe_name_list, + _safe_read_name_list, end_block, start_block, write_float, @@ -208,7 +208,7 @@ def _write_proc_history(fid, info): def _sss_ctc_ch_name_clean(tag_data): """Clean channel names from CTC files.""" - chs = _safe_name_list(tag_data, "read", "ch_names") + chs = _safe_read_name_list(tag_data) # CTC files can have null chars in the last entry, e.g.: # [..., 'MEG2642', 'MEG2643', 'MEG2641\x00 ... \x00'] if len(chs) > 0: diff --git a/mne/_fiff/proj.py b/mne/_fiff/proj.py index cf01df3fdfb..e808f38de06 100644 --- a/mne/_fiff/proj.py +++ b/mne/_fiff/proj.py @@ -24,7 +24,7 @@ from .tag import _rename_list, find_tag from .tree import dir_tree_find from .write import ( - _safe_name_list, + _safe_read_name_list, end_block, start_block, write_float, @@ -609,7 +609,7 @@ def _read_proj(fid, node, *, ch_names_mapping=None, verbose=None): tag = find_tag(fid, item, FIFF.FIFF_PROJ_ITEM_CH_NAME_LIST) if tag is not None: - names = _safe_name_list(tag.data, "read", "names") + names = _safe_read_name_list(tag.data) else: raise ValueError("Projection item channel list missing") diff --git a/mne/_fiff/write.py b/mne/_fiff/write.py index 2bc48d9788c..149bc4e63af 100644 --- a/mne/_fiff/write.py +++ b/mne/_fiff/write.py @@ -7,6 +7,7 @@ import re import time import uuid +from collections.abc import Sequence from contextlib import contextmanager from gzip import GzipFile @@ -151,22 +152,19 @@ def write_name_list(fid, kind, data): def write_name_list_sanitized(fid, kind, lst, *, name="ch_names"): """Write a sanitized, colon-separated list of names.""" - write_string(fid, kind, _safe_name_list(lst, "write", name)) + write_string(fid, kind, _safe_write_name_list(lst, name)) -def _safe_name_list(lst, operation, name): - if operation == "write": - assert isinstance(lst, list | tuple | np.ndarray), type(lst) - if any("{COLON}" in val for val in lst): - raise ValueError(f'The substring "{{COLON}}" in {name} not supported.') - return ":".join(val.replace(":", "{COLON}") for val in lst) - else: - # take a sanitized string and return a list of strings - assert operation == "read" - assert lst is None or isinstance(lst, str) - if not lst: # None or empty string - return [] - return [val.replace("{COLON}", ":") for val in lst.split(":")] +def _safe_write_name_list(lst: Sequence[str], name: str) -> str | list[str]: + if any("{COLON}" in val for val in lst): + raise ValueError(f'The substring "{{COLON}}" in {name} not supported.') + return ":".join(val.replace(":", "{COLON}") for val in lst) + + +def _safe_read_name_list(lst: str | None) -> list[str]: + if not lst: # None or empty string + return [] + return [val.replace("{COLON}", ":") for val in lst.split(":")] def write_float_matrix(fid, kind, mat): diff --git a/mne/annotations.py b/mne/annotations.py index 9a933554153..232bf308cb4 100644 --- a/mne/annotations.py +++ b/mne/annotations.py @@ -21,7 +21,8 @@ from ._fiff.tag import read_tag from ._fiff.tree import dir_tree_find from ._fiff.write import ( - _safe_name_list, + _safe_read_name_list, + _safe_write_name_list, end_block, start_and_end_file, start_block, @@ -202,7 +203,9 @@ def _check_description(description, n): if description.shape == (1,): description = np.repeat(description, n) _check_length(description, n, name="description") - _safe_name_list(description, "write", "description") + # ↓ just ensures no "{COLON}" present in any descriptions; + # ↓ `.tolist()` is done for typing purposes + _safe_write_name_list(description.tolist(), "description") return description @@ -1875,7 +1878,7 @@ def _write_annotations_csv(fname, annot): annot = annot.to_data_frame() if "ch_names" in annot: annot["ch_names"] = [ - _safe_name_list(ch, "write", name=f'annot["ch_names"][{ci}') + _safe_write_name_list(ch, name=f'annot["ch_names"][{ci}') for ci, ch in enumerate(annot["ch_names"]) ] extras_columns = set(annot.columns) - { @@ -1906,7 +1909,7 @@ def _write_annotations_txt(fname, annot): content += ", ch_names" data.append( [ - _safe_name_list(ch, "write", f"annot.ch_names[{ci}]") + _safe_write_name_list(ch, name=f"annot.ch_names[{ci}]") for ci, ch in enumerate(annot.ch_names) ] ) @@ -2089,10 +2092,7 @@ def _read_annotations_csv(fname): description = df["description"].values ch_names = None if "ch_names" in df.columns: - ch_names = [ - _safe_name_list(val, "read", "annotation channel name") - for val in df["ch_names"].values - ] + ch_names = [_safe_read_name_list(val) for val in df["ch_names"].values] extra_columns = list( df.columns.difference(["onset", "duration", "description", "ch_names"]) ) @@ -2241,7 +2241,7 @@ def _read_annotations_txt(fname): desc = [str(d.decode()).strip() for d in np.atleast_1d(desc)] if ch_names is not None: ch_names = [ - _safe_name_list(ch.decode().strip(), "read", f"ch_names[{ci}]") + _safe_read_name_list(ch.decode().strip()) for ci, ch in enumerate(np.atleast_1d(ch_names)) ] @@ -2277,7 +2277,7 @@ def _read_annotations_fif(fid, tree): duration = tag.data duration = list() if duration is None else duration - onset elif kind == FIFF.FIFF_COMMENT: - description = _safe_name_list(tag.data, "read", "description") + description = _safe_read_name_list(tag.data) elif kind == FIFF.FIFF_MEAS_DATE: orig_time = tag.data try: diff --git a/mne/cov.py b/mne/cov.py index 04f197dc4af..dda28c57d18 100644 --- a/mne/cov.py +++ b/mne/cov.py @@ -2441,7 +2441,7 @@ def whiten_evoked( def _read_cov(fid, node, cov_kind, limited=False, verbose=None): """Read a noise covariance matrix.""" # Find all covariance matrices - from ._fiff.write import _safe_name_list + from ._fiff.write import _safe_read_name_list covs = dir_tree_find(node, FIFF.FIFFB_MNE_COV) if len(covs) == 0: @@ -2482,7 +2482,7 @@ def _read_cov(fid, node, cov_kind, limited=False, verbose=None): if tag is None: names = [] else: - names = _safe_name_list(tag.data, "read", "names") + names = _safe_read_name_list(tag.data) if len(names) != dim: raise ValueError( "Number of names does not match covariance matrix dimension"