utils

tit.sim.utils

Shared utilities for TI/mTI simulations.

• Montage file I/O (montage_list.json CRUD)
• Montage loading (EEG-cap + flex/freehand)
• Directory setup (BIDS output structure)
• Montage visualization
• Post-processing helpers (field extraction, NIfTI, T1→MNI, file moves)
• Simulation orchestration (sequential + parallel)

ensure_montage_file

ensure_montage_file(project_dir: str) -> str

Return path to montage_list.json, creating it with default schema if absent.

Source code in tit/sim/utils.py

def ensure_montage_file(project_dir: str) -> str:
    """Return path to montage_list.json, creating it with default schema if absent."""
    path = _montage_list_path(project_dir)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    if not os.path.exists(path):
        with open(path, "w") as f:
            json.dump({"nets": {}}, f, indent=4)
    return path

upsert_montage

upsert_montage(*, project_dir: str, eeg_net: str, montage_name: str, electrode_pairs: list[list[str]], mode: str) -> None

mode: 'U' → uni_polar_montages, 'M' → multi_polar_montages

Source code in tit/sim/utils.py

def upsert_montage(
    *,
    project_dir: str,
    eeg_net: str,
    montage_name: str,
    electrode_pairs: list[list[str]],
    mode: str,
) -> None:
    """mode: 'U' → uni_polar_montages, 'M' → multi_polar_montages"""
    data = load_montage_data(project_dir)
    net = data["nets"].setdefault(
        eeg_net, {"uni_polar_montages": {}, "multi_polar_montages": {}}
    )
    key = "uni_polar_montages" if mode.upper() == "U" else "multi_polar_montages"
    net[key][montage_name] = electrode_pairs
    save_montage_data(project_dir, data)

list_montage_names

list_montage_names(project_dir: str, eeg_net: str, *, mode: str) -> list[str]

mode: 'U' or 'M'. Returns [] for missing nets.

Source code in tit/sim/utils.py

def list_montage_names(project_dir: str, eeg_net: str, *, mode: str) -> list[str]:
    """mode: 'U' or 'M'. Returns [] for missing nets."""
    data = load_montage_data(project_dir)
    net = data.get("nets", {}).get(eeg_net, {})
    key = "uni_polar_montages" if mode.upper() == "U" else "multi_polar_montages"
    return sorted(net.get(key, {}).keys())

extract_fields

extract_fields(input_mesh: str, output_dir: str, base_name: str, m2m_dir: str, subject_id: str, logger) -> None

Extract GM (tag 2) and WM (tag 1) meshes from a full-head mesh.

Source code in tit/sim/utils.py

def extract_fields(
    input_mesh: str,
    output_dir: str,
    base_name: str,
    m2m_dir: str,
    subject_id: str,
    logger,
) -> None:
    """Extract GM (tag 2) and WM (tag 1) meshes from a full-head mesh."""
    from simnibs import mesh_io

    full_mesh = mesh_io.read_msh(input_mesh)
    gm_out = os.path.join(output_dir, f"grey_{base_name}.msh")
    wm_out = os.path.join(output_dir, f"white_{base_name}.msh")
    mesh_io.write_msh(full_mesh.crop_mesh(tags=[2]), gm_out)
    mesh_io.write_msh(full_mesh.crop_mesh(tags=[1]), wm_out)

transform_to_nifti

transform_to_nifti(mesh_dir: str, output_dir: str, subject_id: str, m2m_dir: str, logger, fields: list[str] | None = None, skip_patterns: list[str] | None = None) -> None

Convert mesh files to NIfTI (subject + MNI space).

Source code in tit/sim/utils.py

def transform_to_nifti(
    mesh_dir: str,
    output_dir: str,
    subject_id: str,
    m2m_dir: str,
    logger,
    fields: list[str] | None = None,
    skip_patterns: list[str] | None = None,
) -> None:
    """Convert mesh files to NIfTI (subject + MNI space)."""
    from tit.tools.mesh2nii import convert_mesh_dir

    convert_mesh_dir(
        mesh_dir=mesh_dir,
        output_dir=output_dir,
        m2m_dir=m2m_dir,
        fields=fields,
        skip_patterns=skip_patterns,
    )

convert_t1_to_mni

convert_t1_to_mni(m2m_dir: str, subject_id: str, logger) -> None

Convert T1 to MNI space (no-op if already done).

Source code in tit/sim/utils.py

def convert_t1_to_mni(m2m_dir: str, subject_id: str, logger) -> None:
    """Convert T1 to MNI space (no-op if already done)."""
    t1 = os.path.join(m2m_dir, "T1.nii.gz")
    out = os.path.join(m2m_dir, f"T1_{subject_id}")
    result = subprocess.run(
        ["subject2mni", "-i", t1, "-m", m2m_dir, "-o", out],
        capture_output=True,
        text=True,
        timeout=300,
    )
    if result.returncode != 0:
        logger.warning(f"T1 MNI conversion warning: {result.stderr}")

run_simulation

run_simulation(config: SimulationConfig, logger=None, progress_callback: Callable[[int, int, str], None] | None = None) -> list[dict]

Run TI/mTI simulations sequentially. Mode is auto-detected per montage. Montages are read from config.montages. Returns list of result dicts: montage_name, montage_type, status, output_mesh.

Source code in tit/sim/utils.py

def run_simulation(
    config: SimulationConfig,
    logger=None,
    progress_callback: Callable[[int, int, str], None] | None = None,
) -> list[dict]:
    """
    Run TI/mTI simulations sequentially. Mode is auto-detected per montage.
    Montages are read from ``config.montages``.
    Returns list of result dicts: montage_name, montage_type, status, output_mesh.
    """
    if logger is None:
        pm = get_path_manager()
        log_dir = pm.logs(config.subject_id)
        os.makedirs(log_dir, exist_ok=True)
        log_file = os.path.join(
            log_dir, f'Simulator_{time.strftime("%Y%m%d_%H%M%S")}.log'
        )
        logger = _make_file_logger("tit.sim", log_file)

    pm = get_path_manager()
    simulation_dir = pm.simulations(config.subject_id)

    from tit.sim.TI import TISimulation
    from tit.sim.mTI import mTISimulation

    montages = config.montages
    results = []
    total = len(montages)
    for idx, montage in enumerate(montages):
        logger.info(
            f"[{idx+1}/{total}] {montage.simulation_mode.value}: {montage.name}"
        )
        if progress_callback:
            progress_callback(idx, total, montage.name)
        cls = (
            TISimulation
            if montage.simulation_mode == SimulationMode.TI
            else mTISimulation
        )
        results.append(cls(config, montage, logger).run(simulation_dir))
    if progress_callback:
        progress_callback(total, total, "Complete")
    return results