Skip to content

buckets

tit.opt.ex.buckets

Helpers for reusable exhaustive-search electrode buckets.

Ported from collaborator Larissa Albantakis's branch alba/ex-search-multipolar and generalized: the loader/saver/normalizer accept an arbitrary ordered bucket-key list instead of a hard-coded 4-key scheme, so both 2-pair (:data:BUCKET_KEYS) and 4-pair (:data:tit.opt.mex.logic.MEX_BUCKET_KEYS) bucket files share the same JSON/CSV/TSV round-trip.

normalize_buckets

normalize_buckets(raw: dict, keys: Sequence[str] = BUCKET_KEYS) -> dict[str, list[str]]

Normalize a bucket mapping to canonical bucket keys.

keys defaults to the four 2-pair ex-search bucket keys (:data:BUCKET_KEYS); pass the eight m-ex-search keys (:data:tit.opt.mex.logic.MEX_BUCKET_KEYS) to normalize a 4-pair (8-electrode) bucket mapping through the same function.

Source code in tit/opt/ex/buckets.py
def normalize_buckets(
    raw: dict, keys: Sequence[str] = BUCKET_KEYS
) -> dict[str, list[str]]:
    """Normalize a bucket mapping to canonical bucket keys.

    ``keys`` defaults to the four 2-pair ex-search bucket keys
    (:data:`BUCKET_KEYS`); pass the eight m-ex-search keys
    (:data:`tit.opt.mex.logic.MEX_BUCKET_KEYS`) to normalize a 4-pair
    (8-electrode) bucket mapping through the same function.
    """
    aliases = _aliases_for(keys)
    buckets = {key: [] for key in keys}
    for key, value in raw.items():
        bucket_key = _normalize_bucket_key(str(key), aliases)
        if bucket_key in buckets:
            buckets[bucket_key] = _split_electrodes(value)
    return buckets

load_bucket_file

load_bucket_file(path: str | Path, keys: Sequence[str] = BUCKET_KEYS) -> dict[str, list[str]]

Load bucket definitions from JSON, CSV, or TSV.

JSON may use canonical keys (e1_plus) or GUI-style keys (E1+). Full ex-search/m-ex-search config JSON files with nested electrodes are also accepted. CSV/TSV files should have one row per bucket, with the bucket name in the first column and electrodes in the remaining columns or as a comma/semicolon separated second column. keys selects the bucket scheme -- the four 2-pair keys by default, or an arbitrary key list (e.g. the eight m-ex-search keys) for other electrode-position counts.

Source code in tit/opt/ex/buckets.py
def load_bucket_file(
    path: str | Path, keys: Sequence[str] = BUCKET_KEYS
) -> dict[str, list[str]]:
    """Load bucket definitions from JSON, CSV, or TSV.

    JSON may use canonical keys (``e1_plus``) or GUI-style keys
    (``E1+``). Full ex-search/m-ex-search config JSON files with nested
    ``electrodes`` are also accepted. CSV/TSV files should have one row
    per bucket, with the bucket name in the first column and electrodes
    in the remaining columns or as a comma/semicolon separated second
    column. ``keys`` selects the bucket scheme -- the four 2-pair keys by
    default, or an arbitrary key list (e.g. the eight m-ex-search keys)
    for other electrode-position counts.
    """
    path = Path(path)
    suffix = path.suffix.lower()
    if suffix == ".json":
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):
            raise ValueError("Bucket JSON must contain an object")
        return normalize_buckets(_bucket_json_payload(data), keys)

    delimiter = "\t" if suffix == ".tsv" else ","
    aliases = _aliases_for(keys)
    rows = {}
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.reader(f, delimiter=delimiter)
        for row in reader:
            if not row or not row[0].strip() or row[0].strip().startswith("#"):
                continue
            bucket_key = _normalize_bucket_key(row[0], aliases)
            if bucket_key not in keys:
                continue
            if len(row) == 2:
                rows[bucket_key] = _split_electrodes(row[1])
            else:
                rows[bucket_key] = _split_electrodes(row[1:])
    return normalize_buckets(rows, keys)

save_bucket_file

save_bucket_file(path: str | Path, buckets: dict[str, list[str]], keys: Sequence[str] = BUCKET_KEYS) -> None

Save bucket definitions as JSON.

Source code in tit/opt/ex/buckets.py
def save_bucket_file(
    path: str | Path,
    buckets: dict[str, list[str]],
    keys: Sequence[str] = BUCKET_KEYS,
) -> None:
    """Save bucket definitions as JSON."""
    with open(path, "w", encoding="utf-8") as f:
        json.dump(normalize_buckets(buckets, keys), f, indent=2)
        f.write("\n")

canonical_template_coord_path

canonical_template_coord_path(eeg_net_name: str | Path | None) -> Path | None

Return a canonical 2D EEG-template coordinate file for known net names.

Source code in tit/opt/ex/buckets.py
def canonical_template_coord_path(eeg_net_name: str | Path | None) -> Path | None:
    """Return a canonical 2D EEG-template coordinate file for known net names."""
    if not eeg_net_name:
        return None
    name = Path(str(eeg_net_name)).name
    coord_file = _CANONICAL_TEMPLATE_COORD_FILES.get(name)
    if coord_file is None:
        coord_file = _CANONICAL_TEMPLATE_COORD_FILES.get(Path(name).stem)
    if coord_file is None:
        return None
    path = _RESOURCE_DIR / coord_file
    return path if path.is_file() else None

build_electrode_mirror_map

build_electrode_mirror_map(eeg_csv_path: str | Path, *, midline_tolerance: float = 1e-06) -> dict[str, str]

Map each electrode to its closest left/right mirror in an EEG CSV.

The mirror is defined by reflecting the x-coordinate across the midline while preserving the anterior/posterior coordinate. Midline electrodes are mapped to themselves; downstream channel-pair validation still prevents self-pairs from being evaluated.

Source code in tit/opt/ex/buckets.py
def build_electrode_mirror_map(
    eeg_csv_path: str | Path,
    *,
    midline_tolerance: float = 1e-6,
) -> dict[str, str]:
    """Map each electrode to its closest left/right mirror in an EEG CSV.

    The mirror is defined by reflecting the x-coordinate across the midline
    while preserving the anterior/posterior coordinate. Midline electrodes
    are mapped to themselves; downstream channel-pair validation still
    prevents self-pairs from being evaluated.
    """
    positions = _read_eeg_positions(eeg_csv_path)
    if not positions:
        raise ValueError(f"No electrode positions found in {eeg_csv_path}")

    xs = [coords[0] for coords in positions.values()]
    x_midline = 0.0 if min(xs) < 0 < max(xs) else (min(xs) + max(xs)) / 2

    mirror_map: dict[str, str] = {}
    for label, coords in positions.items():
        x = coords[0]
        if abs(x - x_midline) <= midline_tolerance:
            mirror_map[label] = label
            continue

        target = (2 * x_midline - x, *coords[1:])
        candidates = []
        for other_label, other_coords in positions.items():
            other_x = other_coords[0]
            if other_label == label:
                continue
            if (
                x < x_midline - midline_tolerance
                and other_x < x_midline - midline_tolerance
            ):
                continue
            if (
                x > x_midline + midline_tolerance
                and other_x > x_midline + midline_tolerance
            ):
                continue
            dist2 = sum(
                (other_value - target_value) ** 2
                for other_value, target_value in zip(other_coords, target)
            )
            candidates.append((dist2, other_label))
        if candidates:
            mirror_map[label] = min(candidates)[1]

    return mirror_map

build_quadrant_buckets

build_quadrant_buckets(eeg_csv_path: str | Path) -> dict[str, list[str]]

Split an EEG-position CSV into four RAS quadrants.

Mapping: E1+ = left anterior, E1- = right anterior, E2+ = left posterior, E2- = right posterior.

Source code in tit/opt/ex/buckets.py
def build_quadrant_buckets(eeg_csv_path: str | Path) -> dict[str, list[str]]:
    """Split an EEG-position CSV into four RAS quadrants.

    Mapping: ``E1+`` = left anterior, ``E1-`` = right anterior,
    ``E2+`` = left posterior, ``E2-`` = right posterior.
    """
    buckets = {key: [] for key in BUCKET_KEYS}
    for label, coords in _read_eeg_positions(eeg_csv_path).items():
        x, y = coords[:2]
        if x < 0 and y >= 0:
            buckets["e1_plus"].append(label)
        elif x >= 0 and y >= 0:
            buckets["e1_minus"].append(label)
        elif x < 0 and y < 0:
            buckets["e2_plus"].append(label)
        else:
            buckets["e2_minus"].append(label)

    return {key: sorted(values) for key, values in buckets.items()}