source

tit.source

EEG source-forward preparation and fsaverage field mapping.

Two pipelines that put EEG source reconstruction and TI stimulation fields on a common cortical grid:

• :func:prepare_forward -- build an MNE-compatible EEG forward solution (leadfield + source space + fsaverage morph) from a SimNIBS head model.
• :func:project_fields_to_fsaverage -- project existing simulation field outputs (TI_max, TI_normal, |E|) onto an fsaverage template.

Both run under the SimNIBS interpreter::

simnibs_python -m tit.source config.json

See Also

tit.source.config : ForwardConfig and FsavgMapConfig dataclasses.

ForwardConfig dataclass

ForwardConfig(eeg_net: str = 'GSN-HydroCel-185', fsaverage_spacing: int = 5, cpus: int = 1, overwrite: bool = False)

Parameters for rebuilding a SimNIBS/MNE EEG forward solution.

Attributes

eeg_net : str EEG cap name without the .csv suffix, as found in the subject's m2m_<id>/eeg_positions/ directory (e.g. "GSN-HydroCel-185"). fsaverage_spacing : int fsaverage subdivision factor (5, 6, or 7) for the morph target. cpus : int SimNIBS FEM workers used while computing the leadfield. overwrite : bool Recompute even when valid outputs already exist. The expensive FEM leadfield is still reused when present unless this is True.

FsavgMapConfig dataclass

FsavgMapConfig(fields: tuple[str, ...] = (lambda: VALID_FSAVG_FIELDS)(), fsaverage_spacing: int = 5, workers: int = 1, overwrite: bool = False)

Parameters for projecting simulation field outputs onto fsaverage.

Attributes

fields : tuple of str Which field quantities to project. Any of :data:VALID_FSAVG_FIELDS ("TI_max", "TI_normal", "magnitude"). fsaverage_spacing : int fsaverage subdivision factor (5, 6, or 7) to morph onto. workers : int Number of subjects projected in parallel (1 = serial). overwrite : bool Re-project even when a cached .npz already exists.

project_fields_to_fsaverage

project_fields_to_fsaverage(subjects: list[tuple[str, str]], cfg: FsavgMapConfig) -> list[tuple[str, str, str]]

Project field outputs to fsaverage for many (subject, simulation) pairs.

Parameters

subjects : list of (str, str) (subject_id, simulation_name) pairs to project. cfg : FsavgMapConfig Field selection, fsaverage spacing, worker count, overwrite flag.

Returns

list of (str, str, str) Per-pair (subject_id, status, message) results.

Source code in tit/source/fsaverage.py

def project_fields_to_fsaverage(
    subjects: list[tuple[str, str]],
    cfg: FsavgMapConfig,
) -> list[tuple[str, str, str]]:
    """Project field outputs to fsaverage for many (subject, simulation) pairs.

    Parameters
    ----------
    subjects : list of (str, str)
        ``(subject_id, simulation_name)`` pairs to project.
    cfg : FsavgMapConfig
        Field selection, fsaverage spacing, worker count, overwrite flag.

    Returns
    -------
    list of (str, str, str)
        Per-pair ``(subject_id, status, message)`` results.
    """
    workers = max(1, min(cfg.workers, len(subjects))) if subjects else 1
    logger.info("Projecting %d simulation(s) with %d worker(s)", len(subjects), workers)

    results: list[tuple[str, str, str]] = []
    if workers == 1:
        for subject_id, sim in subjects:
            res = project_subject(subject_id, sim, cfg)
            _log_result(res)
            results.append(res)
    else:
        with ProcessPoolExecutor(max_workers=workers) as pool:
            futures = {
                pool.submit(project_subject, sid, sim, cfg): (sid, sim)
                for sid, sim in subjects
            }
            for future in as_completed(futures):
                res = future.result()
                _log_result(res)
                results.append(res)
    return results

project_subject

project_subject(subject_id: str, sim: str, cfg: FsavgMapConfig) -> tuple[str, str, str]

Project one (subject, simulation) and cache the result as .npz.

Returns (subject_id, status, message) where status is one of {"ok", "cached", "failed"} so batch runs can record and continue.

Source code in tit/source/fsaverage.py

def project_subject(
    subject_id: str,
    sim: str,
    cfg: FsavgMapConfig,
) -> tuple[str, str, str]:
    """Project one (subject, simulation) and cache the result as ``.npz``.

    Returns ``(subject_id, status, message)`` where status is one of
    ``{"ok", "cached", "failed"}`` so batch runs can record and continue.
    """
    pm = get_path_manager()
    out_path = _output_path(pm, subject_id, sim, cfg.fsaverage_spacing)
    if out_path.exists() and not cfg.overwrite:
        return subject_id, "cached", out_path.name
    try:
        maps = _compute_fields(pm, subject_id, sim, cfg)
    except Exception as exc:  # noqa: BLE001 - record per-subject and continue
        return subject_id, "failed", repr(exc)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    np.savez_compressed(out_path, subject_id=subject_id, simulation=sim, **maps)
    medians = ", ".join(f"{k} med={np.median(v):.3f}" for k, v in maps.items())
    return subject_id, "ok", f"{medians} -> {out_path.name}"

prepare_forward

prepare_forward(subject_id: str, cfg: ForwardConfig, *, output_dir: str | Path | None = None) -> tuple[Path, Path, Path]

Rebuild one subject's forward, source-space, and fsaverage morph files.

Parameters

subject_id : str Subject identifier without the sub- prefix (e.g. "101"). cfg : ForwardConfig EEG net, fsaverage spacing, FEM worker count, and overwrite flag. output_dir : str or pathlib.Path or None Destination directory. Defaults to pm.forward(subject_id) (derivatives/SimNIBS/sub-<id>/forward/).

Returns

tuple of pathlib.Path (fwd_path, src_path, morph_path).

Source code in tit/source/forward.py

def prepare_forward(
    subject_id: str,
    cfg: ForwardConfig,
    *,
    output_dir: str | Path | None = None,
) -> tuple[Path, Path, Path]:
    """Rebuild one subject's forward, source-space, and fsaverage morph files.

    Parameters
    ----------
    subject_id : str
        Subject identifier without the ``sub-`` prefix (e.g. ``"101"``).
    cfg : ForwardConfig
        EEG net, fsaverage spacing, FEM worker count, and overwrite flag.
    output_dir : str or pathlib.Path or None
        Destination directory.  Defaults to ``pm.forward(subject_id)``
        (``derivatives/SimNIBS/sub-<id>/forward/``).

    Returns
    -------
    tuple of pathlib.Path
        ``(fwd_path, src_path, morph_path)``.
    """
    pm = get_path_manager()
    m2m_dir = Path(pm.m2m(subject_id))
    if not m2m_dir.is_dir():
        raise FileNotFoundError(f"m2m directory not found: {m2m_dir}")

    forward_dir = (
        Path(output_dir) if output_dir is not None else Path(pm.forward(subject_id))
    )
    forward_dir.mkdir(parents=True, exist_ok=True)

    stem = f"sub-{subject_id}_net-{cfg.eeg_net}"
    expected = (
        forward_dir / f"{stem}-fwd.fif",
        forward_dir / f"{stem}-src.fif",
        forward_dir / f"{stem}-morph.h5",
    )
    if not cfg.overwrite and _forward_outputs_valid(*expected):
        logger.info("Forward outputs already present for %s; skipping.", subject_id)
        return expected

    electrodes, fiducials = _read_simnibs_montage(m2m_dir, cfg.eeg_net)
    montage_info, trans = _build_montage_info(electrodes, fiducials)

    info_path = forward_dir / f"{stem}-info.fif"
    trans_path = forward_dir / f"{stem}-trans.fif"
    mne.io.RawArray(
        np.zeros((len(montage_info.ch_names), 1)), montage_info, verbose=False
    ).save(str(info_path), overwrite=True, verbose=False)
    mne.write_trans(str(trans_path), trans, overwrite=True)

    # Redundancy guard: reuse the FEM leadfield if one is already cached.
    leadfield_hdf5 = None if cfg.overwrite else _find_existing_leadfield(forward_dir)
    if leadfield_hdf5 is not None:
        logger.info("Reusing cached leadfield: %s", leadfield_hdf5.name)
    else:
        eeg_csv = m2m_dir / "eeg_positions" / f"{cfg.eeg_net}.csv"
        leadfield_hdf5 = _compute_leadfield(m2m_dir, forward_dir, eeg_csv, cfg.cpus)

    _run(
        [
            "prepare_eeg_forward",
            "mne",
            str(m2m_dir),
            str(leadfield_hdf5),
            str(info_path),
            str(trans_path),
            "--fsaverage",
            str(cfg.fsaverage_spacing),
        ],
        f"prepare_eeg_forward for sub-{subject_id}",
        cwd=forward_dir,
    )
    return _rename_generated_outputs(forward_dir, stem)