ex

tit.opt.ex

TI Exhaustive Search Module.

ExConfig dataclass

ExConfig(subject_id: str, leadfield_hdf: str, roi_name: str, electrodes: BucketElectrodes | PoolElectrodes, total_current: float = 2.0, current_step: float = 0.5, channel_limit: float | None = None, roi_radius: float = 3.0, roi_names: list[str] | None = None, roi_atlas: list[AtlasROI] | None = None, roi_coordinate_space: Literal['subject', 'mni'] = 'subject', run_name: str | None = None, n_jobs: int = -1, symmetric_bucket: bool = False, symmetry_eeg_csv: str | None = None, symmetry_pairing: str = 'within_pairs')

Full configuration for exhaustive search optimization.

Exhaustive search evaluates every valid electrode combination from a user-defined pool or bucket set, sweeping current amplitudes at discrete steps.

Attributes

subject_id : str Subject identifier matching the m2m directory name. leadfield_hdf : str Filename of the precomputed leadfield HDF5 (e.g. "ernie_leadfield_EEG10-10_UI_Jurak_2007.hdf5"), resolved under the subject's leadfields/ directory (:meth:tit.paths.PathManager.leadfields); an absolute path is also accepted. roi_name : str ROI CSV filename (e.g. "target.csv") in the subject's ROIs/ directory (:meth:tit.paths.PathManager.rois). The ".csv" suffix is appended automatically if missing. Used as the metric-key prefix and (with the net name) the output-directory label. roi_names : list of str or None Optional list of ROI CSV filenames to union into a single target. When provided (combined mode), the spherical masks of every listed ROI are OR-folded into one region. None (default) keeps single-ROI behavior driven by roi_name. An explicit empty list means "no spherical centers at all" -- useful for a purely atlas-driven ROI (roi_atlas only). Each entry gets the ".csv" suffix appended if missing. roi_atlas : list of AtlasROI or None Volumetric atlas or mask ROI(s) to union with the spherical centers from roi_name/roi_names. None (default) keeps the existing spherical-only behavior. roi_coordinate_space : str Space of the roi_name/roi_names CSV centers -- "subject" (default) or "mni". MNI centers are transformed to subject space with simnibs.mni2subject_coords before the search runs. Does not affect roi_atlas, which declares its own atlas_space. electrodes : BucketElectrodes or PoolElectrodes Electrode specification, either a single shared pool (:class:PoolElectrodes) or separate per-channel buckets (:class:BucketElectrodes). A plain dict is auto-converted in __post_init__. total_current : float Total injected current in mA, split across the two channels. Default 2.0. current_step : float Current amplitude step size in mA for the sweep (used only with a channel_limit). Default 0.5. channel_limit : float or None Maximum current per channel in mA. None (default) runs the single balanced split, total_current / 2 on each channel; set a limit above total_current / 2 to sweep splits in current_step increments. roi_radius : float Spherical ROI radius in mm for the target region. Default 3.0. run_name : str or None Optional name for this run. Defaults to a datetime stamp. The run name is the output directory, so a repeated name overwrites the earlier run in place. n_jobs : int Worker processes evaluating candidates in parallel. -1 (default) uses the global CPU limit (Settings; 70 % of the container's cores by default, see :func:tit.cpu.cpu_limit); a larger explicit value is clamped to that limit; 1 evaluates in-process. Results and CSV ordering do not depend on it. symmetric_bucket : bool When True in bucket mode, evaluate only left/right mirrored montages (see :func:tit.opt.ex.buckets.build_electrode_mirror_map). symmetry_eeg_csv : str or None EEG-position CSV used to derive mirrored electrode pairs. If unset, it is inferred from the leadfield's net name. symmetry_pairing : str Symmetry interpretation when symmetric_bucket is True. "within_pairs": each channel's minus electrode is the mirror of its plus electrode (e.g. F7-F8). "cross_pairs": channel 2 is the mirror image of channel 1 (e2+ = mirror(e1+), e2- = mirror(e1-)).

Raises

ValueError If current_step, total_current, channel_limit or roi_radius is not a finite number above 0, if symmetric_bucket is set with pool electrodes, if symmetry_pairing is not "within_pairs"/"cross_pairs", or if roi_coordinate_space is not "subject" or "mni".

Examples

from tit.opt import ExConfig cfg = ExConfig( ... subject_id="ernie", ... leadfield_hdf="ernie_leadfield_EEG10-10_UI_Jurak_2007.hdf5", ... roi_name="L-Insula", # ".csv" is appended ... electrodes=ExConfig.PoolElectrodes( ... electrodes=["Fp1", "Fp2", "C3", "C4", "Cz", "Pz", "T7", "T8"]), ... total_current=2.0, current_step=0.5, channel_limit=1.2, ... roi_coordinate_space="mni", ... ) cfg.roi_name 'L-Insula.csv'

Per-channel buckets instead of one pool:

ExConfig.BucketElectrodes(e1_plus=["F7"], e1_minus=["F8"], ... e2_plus=["P7"], e2_minus=["P8"]).e2_minus ['P8']

Then run_ex_search(cfg) (needs the leadfield HDF5 under the subject's leadfield folder).

See Also

ExResult : Result container returned by :func:~tit.opt.ex.ex.run_ex_search. tit.opt.ex.ex.run_ex_search : Consumes this config.

AtlasROI dataclass

AtlasROI(atlas_path: str, label: int | None = None, atlas_space: Literal['subject', 'mni'] = 'subject')

Volumetric atlas or mask ROI, unioned with the spherical center(s).

Attributes

atlas_path : str Path to a volumetric atlas or mask file -- NIfTI (.nii, .nii.gz) or FreeSurfer (.mgz), e.g. one discovered by :class:tit.atlas.voxel.VoxelAtlasManager. label : int or None Integer label to select within the atlas (elements are included where the voxel value equals label). None treats the whole file as a binary mask (voxel value > 0). atlas_space : str Space of the atlas file, "subject" (default) or "mni". MNI masks are warped to subject space before the search.

Raises

ValueError If atlas_space is not "subject" or "mni".

BucketElectrodes dataclass

BucketElectrodes(e1_plus: list[str], e1_minus: list[str], e2_plus: list[str], e2_minus: list[str])

Separate electrode lists for each bipolar channel position.

Attributes

e1_plus : list of str Candidate electrodes for channel 1 anode. e1_minus : list of str Candidate electrodes for channel 1 cathode. e2_plus : list of str Candidate electrodes for channel 2 anode. e2_minus : list of str Candidate electrodes for channel 2 cathode.

PoolElectrodes dataclass

PoolElectrodes(electrodes: list[str])

Single electrode pool -- all positions draw from the same set.

Attributes

electrodes : list of str List of electrode names available for any channel position.

ExResult dataclass

ExResult(success: bool, output_dir: str, n_combinations: int, results_csv: str | None = None, config_json: str | None = None)

Result from an exhaustive search run.

Attributes

success : bool True if the search completed without error. output_dir : str Absolute path to the output directory. n_combinations : int Total number of electrode/current combinations evaluated. results_csv : str or None Path to the CSV file containing ranked results. None if the run failed before writing results. config_json : str or None Path to the saved configuration JSON. None if the run failed before writing config.

See Also

ExConfig : Configuration consumed by :func:~tit.opt.ex.ex.run_ex_search. tit.opt.ex.ex.run_ex_search : Returns this result.

ExSearchEngine

ExSearchEngine(leadfield_hdf: str, roi_file: str | tuple[str, int] | list[str | tuple[str, int]], roi_name: str, logger: Logger)

Exhaustive TI electrode search engine.

Owns the full pipeline: leadfield loading, ROI resolution and the simulation loop.

Source code in tit/opt/ex/engine.py

def __init__(
    self,
    leadfield_hdf: str,
    roi_file: str | tuple[str, int] | list[str | tuple[str, int]],
    roi_name: str,
    logger: logging.Logger,
):
    self.leadfield_hdf = leadfield_hdf
    self.roi_file = roi_file
    self.roi_name = roi_name
    self.logger = logger

    self.leadfield = None
    self.mesh = None
    self.idx_lf = None
    self.roi_coords = None
    self.roi_centers = None
    self.roi_indices = None
    self.roi_volumes = None
    self.gm_indices = None
    self.gm_volumes = None
    self._eval_subset = None

initialize

initialize(roi_radius: float = 3.0) -> None

Load leadfield, resolve the ROI (CSV/mask/atlas), find ROI + GM elements.

Source code in tit/opt/ex/engine.py

def initialize(self, roi_radius: float = 3.0) -> None:
    """Load leadfield, resolve the ROI (CSV/mask/atlas), find ROI + GM elements."""
    self._load_leadfield()
    self._load_roi_coordinates()
    self._find_roi_elements(roi_radius)
    self._find_gm_elements()
    self._eval_subset = None

compute_ti_fields

compute_ti_fields(e1_plus: str, e1_minus: str, e2_plus: str, e2_minus: str, current_ratios: list[tuple[float, float]]) -> list[dict[str, float]]

ROI metrics of one electrode montage at every current split.

The field is linear in the injected current, so each channel's unit-current field is gathered once and scaled per split -- the same current * (lf[a] - lf[b]) arithmetic TI.get_field performs, restricted to the elements the metrics read.

Source code in tit/opt/ex/engine.py

def compute_ti_fields(
    self,
    e1_plus: str,
    e1_minus: str,
    e2_plus: str,
    e2_minus: str,
    current_ratios: list[tuple[float, float]],
) -> list[dict[str, float]]:
    """ROI metrics of one electrode montage at every current split.

    The field is linear in the injected current, so each channel's
    unit-current field is gathered once and scaled per split -- the
    same ``current * (lf[a] - lf[b])`` arithmetic ``TI.get_field``
    performs, restricted to the elements the metrics read.
    """
    lf = self.leadfield
    idx = self.idx_lf
    subset, roi_pos, gm_pos = self._evaluation_subset()

    unit1 = TI.get_field([e1_plus, e1_minus, 1.0], lf, idx)[subset]
    unit2 = TI.get_field([e2_plus, e2_minus, 1.0], lf, idx)[subset]

    results = []
    for current_ch1_mA, current_ch2_mA in current_ratios:
        ef1 = (current_ch1_mA / 1000) * unit1
        ef2 = (current_ch2_mA / 1000) * unit2
        ti_max = envelope_max([ef1, ef2])
        data = self._roi_metrics(ti_max[roi_pos], ti_max[gm_pos])
        peak = hf_peak([ef1, ef2])
        data.update(
            self._carrier_metrics(
                peak[roi_pos], peak[gm_pos], data[f"{self.roi_name}_Focality"]
            )
        )
        data["current_ch1_mA"] = current_ch1_mA
        data["current_ch2_mA"] = current_ch2_mA
        results.append(data)
    return results

run

run(e1_plus: list[str], e1_minus: list[str], e2_plus: list[str], e2_minus: list[str], current_ratios: list[tuple[float, float]], all_combinations: bool, output_dir: str, n_jobs: int = 1, symmetry_mirror_map: dict[str, str] | None = None, symmetry_pairing: str = 'within_pairs') -> dict[str, dict[str, float]]

Run the full simulation loop. Returns {mesh_key: metrics}.

Candidates are evaluated in enumeration order, one electrode montage (all its current splits) per task, on n_jobs forked workers (n_jobs < 1: the job's CPU budget, see :func:tit.opt.ex.parallel.resolve_n_jobs; 1: in-process). A symmetry_mirror_map (bucket mode) restricts the enumeration to left/right mirrored montages (see :mod:tit.opt.ex.symmetry).

Signals are left to the caller: Ctrl-C raises KeyboardInterrupt and a job cancel's SIGTERM ends the process (tit/jobs/runner.py), so an interrupted search returns nothing and its run folder never gets the files of a finished run.

Source code in tit/opt/ex/engine.py

def run(
    self,
    e1_plus: list[str],
    e1_minus: list[str],
    e2_plus: list[str],
    e2_minus: list[str],
    current_ratios: list[tuple[float, float]],
    all_combinations: bool,
    output_dir: str,
    n_jobs: int = 1,
    symmetry_mirror_map: dict[str, str] | None = None,
    symmetry_pairing: str = "within_pairs",
) -> dict[str, dict[str, float]]:
    """Run the full simulation loop. Returns {mesh_key: metrics}.

    Candidates are evaluated in enumeration order, one electrode
    montage (all its current splits) per task, on ``n_jobs`` forked
    workers (``n_jobs < 1``: the job's CPU budget, see
    :func:`tit.opt.ex.parallel.resolve_n_jobs`; ``1``: in-process).
    A *symmetry_mirror_map* (bucket mode) restricts the enumeration to
    left/right mirrored montages (see :mod:`tit.opt.ex.symmetry`).

    Signals are left to the caller: Ctrl-C raises ``KeyboardInterrupt``
    and a job cancel's SIGTERM ends the process (``tit/jobs/runner.py``),
    so an interrupted search returns nothing and its run folder never
    gets the files of a finished run.
    """
    total = count_combinations(
        e1_plus,
        e1_minus,
        e2_plus,
        e2_minus,
        current_ratios,
        all_combinations,
        symmetry_mirror_map,
        symmetry_pairing,
    )
    n_jobs = resolve_n_jobs(n_jobs)
    self._log_config_summary(
        current_ratios,
        all_combinations,
        total,
        n_jobs,
        symmetry_pairing if symmetry_mirror_map is not None else None,
    )

    results: dict[str, dict[str, float]] = {}
    start_time = time.time()
    ratios = list(current_ratios)

    montages = list(
        _electrode_combinations(
            e1_plus,
            e1_minus,
            e2_plus,
            e2_minus,
            all_combinations,
            symmetry_mirror_map,
            symmetry_pairing,
            both_channel_orders=not ratios_mirror_symmetric(ratios),
        )
    )
    evaluations = evaluate_ordered(
        self,
        "compute_ti_fields",
        ((*montage, ratios) for montage in montages),
        n_jobs,
        n_tasks=len(montages),
    )

    i = 0
    for (ep1, em1, ep2, em2), montage_results in zip(montages, evaluations):
        montage_start = time.time()
        for (ch1, ch2), data in zip(ratios, montage_results):
            i += 1
            name = (
                f"{ep1}_{em1}_and_{ep2}_{em2}"
                f"_I1-{format_current_mA(ch1)}mA_I2-{format_current_mA(ch2)}mA"
            )
            key = f"TI_field_{name}.msh"

            elapsed = time.time() - start_time
            rate = i / elapsed if elapsed > 0 else 0
            eta = (total - i) / rate if rate > 0 else 0

            self.logger.info(f"[{i}/{total}] {name}")
            self.logger.info(
                f"  {100 * i / total:.1f}% | {rate:.2f}/s | ETA {eta / 60:.1f}min"
            )

            data["electrodes"] = (ep1, em1, ep2, em2)
            results[key] = data
            self.logger.info(
                f"  {(time.time() - montage_start) / len(ratios):.2f}s | "
                f"TImax={data[f'{self.roi_name}_TImax_ROI']:.4f} "
                f"TImean={data[f'{self.roi_name}_TImean_ROI']:.4f} "
                f"Foc={data[f'{self.roi_name}_Focality']:.4f}"
            )

    if results:
        t = time.time() - start_time
        self.logger.info(f"\n{'=' * 60}")
        self.logger.info(
            f"Done: {len(results)}/{total} in {t / 60:.1f}min "
            f"({t / len(results):.2f}s each)"
        )
        self.logger.info(f"Output: {output_dir}")

    return results

run_ex_search

run_ex_search(config: ExConfig) -> ExResult

Run an exhaustive two-pair TI search over a precomputed leadfield.

Enumerates every electrode-pair combination allowed by config.electrodes and every current split allowed by total_current/current_step/channel_limit, scores each against the ROI, and writes a ranked CSV plus a config JSON to the run directory under the subject's ex-search/ folder.

Parameters

config : ExConfig Fully specified search configuration (subject, leadfield file, ROI, electrode pool or buckets, current sweep).

Returns

ExResult success, output_dir, n_combinations evaluated, and the results_csv / config_json paths.

Raises

ValueError If no candidate montage can be enumerated (empty pool/buckets or an over-restrictive symmetry rule), an electrode is not in the leadfield (raised before it is loaded), a referenced ROI CSV or atlas file does not exist, or the ROI selects no leadfield mesh element (raised before the search; no results are written). FileNotFoundError If config.leadfield_hdf cannot be found under the subject's leadfields/ directory.

Examples

from tit.opt import ExConfig, run_ex_search cfg = ExConfig( ... subject_id="ernie", ... leadfield_hdf="ernie_leadfield_EEG10-10_UI_Jurak_2007.hdf5", ... roi_name="L-Insula", ... electrodes=ExConfig.PoolElectrodes(electrodes=["Fp1", "Fp2", "C3", "C4"]), ... total_current=2.0, current_step=0.5, channel_limit=1.2, ... run_name="insula_pool", ... ) res = run_ex_search(cfg) # doctest: +SKIP res.success, res.n_combinations, res.results_csv # doctest: +SKIP (True, 18, '.../ex-search/insula_pool/final_output.csv')

See Also

ExConfig : Configuration dataclass for this search. ExResult : Returned container. tit.opt.mex.mex.run_m_ex_search : Four-pair (mTI) variant.

Source code in tit/opt/ex/ex.py

def run_ex_search(config: ExConfig) -> ExResult:
    """Run an exhaustive two-pair TI search over a precomputed leadfield.

    Enumerates every electrode-pair combination allowed by
    ``config.electrodes`` and every current split allowed by
    ``total_current``/``current_step``/``channel_limit``, scores each
    against the ROI, and writes a ranked CSV plus a config JSON to the
    run directory under the subject's ``ex-search/`` folder.

    Parameters
    ----------
    config : ExConfig
        Fully specified search configuration (subject, leadfield file,
        ROI, electrode pool or buckets, current sweep).

    Returns
    -------
    ExResult
        ``success``, ``output_dir``, ``n_combinations`` evaluated, and the
        ``results_csv`` / ``config_json`` paths.

    Raises
    ------
    ValueError
        If no candidate montage can be enumerated (empty pool/buckets or
        an over-restrictive symmetry rule), an electrode is not in the
        leadfield (raised before it is loaded), a referenced ROI CSV or atlas
        file does not exist, or the ROI selects no leadfield mesh element
        (raised before the search; no results are written).
    FileNotFoundError
        If ``config.leadfield_hdf`` cannot be found under the subject's
        ``leadfields/`` directory.

    Examples
    --------
    >>> from tit.opt import ExConfig, run_ex_search
    >>> cfg = ExConfig(
    ...     subject_id="ernie",
    ...     leadfield_hdf="ernie_leadfield_EEG10-10_UI_Jurak_2007.hdf5",
    ...     roi_name="L-Insula",
    ...     electrodes=ExConfig.PoolElectrodes(electrodes=["Fp1", "Fp2", "C3", "C4"]),
    ...     total_current=2.0, current_step=0.5, channel_limit=1.2,
    ...     run_name="insula_pool",
    ... )
    >>> res = run_ex_search(cfg)  # doctest: +SKIP
    >>> res.success, res.n_combinations, res.results_csv  # doctest: +SKIP
    (True, 18, '.../ex-search/insula_pool/final_output.csv')

    See Also
    --------
    ExConfig : Configuration dataclass for this search.
    ExResult : Returned container.
    tit.opt.mex.mex.run_m_ex_search : Four-pair (mTI) variant.
    """
    from tit.telemetry import track_operation
    from tit import constants as const

    with track_operation(const.TELEMETRY_OP_EX_SEARCH):
        return _run_ex_search_inner(config)