fsaverage

tit.source.fsaverage

Project existing simulation field outputs onto an fsaverage template.

Complementary to :mod:tit.source.forward: where the forward pipeline lets a researcher reconstruct EEG sources onto fsaverage, this pipeline puts the stimulation fields on the same grid, so source activity and TI exposure can be compared vertexwise.

For an already-completed TI simulation it reads the pipeline's own central-surface overlays and morphs the requested scalar fields to fsaverage:

• TI_max -- orientation-maximized TI envelope |E| (central overlay)
• TI_normal -- directional TI envelope along the cortical normal
• magnitude -- combined carrier exposure |E1 + E2| on the central surface

Unlike SimNIBS's native map_to_fsavg (which only runs at simulation time and only emits TI_max), this works post-hoc on any finished simulation and on the derived TI_normal / magnitude quantities.

Runs under simnibs_python (reads SimNIBS meshes)::

simnibs_python -m tit.source fsavg_config.json

See Also

tit.source.config.FsavgMapConfig : Parameter object. tit.source.forward.prepare_forward : The companion forward pipeline.

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}"

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