catalog

tit.catalog

Project catalog: subject and simulation discovery for the UI.

Built only on :class:tit.paths.PathManager (plus the existing per-domain helpers it composes: :mod:tit.sim.utils montage I/O, :mod:tit.atlas discovery, :mod:tit.opt.ex.roi/:mod:tit.opt.leadfield ROI and leadfield helpers, :mod:tit.opt.flex.manifest); the server routes (tit.server.routes.catalog, tit.server.routes.catalog_v1) serialise these dicts as-is, so the BIDS / derivative layout rules are never re-implemented outside tit.

v1 additions (this module is extended, not replaced, for Stage 1 / lane B2 — see docs/dev/DECISIONS.md § 2026-08-27 (Electron and the job server) §3 "Catalog"): subject/simulation detail, montage CRUD, EEG nets, atlases + regions, ROI CRUD, leadfields, flex/ex/mex runs, analyses, reports, freehand configs, the project-level group catalog, Quick Notes, and the subject-info presence matrix. A run/analysis directory without the manifest file that marks it complete (flex_meta.json, run_config.json, or the analysis.json+results.csv pair) is ignored, so a cancelled or in-progress run never appears as a result.

subject_ids

subject_ids(pm: PathManager) -> list[str]

Union of raw-BIDS, FastSurfer, legacy-FreeSurfer and m2m subjects.

Naturally sorted. derivatives/freesurfer is still unioned in so a project processed before FastSurfer replaced recon-all keeps listing its subjects.

Deliberately excludes subjects known only from sourcedata/ (DICOMs staged, nothing converted yet) -- this is the onboarded set every other catalog function gates on (montages, ROIs, EEG nets, ...), where a subject with no derivative of any kind would be meaningless. See :func:sourcedata_only_subject_ids and, for the two routes that do need to see a not-yet-onboarded subject, :func:list_subjects / :func:subject_detail.

Source code in tit/catalog.py

def subject_ids(pm: PathManager) -> list[str]:
    """Union of raw-BIDS, FastSurfer, legacy-FreeSurfer and m2m subjects.

    Naturally sorted. ``derivatives/freesurfer`` is still unioned in so a
    project processed before FastSurfer replaced ``recon-all`` keeps listing
    its subjects.

    Deliberately excludes subjects known only from ``sourcedata/`` (DICOMs
    staged, nothing converted yet) -- this is the *onboarded* set every other
    catalog function gates on (montages, ROIs, EEG nets, ...), where a
    subject with no derivative of any kind would be meaningless. See
    :func:`sourcedata_only_subject_ids` and, for the two routes that do need
    to see a not-yet-onboarded subject, :func:`list_subjects` /
    :func:`subject_detail`.
    """
    if not pm.project_dir:
        return []
    ids = set()
    for root, needs_m2m in (
        (pm.project_dir, False),
        (pm.fastsurfer(), False),
        (pm.freesurfer(), False),
        (pm.simnibs(), True),
    ):
        if not root:
            continue
        for name in _metadata_names(root, pm.project_dir):
            sid = name.removeprefix("sub-")
            if not name.startswith("sub-") or not is_valid_subject_id(sid):
                continue
            if os.path.isdir(os.path.join(root, name)) and (
                not needs_m2m or _project_isdir(pm, pm.m2m(sid))
            ):
                ids.add(sid)
    return sorted(ids, key=natural_key)

sourcedata_only_subject_ids

sourcedata_only_subject_ids(pm: PathManager) -> list[str]

Subject ids with raw data staged under sourcedata/ and nowhere else.

:func:subject_ids would never mention these -- no BIDS directory, no m2m, no FastSurfer/FreeSurfer -- so a project's own newly-arrived DICOMs (Dataset 000's sub-102, before its DICOM-conversion stage has ever run) were invisible to every page built on the catalog (lane FX5, docs/dev/DECISIONS.md § 2026-09-03 (One Docker image and a real development loop)). Naturally sorted; a subject that already appears in :func:subject_ids is never repeated here even if sourcedata/ also holds a copy of its DICOMs.

Source code in tit/catalog.py

def sourcedata_only_subject_ids(pm: PathManager) -> list[str]:
    """Subject ids with raw data staged under ``sourcedata/`` and nowhere else.

    :func:`subject_ids` would never mention these -- no BIDS directory, no
    m2m, no FastSurfer/FreeSurfer -- so a project's own newly-arrived DICOMs
    (Dataset 000's ``sub-102``, before its DICOM-conversion stage has ever
    run) were invisible to every page built on the catalog (lane FX5,
    ``docs/dev/DECISIONS.md § 2026-09-03 (One Docker image and a real development loop)``). Naturally sorted; a subject that
    already appears in :func:`subject_ids` is never repeated here even if
    ``sourcedata/`` also holds a copy of its DICOMs.
    """
    if not pm.project_dir:
        return []
    try:
        entries = _metadata_names(pm.sourcedata(), pm.project_dir)
    except OSError:
        return []
    known = set(subject_ids(pm))
    ids = []
    for name in entries:
        if not name.startswith("sub-"):
            continue
        sid = name[len("sub-") :]
        if not is_valid_subject_id(sid) or sid in known:
            continue
        if _has_sourcedata_raw(pm, sid):
            ids.append(sid)
    return sorted(ids, key=natural_key)

list_subjects

list_subjects(pm: PathManager) -> list[dict]

One entry per subject: presence flags and the simulation count.

Also lists subjects known only from sourcedata/ (DICOMs staged, no BIDS directory yet -- see :func:sourcedata_only_subject_ids): the Subjects page and Pre-processing's own subjects table are exactly where a project's newest subject needs to be plannable, before anything else about them exists. An entry carries has_sourcedata: True when staged raw data is present (so a client can tell "staged, not yet converted" from has_raw False); the key is left out entirely otherwise -- the overwhelming common case (a subject with no sourcedata copy at all), and the shape every existing caller/test built before this field existed stays byte-identical (/api/catalog/subjects is served with response_model_exclude_unset, tit/server/routes/catalog.py).

Source code in tit/catalog.py

def list_subjects(pm: PathManager) -> list[dict]:
    """One entry per subject: presence flags and the simulation count.

    Also lists subjects known only from ``sourcedata/`` (DICOMs staged, no
    BIDS directory yet -- see :func:`sourcedata_only_subject_ids`): the
    Subjects page and Pre-processing's own subjects table are exactly where
    a project's newest subject needs to be plannable, before anything else
    about them exists. An entry carries ``has_sourcedata: True`` when staged
    raw data is present (so a client can tell "staged, not yet converted"
    from ``has_raw`` False); the key is left out entirely otherwise -- the
    overwhelming common case (a subject with no sourcedata copy at all), and
    the shape every existing caller/test built before this field existed
    stays byte-identical (``/api/catalog/subjects`` is served with
    ``response_model_exclude_unset``, ``tit/server/routes/catalog.py``).
    """
    ids = sorted(
        set(subject_ids(pm)) | set(sourcedata_only_subject_ids(pm)), key=natural_key
    )
    rows = []
    for sid in ids:
        row = {
            "id": sid,
            "has_raw": _project_isdir(pm, pm.bids_subject(sid)),
            "has_fastsurfer": _project_isdir(pm, pm.fastsurfer_subject(sid)),
            "has_freesurfer": _project_isdir(pm, pm.freesurfer_subject(sid)),
            "has_m2m": _project_isdir(pm, pm.m2m(sid)),
            "n_simulations": len(_simulation_names(pm, sid)),
        }
        if _has_sourcedata_raw(pm, sid):
            row["has_sourcedata"] = True
        rows.append(row)
    return rows

list_simulations

list_simulations(pm: PathManager, sid: str) -> list[dict] | None

Simulations of sid (TI/ and mTI/ presence), None if unknown.

Source code in tit/catalog.py

def list_simulations(pm: PathManager, sid: str) -> list[dict] | None:
    """Simulations of *sid* (``TI/`` and ``mTI/`` presence), ``None`` if unknown."""
    if sid not in subject_ids(pm):
        return None
    items: list[dict] = []
    for name in _simulation_names(pm, sid):
        item = {
            "name": name,
            "path": pm.simulation(sid, name),
            "has_ti": _project_isdir(pm, os.path.dirname(pm.ti_mesh_dir(sid, name))),
            "has_mti": _project_isdir(pm, os.path.dirname(pm.mti_mesh_dir(sid, name))),
        }
        # The v1 contract enriches every list item with the SimulationDetail keys (fields,
        # montages, niftis, meshes, ...); the v0 keys above always win. Found on real data:
        # the Results page crashed on ``fields`` missing from the list shape.
        try:
            detail = simulation_detail(pm, sid, name) or {}
        except (
            Exception
        ):  # pragma: no cover - a broken run dir must not hide the others
            detail = {}
        for key, value in detail.items():
            item.setdefault(key, value)
        items.append(item)
    return items

output_files

output_files(root: str, *, project_root: str | None = None) -> list[dict]

Every file under a run's output folder as {path, kind, label, bytes} rows.

The one filesystem read behind a finished job's Artifacts tab (GET /api/jobs/{id}/artifacts): what is on disk, not what a runner remembered to register. label is the path relative to root; bytes is the file size (None if it cannot be stat'ed). Same bounded walk and project jail as :func:_dir_artifacts.

Source code in tit/catalog.py

def output_files(root: str, *, project_root: str | None = None) -> list[dict]:
    """Every file under a run's output folder as ``{path, kind, label, bytes}`` rows.

    The one filesystem read behind a finished job's Artifacts tab
    (``GET /api/jobs/{id}/artifacts``): what is on disk, not what a runner remembered to
    register. ``label`` is the path relative to *root*; ``bytes`` is the file size (``None``
    if it cannot be stat'ed). Same bounded walk and project jail as :func:`_dir_artifacts`.
    """
    out: list[dict] = []
    for path in _walk_files(root, project_root=project_root):
        name = os.path.basename(path)
        lowered = name.lower()
        if lowered.endswith(".tetravox.json"):
            kind = "scene"
        elif lowered.endswith(".nii.gz"):
            kind = "nifti"
        else:
            kind = _OUTPUT_KIND_BY_EXT.get(os.path.splitext(lowered)[1], "file")
        try:
            size: int | None = os.path.getsize(path)
        except OSError:
            size = None
        out.append(
            {
                "path": path,
                "kind": kind,
                "label": os.path.relpath(path, root),
                "bytes": size,
            }
        )
    return out

classify_view_file

classify_view_file(path: str) -> str | None

What path is, as one of the seven kinds a scene understands — or None.

volume · label-volume · surface · mesh · annotation · morph · surface-data.

Decided from the name and its neighbours only. No file is opened: this runs once per row of a menu that is redrawn on every keystroke, and a menu that reads its way through a FreeSurfer surf/ directory is a menu that stutters. The one filesystem touch is :func:_has_lut_sidecar, a single os.path.isfile on a sibling.

None means "not something a scene can use" and is returned for everything unrecognised — .mat, .txt, .geo, .sigma, .label, .ctab, a log, and the derived FreeSurfer files (lh.pial.T1, lh.smoothwm.K.crv) that are recon-all's bookkeeping rather than anyone's input. Refusing to guess is the point: a file offered under the wrong kind fails at Open, which is a much worse moment to find out than not being offered at all.

Source code in tit/catalog.py

def classify_view_file(path: str) -> str | None:
    """What *path* is, as one of the seven kinds a scene understands — or ``None``.

    ``volume`` · ``label-volume`` · ``surface`` · ``mesh`` · ``annotation`` · ``morph`` ·
    ``surface-data``.

    Decided from the **name and its neighbours only**. No file is opened: this runs once per row
    of a menu that is redrawn on every keystroke, and a menu that reads its way through a
    FreeSurfer `surf/` directory is a menu that stutters. The one filesystem touch is
    :func:`_has_lut_sidecar`, a single `os.path.isfile` on a sibling.

    ``None`` means "not something a scene can use" and is returned for everything unrecognised —
    `.mat`, `.txt`, `.geo`, `.sigma`, `.label`, `.ctab`, a log, and the derived FreeSurfer files
    (`lh.pial.T1`, `lh.smoothwm.K.crv`) that are recon-all's bookkeeping rather than anyone's
    input. Refusing to guess is the point: a file offered under the wrong kind fails at Open,
    which is a much worse moment to find out than not being offered at all.
    """
    basename = os.path.basename(path)
    lowered = basename.lower()

    # A tetrahedral FEM mesh. The only extension that earns the word.
    if lowered.endswith(".msh"):
        return "mesh"

    if lowered.endswith(".annot"):
        return "annotation"

    if lowered.endswith(".gii"):
        # GIfTI says what it carries in its own second-to-last suffix. `.func`/`.shape`/`.time`
        # are per-vertex numbers *for* a surface; `.surf` and a bare `.gii` are the geometry.
        # SimNIBS writes the geometry bare (`lh.central.gii`), which is why the bare case is
        # geometry and not the other way round.
        if re.search(r"\.(func|shape|time)\.gii$", lowered):
            return "surface-data"
        return "surface"

    if lowered.endswith(_TRIANGLE_EXTS):
        return "surface"

    if lowered.endswith(_VOLUME_EXTS):
        stem = _strip_view_ext(basename)
        if _has_lut_sidecar(path) or any(
            pattern.match(stem) for pattern in _LABEL_VOLUME_PATTERNS
        ):
            return "label-volume"
        return "volume"

    # FreeSurfer's extensionless binaries, which is where the distinction is easiest to get wrong:
    # `lh.pial` and `lh.thickness` look identical to a filename matcher that only splits on dots.
    hemi = _hemi_split(basename)
    if hemi is not None:
        _, rest = hemi
        if rest in _FS_SURFACE_STEMS:
            return "surface"
        if rest in _FS_MORPH_STEMS:
            return "morph"

    return None

surface_attachments

surface_attachments(surface_path: str, *, extra_dirs: tuple[str, ...] = (), project_root: str | None = None) -> list[str]

Every annotation / morph / data-GIfTI file that belongs on surface_path.

Matched by hemisphere and nothing else, because that is the only correspondence FreeSurfer actually promises: lh.thickness has one value per vertex of every lh.* surface, since they all share a vertex numbering. Which lh. surface you hang it on is the viewer's choice, not a fact about the file. (Tetravox's own worker checks the vertex count when the file is attached, so a genuine mismatch is refused there with both counts named — this side does not need to read a byte to be safe, only to be plausible.)

Searched in the surface's own directory plus extra_dirs — SimNIBS keeps the geometry in m2m_<sid>/surfaces/ but writes the parcellations it made into m2m_<sid>/segmentation/, two directories apart, which is exactly why nothing offered them before.

Source code in tit/catalog.py

def surface_attachments(
    surface_path: str,
    *,
    extra_dirs: tuple[str, ...] = (),
    project_root: str | None = None,
) -> list[str]:
    """Every annotation / morph / data-GIfTI file that belongs on *surface_path*.

    Matched by hemisphere and nothing else, because that is the only correspondence FreeSurfer
    actually promises: `lh.thickness` has one value per vertex of *every* `lh.*` surface, since
    they all share a vertex numbering. Which `lh.` surface you hang it on is the viewer's choice,
    not a fact about the file. (Tetravox's own worker checks the vertex count when the file is
    attached, so a genuine mismatch is refused there with both counts named — this side does not
    need to read a byte to be safe, only to be plausible.)

    Searched in the surface's own directory plus *extra_dirs* — SimNIBS keeps the geometry in
    `m2m_<sid>/surfaces/` but writes the parcellations it made into `m2m_<sid>/segmentation/`,
    two directories apart, which is exactly why nothing offered them before.
    """
    if project_root and not is_within(project_root, surface_path):
        return []
    hemi = _hemi_split(os.path.basename(surface_path))
    if hemi is None:
        return []
    prefix = f"{hemi[0]}."
    found: list[str] = []
    for directory in (os.path.dirname(surface_path), *extra_dirs):
        for name in _metadata_names(directory, project_root):
            if not name.startswith(prefix):
                continue
            candidate = os.path.join(directory, name)
            if not os.path.isfile(candidate):
                continue
            if classify_view_file(candidate) in VIEW_ATTACHMENT_KINDS:
                found.append(candidate)
    return found

subject_detail

subject_detail(pm: PathManager, sid: str) -> dict | None

Full detail for one subject, None if unknown.

"Unknown" also admits a sourcedata-only subject (see :func:sourcedata_only_subject_ids) -- the Subjects page fetches this for every row :func:list_subjects returns, sourcedata-only rows included, and a 404 there would just blank the detail pane rather than error, but there is real information to return (has_sourcedata) so it is returned rather than dropped.

Source code in tit/catalog.py

def subject_detail(pm: PathManager, sid: str) -> dict | None:
    """Full detail for one subject, ``None`` if unknown.

    "Unknown" also admits a sourcedata-only subject (see
    :func:`sourcedata_only_subject_ids`) -- the Subjects page fetches this
    for every row :func:`list_subjects` returns, sourcedata-only rows
    included, and a 404 there would just blank the detail pane rather than
    error, but there is real information to return (``has_sourcedata``) so
    it is returned rather than dropped.
    """
    if sid not in subject_ids(pm) and sid not in sourcedata_only_subject_ids(pm):
        return None
    has_m2m = _project_isdir(pm, pm.m2m(sid))
    # Only caps that carry electrodes: see :func:`_eeg_caps_with_electrodes`.
    eeg_nets_list = (
        [n for n, _ in _eeg_caps_with_electrodes(pm, sid)] if has_m2m else []
    )
    has_leadfields = (
        sorted({f"{item['net']}.csv" for item in (list_leadfields(pm, sid) or [])})
        if has_m2m
        else []
    )
    return {
        "id": sid,
        "has_raw": _project_isdir(pm, pm.bids_subject(sid)),
        "has_fastsurfer": _project_isdir(pm, pm.fastsurfer_subject(sid)),
        "has_freesurfer": _project_isdir(pm, pm.freesurfer_subject(sid)),
        "has_m2m": has_m2m,
        "n_simulations": len(_simulation_names(pm, sid)),
        "m2m_path": pm.m2m(sid) if has_m2m else None,
        "eeg_nets": eeg_nets_list,
        "has_leadfields": has_leadfields,
        "has_dwi": _project_isdir(pm, pm.bids_dwi(sid)),
        "has_ct": _has_ct(pm, sid),
        "has_sourcedata": _has_sourcedata_raw(pm, sid),
    }

simulation_detail

simulation_detail(pm: PathManager, sid: str, sim: str) -> dict | None

Full detail for one simulation, None if unknown.

Source code in tit/catalog.py

def simulation_detail(pm: PathManager, sid: str, sim: str) -> dict | None:
    """Full detail for one simulation, ``None`` if unknown."""
    if sid not in subject_ids(pm) or sim not in _simulation_names(pm, sid):
        return None
    sim_dir = pm.simulation(sid, sim)
    if not _project_paths_safe(pm, sim_dir):
        return None
    has_ti = _project_isdir(pm, os.path.dirname(pm.ti_mesh_dir(sid, sim)))
    has_mti = _project_isdir(pm, os.path.dirname(pm.mti_mesh_dir(sid, sim)))
    config_path = os.path.join(sim_dir, "documentation", "config.json")
    config = (
        _read_json(config_path) if _project_paths_safe(pm, config_path) else None
    ) or {}
    montage_name = config.get("montage_name") or sim

    niftis: list[dict] = []
    meshes: list[dict] = []
    for mode in _MODE_DIRS:
        mode_dir = os.path.join(sim_dir, mode)
        if not _project_isdir(pm, mode_dir):
            continue
        niftis.extend(_mode_niftis(mode_dir, project_root=pm.project_dir))
        meshes.extend(
            _dir_meshes(
                os.path.join(mode_dir, "mesh"), "field", project_root=pm.project_dir
            )
        )
        meshes.extend(
            _dir_meshes(
                os.path.join(mode_dir, "mesh", "surfaces"),
                "surface",
                project_root=pm.project_dir,
            )
        )
    hf_dir = os.path.join(sim_dir, "high_Frequency")
    if _project_isdir(pm, os.path.join(hf_dir, "niftis")):
        niftis.extend(
            _hf_niftis(os.path.join(hf_dir, "niftis"), project_root=pm.project_dir)
        )
    if _project_isdir(pm, os.path.join(hf_dir, "mesh")):
        meshes.extend(
            _dir_meshes(
                os.path.join(hf_dir, "mesh"),
                "high_frequency",
                project_root=pm.project_dir,
            )
        )

    fields = sorted({n["field"] for n in niftis if n["field"]})
    spaces = sorted({n["space"] for n in niftis})
    all_reports = reports(pm, sid) or []
    return {
        "name": sim,
        "path": sim_dir,
        "has_ti": has_ti,
        "has_mti": has_mti,
        "montages": [montage_name],
        "fields": fields,
        "space": spaces,
        "report_ids": _report_ids_for_simulation(all_reports, sim),
        "niftis": niftis,
        "meshes": meshes,
    }

simulation_figures

simulation_figures(pm: PathManager, sid: str, sim: str) -> list[dict] | None

Pictures a simulation run saved of itself; None if subject/simulation is unknown.

Today that is the montage visualisation tit.tools.montage_visualizer writes as <sim>/<TI|mTI>/montage_imgs/<name>_highlighted_visualization.png -- the EEG net with this run's electrodes highlighted. The Results pane shows it beside the channel chips, which name the same montage in text, and in the run's Figures grid.

Not folded into SimulationDetail: that response has a Pydantic response_model generated from the frozen v1 contract, so a new field there is a contract change. This is the same shape as :func:electrode_overlays above -- a small presence query beside the main read.

Source code in tit/catalog.py

def simulation_figures(pm: PathManager, sid: str, sim: str) -> list[dict] | None:
    """Pictures a simulation run saved of itself; ``None`` if subject/simulation is unknown.

    Today that is the montage visualisation ``tit.tools.montage_visualizer`` writes as
    ``<sim>/<TI|mTI>/montage_imgs/<name>_highlighted_visualization.png`` -- the EEG net with
    this run's electrodes highlighted. The Results pane shows it beside the channel chips,
    which name the same montage in text, and in the run's Figures grid.

    Not folded into ``SimulationDetail``: that response has a Pydantic ``response_model``
    generated from the frozen v1 contract, so a new field there is a contract change. This is
    the same shape as :func:`electrode_overlays` above -- a small presence query beside the
    main read.
    """
    if sid not in subject_ids(pm) or sim not in _simulation_names(pm, sid):
        return None
    sim_dir = pm.simulation(sid, sim)
    out: list[dict] = []
    for mode in _MODE_DIRS:
        path = _jailed(
            pm,
            os.path.join(
                sim_dir, mode, "montage_imgs", f"{sim}_highlighted_visualization.png"
            ),
        )
        if path and os.path.isfile(path):
            out.append(
                {
                    "path": path,
                    "kind": "image",
                    "label": f"{mode} montage",
                }
            )
    return out

electrode_overlays

electrode_overlays(pm: PathManager, sid: str, sim: str) -> list[dict] | None

Electrode-overlay NIfTI presence for one simulation, per TI/mTI mode.

None if the subject/simulation is unknown. This is the listing half of the electrode-overlay v1 gap documented in tit/server/routes/viewers.py and pages/viewer/PARITY.md #1: creating the overlay is a tools job (tit.tools.electrode_overlay), and once it exists on disk, :func:tit.viewspec._electrode_overlay_layer already picks it up for kind=simulation ViewSpecs -- this endpoint just tells the Viewer page whether that file is there yet (and its path), without a separate job kind or catalog change for the overlay-building side.

Source code in tit/catalog.py

def electrode_overlays(pm: PathManager, sid: str, sim: str) -> list[dict] | None:
    """Electrode-overlay NIfTI presence for one simulation, per TI/mTI mode.

    ``None`` if the subject/simulation is unknown. This is the listing half
    of the electrode-overlay v1 gap documented in
    ``tit/server/routes/viewers.py`` and ``pages/viewer/PARITY.md`` #1:
    creating the overlay is a ``tools`` job
    (``tit.tools.electrode_overlay``), and once it exists on disk,
    :func:`tit.viewspec._electrode_overlay_layer` already picks it up for
    ``kind=simulation`` ViewSpecs -- this endpoint just tells the Viewer page
    whether that file is there yet (and its path), without a separate job
    kind or catalog change for the overlay-*building* side.
    """
    if sid not in subject_ids(pm) or sim not in _simulation_names(pm, sid):
        return None
    sim_dir = pm.simulation(sid, sim)
    out = []
    for mode in _MODE_DIRS:
        path = os.path.join(
            sim_dir, mode, "montage_imgs", "electrode_overlay_subject.nii.gz"
        )
        out.append(
            {
                "mode": mode,
                "path": path,
                "exists": (_project_paths_safe(pm, path) and os.path.isfile(path)),
            }
        )
    return out

get_montages

get_montages(pm: PathManager) -> dict

montage_list.json, reshaped to the v1 Montages schema.

Reads the explicitly selected project's montage file through the shared writer API.

Source code in tit/catalog.py

def get_montages(pm: PathManager) -> dict:
    """``montage_list.json``, reshaped to the v1 ``Montages`` schema.

    Reads the explicitly selected project's montage file through the shared writer API.
    """
    from tit.sim.utils import load_montage_data

    if not _project_paths_safe(pm, pm.montage_config()):
        return {"nets": {}}
    data = load_montage_data(pm=pm)
    nets = {
        net: {
            "uni_polar": entry.get("uni_polar_montages", {}),
            "multi_polar": entry.get("multi_polar_montages", {}),
        }
        for net, entry in data.get("nets", {}).items()
    }
    return {"nets": nets}

put_montage

put_montage(pm: PathManager, net: str, kind: str, name: str, pairs: list[list[str]]) -> list[list[str]]

Create or overwrite one montage; returns pairs back.

Source code in tit/catalog.py

def put_montage(
    pm: PathManager, net: str, kind: str, name: str, pairs: list[list[str]]
) -> list[list[str]]:
    """Create or overwrite one montage; returns *pairs* back."""
    if not is_safe_name(name):
        raise ValueError(
            "name must match ^[A-Za-z0-9_-]{1,64}$ (no path separators or '..')"
        )
    from tit.sim.utils import upsert_montage

    if not _project_paths_safe(pm, pm.montage_config()):
        raise ValueError("Montage path must remain inside the project")
    mode = "U" if kind == "uni_polar" else "M"
    upsert_montage(
        eeg_net=net,
        montage_name=name,
        electrode_pairs=[list(pair) for pair in pairs],
        mode=mode,
        pm=pm,
    )
    return pairs

delete_montage

delete_montage(pm: PathManager, net: str, kind: str, name: str) -> bool

Delete one montage; False if it did not exist.

Source code in tit/catalog.py

def delete_montage(pm: PathManager, net: str, kind: str, name: str) -> bool:
    """Delete one montage; ``False`` if it did not exist."""
    from tit.sim.utils import load_montage_data, save_montage_data

    if not _project_paths_safe(pm, pm.montage_config()):
        return False
    data = load_montage_data(pm=pm)
    key = "uni_polar_montages" if kind == "uni_polar" else "multi_polar_montages"
    montages = data.get("nets", {}).get(net, {}).get(key, {})
    if name not in montages:
        return False
    del montages[name]
    save_montage_data(data, pm=pm)
    return True

eeg_nets

eeg_nets(pm: PathManager, sid: str) -> list[dict] | None

EEG nets available to sid, with electrode labels; None if unknown.

Source code in tit/catalog.py

def eeg_nets(pm: PathManager, sid: str) -> list[dict] | None:
    """EEG nets available to *sid*, with electrode labels; ``None`` if unknown."""
    if sid not in subject_ids(pm):
        return None
    return [
        {"name": name, "electrodes": electrodes, "n": len(electrodes)}
        for name, electrodes in _eeg_caps_with_electrodes(pm, sid)
    ]

atlases

atlases(pm: PathManager, sid: str, space: str | None = None, kind: str | None = None) -> list[dict] | None

Atlases available to sid; None if the subject is unknown.

Every entry carries kind: "surface" for a FreeSurfer .annot parcellation, "volume" for a label volume. For MNI space the list, its order and each entry's kind, template and licence come from resources/atlas/manifest.json (:mod:tit.atlas.manifest) rather than from a filename list, so an atlas is only ever offered to the targeting mode that can read it: the cortical mode asks for kind="cortical" and gets surfaces, the subcortical mode asks for kind="subcortical" and gets volumes. Before the manifest existed nothing recorded the kind of an MNI atlas and the subcortical mode was handed every shipped MNI file whatever it was.

Source code in tit/catalog.py

def atlases(
    pm: PathManager, sid: str, space: str | None = None, kind: str | None = None
) -> list[dict] | None:
    """Atlases available to *sid*; ``None`` if the subject is unknown.

    Every entry carries ``kind``: ``"surface"`` for a FreeSurfer ``.annot``
    parcellation, ``"volume"`` for a label volume. For MNI space the list, its
    order and each entry's kind, template and licence come from
    ``resources/atlas/manifest.json`` (:mod:`tit.atlas.manifest`) rather than
    from a filename list, so an atlas is only ever offered to the targeting mode
    that can read it: the cortical mode asks for ``kind="cortical"`` and gets
    surfaces, the subcortical mode asks for ``kind="subcortical"`` and gets
    volumes. Before the manifest existed nothing recorded the kind of an MNI
    atlas and the subcortical mode was handed every shipped MNI file whatever it
    was.
    """
    from tit.atlas.manifest import KIND_FOR_MODE, kind_for_path, mni_atlas_entries

    if sid not in subject_ids(pm):
        return None
    space = (space or "subject").lower()
    wanted = KIND_FOR_MODE.get(kind) if kind else None
    out: list[dict] = []

    if kind in (None, "cortical") and space == "subject":
        seg_dir = os.path.join(pm.m2m(sid), "segmentation")
        mesh_mgr = MeshAtlasManager(seg_dir if _project_paths_safe(pm, seg_dir) else "")
        for name in mesh_mgr.list_atlases():
            lh_path = mesh_mgr.find_atlas_file(name, "lh")
            if lh_path and not _project_paths_safe(pm, lh_path):
                continue
            out.append(
                {
                    "id": name,
                    "name": name,
                    "path": lh_path or "",
                    "kind": "surface",
                    "hemispheres": ["lh", "rh"],
                }
            )

    if space == "mni":
        # The manifest decides, not the mode: a volume atlas is offered to the
        # subcortical/volumetric flow and a surface one to the cortical flow,
        # and neither mode is ever handed an atlas it cannot read.
        for entry in mni_atlas_entries(mni_resources_dir()):
            if wanted is not None and entry["kind"] != wanted:
                continue
            out.append(
                {
                    "id": entry["id"],
                    "name": entry.get("name") or entry["id"],
                    "path": entry["path"],
                    "kind": entry["kind"],
                    "template": entry.get("template", ""),
                    "license": entry.get("license", ""),
                    "citation": entry.get("citation", ""),
                }
            )
    elif kind in (None, "subcortical"):
        seg_dir = os.path.join(pm.m2m(sid), "segmentation")
        voxel_mgr = VoxelAtlasManager(
            fastsurfer_mri_dir=(
                pm.fastsurfer_mri(sid)
                if _project_paths_safe(pm, pm.fastsurfer_mri(sid))
                else ""
            ),
            freesurfer_mri_dir=(
                pm.freesurfer_mri(sid)
                if _project_paths_safe(pm, pm.freesurfer_mri(sid))
                else ""
            ),
            seg_dir=seg_dir if _project_paths_safe(pm, seg_dir) else "",
            masks_dir=(pm.masks(sid) if _project_paths_safe(pm, pm.masks(sid)) else ""),
        )
        for display_name, path in voxel_mgr.list_atlases():
            if not _project_paths_safe(pm, path):
                continue
            entry: dict = {
                "id": display_name,
                "name": display_name,
                "path": path,
                "kind": kind_for_path(path),
            }
            hemi = VOXEL_ATLASES.get(display_name)
            if hemi in ("lh", "rh"):
                entry["hemispheres"] = [hemi]
            out.append(entry)

    return out

atlas_regions

atlas_regions(pm: PathManager, sid: str, atlas_id: str, hemi: str | None = None) -> list[dict] | None

Regions of atlas_id for sid; None if the subject/atlas is unknown.

Per the reconciled v1 Region schema: for a cortical (surface/annotation) atlas, id is the FreeSurfer .annot label index within that region's own hemisphere file -- exactly the integer FlexConfig.AtlasROI.label / ExConfig's equivalent needs, resolved the same way :func:tit.opt.roi_spec.resolve_cortical_region_index_map does -- and hemi is that region's hemisphere. For a subcortical (volumetric) atlas, id is the voxel label value and hemi is None (a region whose label cannot be parsed as an integer is dropped rather than returned with a non-conforming id).

Source code in tit/catalog.py

def atlas_regions(
    pm: PathManager, sid: str, atlas_id: str, hemi: str | None = None
) -> list[dict] | None:
    """Regions of *atlas_id* for *sid*; ``None`` if the subject/atlas is unknown.

    Per the reconciled v1 ``Region`` schema: for a cortical (surface/annotation)
    atlas, ``id`` is the FreeSurfer ``.annot`` label index *within that
    region's own hemisphere file* -- exactly the integer
    ``FlexConfig.AtlasROI.label`` / ``ExConfig``'s equivalent needs, resolved
    the same way :func:`tit.opt.roi_spec.resolve_cortical_region_index_map`
    does -- and ``hemi`` is that region's hemisphere. For a subcortical
    (volumetric) atlas, ``id`` is the voxel label value and ``hemi`` is
    ``None`` (a region whose label cannot be parsed as an integer is dropped
    rather than returned with a non-conforming id).
    """
    if hemi not in (None, "both", "lh", "rh") or sid not in subject_ids(pm):
        return None
    seg_dir = os.path.join(pm.m2m(sid), "segmentation")

    mesh_mgr = MeshAtlasManager(seg_dir if _project_paths_safe(pm, seg_dir) else "")
    if atlas_id in mesh_mgr.list_atlases():
        hemis = ("lh", "rh") if hemi in (None, "both") else (hemi,)
        out: list[dict] = []
        for h in hemis:
            annot_path = mesh_mgr.find_atlas_file(atlas_id, h)
            if not annot_path:
                continue
            if not is_within(pm.project_dir, annot_path):
                return None
            for index, name in mesh_mgr.list_annot_regions(annot_path):
                if name == "unknown":
                    continue
                out.append({"id": index, "name": name, "hemi": h})
        return out

    voxel_mgr = VoxelAtlasManager(
        fastsurfer_mri_dir=(
            pm.fastsurfer_mri(sid)
            if _project_paths_safe(pm, pm.fastsurfer_mri(sid))
            else ""
        ),
        freesurfer_mri_dir=(
            pm.freesurfer_mri(sid)
            if _project_paths_safe(pm, pm.freesurfer_mri(sid))
            else ""
        ),
        seg_dir=seg_dir if _project_paths_safe(pm, seg_dir) else "",
        masks_dir=pm.masks(sid) if _project_paths_safe(pm, pm.masks(sid)) else "",
    )
    atlas_path = dict(voxel_mgr.list_atlases()).get(atlas_id)
    atlas_root = pm.project_dir
    if atlas_path is None:
        atlas_root = mni_resources_dir()
        atlas_path = {
            os.path.basename(path): path
            for path in VoxelAtlasManager.detect_mni_atlases(atlas_root)
        }.get(atlas_id)
    if atlas_path is None:
        return None
    atlas_path = _jailed_atlas_path(atlas_root, atlas_path)
    if atlas_path is None:
        return None

    entries = voxel_mgr.list_regions(atlas_path)
    out = []
    for entry in entries:
        try:
            region_id = parse_region_label(entry)
        except ValueError:
            continue  # non-integer label: cannot satisfy Region.id: integer
        name = re.sub(r"\s*\(ID:\s*-?\d+\)\s*$", "", entry)
        out.append({"id": region_id, "name": name, "hemi": None})
    return out

nifti_labels

nifti_labels(pm: PathManager, sid: str, path: str | None = None) -> list[dict] | None

Unique integer labels present in one label volume, named where a LUT applies.

The 3D Visual Exporter's sub-cortical mode asks the user for label numbers (10,49); 2.5.0 answered that with a Qt dialog that parsed a FreeSurfer LUT itself. This is that browser's data, from the toolbox's own segstats path (:func:tit.atlas.segstats.compute_segstats + :func:~tit.atlas.segstats.resolve_lut_for_atlas), which also writes the <name>_labels.txt sidecar cache beside the volume -- so the first call on a subject pays for the scan and every later one reads the cache, same as :meth:VoxelAtlasManager.list_regions.

Parameters

pm : PathManager sid : str Subject whose m2m holds the default volume. path : str or None A specific label volume. None/empty means the subject's own <m2m>/segmentation/labeling.nii.gz -- the sub-cortical mode's default, and the only path the panel ever sends for an untouched form.

Returns

list of dict or None [{"id": int, "name": str, "n_voxels": int}] sorted by id; None for an unknown subject, a path outside the project jail, or a file that is not there. The caller (routes/catalog_v1) turns None into a 404 -- deliberately one status for all three, so probing this route cannot tell a file that exists outside the jail from one that does not exist at all.

Source code in tit/catalog.py

def nifti_labels(
    pm: PathManager, sid: str, path: str | None = None
) -> list[dict] | None:
    """Unique integer labels present in one label volume, named where a LUT applies.

    The 3D Visual Exporter's sub-cortical mode asks the user for label *numbers*
    (``10,49``); 2.5.0 answered that with a Qt dialog that parsed a FreeSurfer LUT
    itself. This is that browser's data, from the toolbox's own segstats path
    (:func:`tit.atlas.segstats.compute_segstats` +
    :func:`~tit.atlas.segstats.resolve_lut_for_atlas`), which also writes the
    ``<name>_labels.txt`` sidecar cache beside the volume -- so the first call on a
    subject pays for the scan and every later one reads the cache, same as
    :meth:`VoxelAtlasManager.list_regions`.

    Parameters
    ----------
    pm : PathManager
    sid : str
        Subject whose ``m2m`` holds the default volume.
    path : str or None
        A specific label volume. ``None``/empty means the subject's own
        ``<m2m>/segmentation/labeling.nii.gz`` -- the sub-cortical mode's default,
        and the only path the panel ever sends for an untouched form.

    Returns
    -------
    list of dict or None
        ``[{"id": int, "name": str, "n_voxels": int}]`` sorted by ``id``; ``None``
        for an unknown subject, a path outside the project jail, or a file that is
        not there. The caller (``routes/catalog_v1``) turns ``None`` into a 404 --
        deliberately one status for all three, so probing this route cannot tell a
        file that exists outside the jail from one that does not exist at all.
    """
    if sid not in subject_ids(pm):
        return None

    raw = (path or "").strip()
    if not raw:
        m2m = pm.m2m(sid)
        if not m2m:
            return None
        raw = os.path.join(m2m, "segmentation", "labeling.nii.gz")

    from tit.viewspec import jail_roots, resolve_jailed

    resolved = resolve_jailed(raw)
    if resolved is None:
        return None
    if not any(
        _jailed_atlas_path(str(root), str(resolved)) is not None
        for root in jail_roots()
    ):
        return None

    cached = _read_segstats_sum(_segstats_sidecar(str(resolved)))
    if cached is not None:
        return cached

    from tit.atlas.segstats import (
        compute_segstats,
        resolve_lut_for_atlas,
        write_segstats_sum,
    )

    try:
        lut = resolve_lut_for_atlas(str(resolved))
        stats = compute_segstats(str(resolved), lut)
    except (OSError, ValueError):
        # An unreadable or non-label volume is "nothing to browse", not a 500: the
        # path came from a text field the user can type anything into.
        return None
    try:
        write_segstats_sum(stats, _segstats_sidecar(str(resolved)))
    except OSError:
        # A read-only project still gets its answer -- it just pays for the scan again.
        pass
    return [
        {"id": int(s.seg_id), "name": s.name, "n_voxels": int(s.n_voxels)}
        for s in stats
    ]

is_safe_name

is_safe_name(name: Any) -> bool

True if name is safe to use as a filename component.

Used for ROI names, free-hand config names, and montage names on every write path -- never trust a client-supplied string used to build a path.

Source code in tit/catalog.py

def is_safe_name(name: Any) -> bool:
    """``True`` if *name* is safe to use as a filename component.

    Used for ROI names, free-hand config names, and montage names on every
    write path -- never trust a client-supplied string used to build a path.
    """
    return isinstance(name, str) and bool(_SAFE_NAME_RE.match(name))

as_float

as_float(value: Any, field_name: str) -> float

Coerce value to float; raises ValueError naming the field on failure.

bool is rejected even though isinstance(True, int) is True in Python -- a checkbox value has no business landing in a coordinate.

Source code in tit/catalog.py

def as_float(value: Any, field_name: str) -> float:
    """Coerce *value* to ``float``; raises ``ValueError`` naming the field on failure.

    ``bool`` is rejected even though ``isinstance(True, int)`` is ``True`` in
    Python -- a checkbox value has no business landing in a coordinate.
    """
    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
        raise ValueError(f"{field_name} must be a number")
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{field_name} must be a number") from exc

list_rois

list_rois(pm: PathManager, sid: str) -> list[dict] | None

Saved ROIs for sid; None if the subject is unknown.

Source code in tit/catalog.py

def list_rois(pm: PathManager, sid: str) -> list[dict] | None:
    """Saved ROIs for *sid*; ``None`` if the subject is unknown."""
    if sid not in subject_ids(pm):
        return None
    from tit.opt.ex.roi import read_roi_center

    roi_dir = pm.rois(sid)
    if not _project_paths_safe(pm, roi_dir) or not os.path.isdir(roi_dir):
        return []
    out = []
    for name in sorted(os.listdir(roi_dir)):
        if not name.endswith(".csv") or not _project_paths_safe(
            pm, os.path.join(roi_dir, name)
        ):
            continue
        try:
            x, y, z = read_roi_center(os.path.join(roi_dir, name))[:3]
        except (OSError, ValueError):
            continue
        roi_name = name[: -len(".csv")]
        space = "mni" if roi_name.upper().endswith("_MNI") else "subject"
        out.append({"name": roi_name, "x": x, "y": y, "z": z, "space": space})
    return out

create_roi

create_roi(pm: PathManager, sid: str, roi: dict) -> dict

Save a new spherical ROI centre for sid (subject-space CSV convention).

Mirrors :meth:tit.opt.ex.engine.ExSearchEngine.create_roi (a plain x,y,z row plus a roi_list.txt entry) rather than importing it, so this module stays the single writer of the CRUD shape the v1 contract expects (radius/space are accepted but not persisted -- the CSV format has never carried them; see this lane's final report).

Source code in tit/catalog.py

def create_roi(pm: PathManager, sid: str, roi: dict) -> dict:
    """Save a new spherical ROI centre for *sid* (subject-space CSV convention).

    Mirrors :meth:`tit.opt.ex.engine.ExSearchEngine.create_roi` (a plain
    ``x,y,z`` row plus a ``roi_list.txt`` entry) rather than importing it, so
    this module stays the single writer of the CRUD shape the v1 contract
    expects (radius/space are accepted but not persisted -- the CSV format
    has never carried them; see this lane's final report).
    """
    name = str(roi["name"])
    if not is_safe_name(name):
        raise ValueError(
            "name must match ^[A-Za-z0-9_-]{1,64}$ (no path separators or '..')"
        )
    x = as_float(roi["x"], "x")
    y = as_float(roi["y"], "y")
    z = as_float(roi["z"], "z")
    radius = roi.get("radius")
    if radius is not None:
        radius = as_float(radius, "radius")

    roi_dir = pm.rois(sid)
    filename = f"{name}.csv"
    base_name = name
    roi_path = _jailed(pm, os.path.join(roi_dir, filename))
    roi_list = _jailed(pm, os.path.join(roi_dir, "roi_list.txt"))
    if roi_path is None or roi_list is None:
        raise ValueError("ROI paths must remain inside the project")
    os.makedirs(roi_dir, exist_ok=True)

    with open(roi_path, "w", newline="") as f:
        csv.writer(f).writerow([x, y, z])

    existing = []
    if os.path.isfile(roi_list):
        existing = [line.strip() for line in open(roi_list) if line.strip()]
    if filename not in existing:
        with open(roi_list, "a") as f:
            f.write(f"{filename}\n")

    space = "mni" if base_name.upper().endswith("_MNI") else "subject"
    return {
        "name": base_name,
        "x": x,
        "y": y,
        "z": z,
        "space": roi.get("space", space),
        "radius": radius,
    }

delete_roi

delete_roi(pm: PathManager, sid: str, name: str) -> bool

Delete one saved ROI; False if it did not exist (also if name is unsafe).

Source code in tit/catalog.py

def delete_roi(pm: PathManager, sid: str, name: str) -> bool:
    """Delete one saved ROI; ``False`` if it did not exist (also if *name* is unsafe)."""
    if not is_safe_name(name):
        return False
    roi_dir = pm.rois(sid)
    filename = name if name.endswith(".csv") else f"{name}.csv"
    roi_path = _jailed(pm, os.path.join(roi_dir, filename))
    roi_list = _jailed(pm, os.path.join(roi_dir, "roi_list.txt"))
    if roi_path is None or roi_list is None:
        return False
    existed = os.path.isfile(roi_path)
    if existed:
        os.remove(roi_path)

    if os.path.isfile(roi_list):
        lines = [line.strip() for line in open(roi_list) if line.strip()]
        if filename in lines:
            lines.remove(filename)
            with open(roi_list, "w") as f:
                f.write("\n".join(lines) + ("\n" if lines else ""))
    return existed

list_leadfields

list_leadfields(pm: PathManager, sid: str) -> list[dict] | None

Precomputed leadfields for sid; None if the subject is unknown.

Source code in tit/catalog.py

def list_leadfields(pm: PathManager, sid: str) -> list[dict] | None:
    """Precomputed leadfields for *sid*; ``None`` if the subject is unknown."""
    if sid not in subject_ids(pm):
        return None
    root = pm.leadfields(sid)
    out = []
    for name in _metadata_names(root, pm.project_dir):
        if not name.endswith(".hdf5"):
            continue
        path = os.path.join(root, name)
        # Match LeadfieldGenerator's established net naming without its unjailed stat.
        stem = name[:-5]
        net = (
            stem.split("_leadfield_", 1)[-1]
            if "_leadfield_" in stem
            else stem.removesuffix("_leadfield")
        )
        for prefix in (f"{sid}_", sid):
            if net.startswith(prefix):
                net = net[len(prefix) :]
                break
        net = net.strip("_") or "unknown"
        try:
            size = os.path.getsize(path)
        except OSError:
            size = 0
        out.append(
            {
                "net": net,
                "path": path,
                "exists": os.path.isfile(path),
                "size_bytes": size,
            }
        )
    return sorted(out, key=lambda item: (item["net"], item["path"]))

flex_runs

flex_runs(pm: PathManager, sid: str) -> list[dict] | None

Flex-search runs for sid; None if the subject is unknown.

Source code in tit/catalog.py

def flex_runs(pm: PathManager, sid: str) -> list[dict] | None:
    """Flex-search runs for *sid*; ``None`` if the subject is unknown."""
    if sid not in subject_ids(pm):
        return None
    from tit.opt.flex.manifest import read_manifest

    out = []
    if not _project_paths_safe(pm, pm.flex_search(sid)):
        return out
    for name in pm.list_flex_search_runs(sid):
        run_dir = pm.flex_search_run(sid, name)
        if not _project_paths_safe(
            pm, run_dir, os.path.join(run_dir, "flex_meta.json")
        ):
            continue
        manifest = read_manifest(run_dir)
        if manifest is None:  # no flex_meta.json: ignore, per the plan
            continue
        out.append(
            {
                "name": name,
                "path": run_dir,
                "goal": manifest.get("goal", ""),
                "roi": manifest.get("roi") or {},
                "created": manifest.get("created") or _mtime_iso(run_dir),
                "manifest": manifest,
                # The electrodes the run actually produced. `flex_meta.json` records
                # none of them, so a client that reads only `manifest` has no way to
                # turn a run into a `Montage` for submission (simulator PARITY.md #4).
                "mappings": _flex_mappings(run_dir, project_root=pm.project_dir),
                "optimized": _flex_optimized_pairs(
                    run_dir, project_root=pm.project_dir
                ),
                "artifacts": _dir_artifacts(run_dir, project_root=pm.project_dir),
            }
        )
    return out

ex_runs

ex_runs(pm: PathManager, sid: str, kind: str = 'ex') -> list[dict] | None

Ex/mEx-search runs for sid; None if the subject is unknown.

Source code in tit/catalog.py

def ex_runs(pm: PathManager, sid: str, kind: str = "ex") -> list[dict] | None:
    """Ex/mEx-search runs for *sid*; ``None`` if the subject is unknown."""
    if sid not in subject_ids(pm):
        return None
    root = pm.ex_search(sid) if kind == "ex" else pm.m_ex_search(sid)
    if not _project_paths_safe(pm, root) or not os.path.isdir(root):
        return []
    out = []
    for entry in sorted(os.scandir(root), key=lambda e: e.name):
        if not _project_paths_safe(
            pm, entry.path, os.path.join(entry.path, "run_config.json")
        ):
            continue
        if not entry.is_dir() or entry.name.startswith("."):
            continue
        run_config = _read_json(os.path.join(entry.path, "run_config.json"))
        if run_config is None:  # no run_config.json: ignore, per the plan
            continue
        out.append(
            {
                "run_name": entry.name,
                "path": entry.path,
                "eeg_net": _net_from_leadfield_hdf(run_config.get("leadfield_hdf", "")),
                "created": _mtime_iso(entry.path),
                "best": _ex_best(entry.path, project_root=pm.project_dir),
                "artifacts": _dir_artifacts(entry.path, project_root=pm.project_dir),
            }
        )
    return out

ex_run_results

ex_run_results(pm: PathManager, sid: str, kind: str, run: str) -> dict | None

Full final_output.csv of one ex/mEx run, as TableData.

Source code in tit/catalog.py

def ex_run_results(pm: PathManager, sid: str, kind: str, run: str) -> dict | None:
    """Full ``final_output.csv`` of one ex/mEx run, as ``TableData``."""
    if sid not in subject_ids(pm):
        return None
    root = pm.ex_search_run(sid, run) if kind == "ex" else pm.m_ex_search_run(sid, run)
    csv_path = os.path.join(root, "final_output.csv")
    if not _project_paths_safe(pm, csv_path):
        return None
    if not os.path.isfile(csv_path):
        return None
    return _read_csv_table(csv_path)

analyses

analyses(pm: PathManager, sid: str, sim: str) -> list[dict] | None

Analyzer runs for sid/sim; None if either is unknown.

Source code in tit/catalog.py

def analyses(pm: PathManager, sid: str, sim: str) -> list[dict] | None:
    """Analyzer runs for *sid*/*sim*; ``None`` if either is unknown."""
    if sid not in subject_ids(pm) or sim not in _simulation_names(pm, sid):
        return None
    out = []
    for path, name, _standard in _find_analysis_dirs(
        pm.simulation(sid, sim), project_root=pm.project_dir
    ):
        item = _analysis_entry(path, name, project_root=pm.project_dir)
        if item:
            out.append(item)
    return out

analysis_summary

analysis_summary(pm: PathManager, sid: str, sim: str, name: str) -> dict | None

results.csv of one analysis, as TableData.

name is what :func:analyses listed; a caller holding the run's own path relative to the simulation (Analyses/Custom/run1) resolves too, so the one string a script already has does not have to be translated into the listed form.

Source code in tit/catalog.py

def analysis_summary(pm: PathManager, sid: str, sim: str, name: str) -> dict | None:
    """``results.csv`` of one analysis, as ``TableData``.

    *name* is what :func:`analyses` listed; a caller holding the run's own path relative
    to the simulation (``Analyses/Custom/run1``) resolves too, so the one string a script
    already has does not have to be translated into the listed form.
    """
    if sid not in subject_ids(pm) or sim not in _simulation_names(pm, sid):
        return None
    sim_dir = pm.simulation(sid, sim)
    for path, found_name, _standard in _find_analysis_dirs(
        sim_dir, project_root=pm.project_dir
    ):
        relative = os.path.relpath(path, sim_dir).replace(os.sep, "/")
        if name in (found_name, relative):
            return _read_csv_table(os.path.join(path, "results.csv"))
    return None

reports

reports(pm: PathManager, sid: str) -> list[dict] | None

Generated HTML reports for sid; None if the subject is unknown.

Source code in tit/catalog.py

def reports(pm: PathManager, sid: str) -> list[dict] | None:
    """Generated HTML reports for *sid*; ``None`` if the subject is unknown."""
    if sid not in subject_ids(pm):
        return None
    root = _jailed(pm, os.path.join(pm.reports(), f"sub-{sid}"))
    if root is None or not os.path.isdir(root):
        return []
    entries = []
    for name in sorted(os.listdir(root)):
        path = _jailed(pm, os.path.join(root, name)) if name.endswith(".html") else None
        if path:
            entries.append(_report_entry(path, sid))
    return entries

find_report_path

find_report_path(pm: PathManager, report_id: str) -> str | None

Resolve a reports() id back to its file path (used by /api/files/report).

Source code in tit/catalog.py

def find_report_path(pm: PathManager, report_id: str) -> str | None:
    """Resolve a ``reports()`` ``id`` back to its file path (used by ``/api/files/report``)."""
    if "/" not in report_id:
        return None
    sid, stem = report_id.split("/", 1)
    if any(c in stem for c in ("/", "\\", "..")):
        return None
    path = _jailed(pm, os.path.join(pm.reports(), f"sub-{sid}", f"{stem}.html"))
    return path if path and os.path.isfile(path) else None

freehand_configs

freehand_configs(pm: PathManager, sid: str) -> list[dict] | None

Saved free-hand electrode configs for sid; None if unknown.

Reads m2m_<id>/stim_configs/*.json. The on-disk format, written by the free-hand electrode placement UI, is {"name", "type": "U"|"M", "electrode_positions": {label: [x, y, z]}}. The contract's FreehandConfig.type enum is [U, M] (unipolar/ multipolar), matching this on-disk value exactly (fixed from an earlier xyz/label enum that matched nothing real -- see contracts/CHANGES.md); type is passed through verbatim.

Source code in tit/catalog.py

def freehand_configs(pm: PathManager, sid: str) -> list[dict] | None:
    """Saved free-hand electrode configs for *sid*; ``None`` if unknown.

    Reads ``m2m_<id>/stim_configs/*.json``. The on-disk format, written by the
    free-hand electrode placement UI, is
    ``{"name", "type": "U"|"M", "electrode_positions": {label: [x, y, z]}}``.
    The contract's ``FreehandConfig.type`` enum is ``[U, M]`` (unipolar/
    multipolar), matching this on-disk value exactly (fixed from an earlier
    ``xyz``/``label`` enum that matched nothing real -- see
    ``contracts/CHANGES.md``); ``type`` is passed through verbatim.
    """
    if sid not in subject_ids(pm):
        return None
    stim_dir = os.path.join(pm.m2m(sid), "stim_configs")
    if not _project_paths_safe(pm, stim_dir) or not os.path.isdir(stim_dir):
        return []
    out = []
    for name in sorted(os.listdir(stim_dir)):
        if not name.endswith(".json") or not _project_paths_safe(
            pm, os.path.join(stim_dir, name)
        ):
            continue
        cfg = _read_freehand_file(os.path.join(stim_dir, name))
        if cfg:
            out.append(cfg)
    return out

put_freehand_config

put_freehand_config(pm: PathManager, sid: str, name: str, config: dict) -> dict

Create or overwrite one free-hand electrode configuration.

Source code in tit/catalog.py

def put_freehand_config(pm: PathManager, sid: str, name: str, config: dict) -> dict:
    """Create or overwrite one free-hand electrode configuration."""
    if not is_safe_name(name):
        raise ValueError(
            "name must match ^[A-Za-z0-9_-]{1,64}$ (no path separators or '..')"
        )
    stim_dir = os.path.join(pm.m2m(sid), "stim_configs")
    path = _jailed(pm, os.path.join(stim_dir, f"{name}.json"))
    if path is None:
        raise ValueError("Free-hand config path must remain inside the project")
    os.makedirs(stim_dir, exist_ok=True)
    positions = {
        (entry.get("label") or f"E{i + 1}"): [
            as_float(entry["x"], "electrode_positions[].x"),
            as_float(entry["y"], "electrode_positions[].y"),
            as_float(entry["z"], "electrode_positions[].z"),
        ]
        for i, entry in enumerate(config.get("electrode_positions", []))
    }
    data = {
        "name": name,
        "type": config.get("type", "U"),
        "electrode_positions": positions,
    }
    with open(path, "w") as f:
        json.dump(data, f, indent=2)
    return {
        "name": name,
        "type": data["type"],
        "electrode_positions": [
            {"label": label, "x": xyz[0], "y": xyz[1], "z": xyz[2]}
            for label, xyz in positions.items()
        ],
    }

delete_freehand_config

delete_freehand_config(pm: PathManager, sid: str, name: str) -> bool

Remove a saved placement definition, leaving simulation outputs intact.

Source code in tit/catalog.py

def delete_freehand_config(pm: PathManager, sid: str, name: str) -> bool:
    """Remove a saved placement definition, leaving simulation outputs intact."""
    if not is_safe_name(name):
        raise ValueError("name must match ^[A-Za-z0-9_-]{1,64}$")
    if sid not in subject_ids(pm):
        return False
    path = _jailed(pm, os.path.join(pm.m2m(sid), "stim_configs", f"{name}.json"))
    if path is None or not os.path.isfile(path):
        return False
    try:
        os.unlink(path)
    except FileNotFoundError:
        return False
    return True

group_catalog

group_catalog(pm: PathManager) -> dict

Project-level group catalog: stats runs, nilearn visuals, group analyses.

Directory conventions for the nilearn and group_analyses sections are a best-effort guess (derivatives/ti-toolbox/{nilearn_visuals, group_analysis}/) -- unverified against a real project, since Dataset 000 has neither populated. stats (derivatives/ti-toolbox/stats/ <type>/<name>/) is confirmed by :meth:PathManager.stats_output.

Source code in tit/catalog.py

def group_catalog(pm: PathManager) -> dict:
    """Project-level group catalog: stats runs, nilearn visuals, group analyses.

    Directory conventions for the ``nilearn`` and ``group_analyses`` sections
    are a best-effort guess (``derivatives/ti-toolbox/{nilearn_visuals,
    group_analysis}/``) -- unverified against a real project, since Dataset
    000 has neither populated. ``stats`` (``derivatives/ti-toolbox/stats/
    <type>/<name>/``) is confirmed by :meth:`PathManager.stats_output`.
    """

    def _subdirs(root: str) -> list[str]:
        if not _project_paths_safe(pm, root):
            return []
        try:
            return sorted(
                n
                for n in os.listdir(root)
                if _project_paths_safe(pm, os.path.join(root, n))
                and os.path.isdir(os.path.join(root, n))
            )
        except OSError:
            return []

    stats = []
    stats_root = os.path.join(pm.ti_toolbox(), "stats")
    for analysis_type in _subdirs(stats_root):
        if analysis_type == "data":
            continue
        type_dir = os.path.join(stats_root, analysis_type)
        for name in _subdirs(type_dir):
            run_dir = os.path.join(type_dir, name)
            stats.append(
                {
                    "type": analysis_type,
                    "name": name,
                    "path": run_dir,
                    "created": _mtime_iso(run_dir),
                }
            )

    nilearn = []
    nilearn_root = os.path.join(pm.ti_toolbox(), "nilearn_visuals")
    for name in _subdirs(nilearn_root):
        path = os.path.join(nilearn_root, name)
        nilearn.append({"name": name, "path": path, "created": _mtime_iso(path)})

    group_analyses = []
    group_root = os.path.join(pm.ti_toolbox(), "group_analysis")
    for name in _subdirs(group_root):
        path = os.path.join(group_root, name)
        group_analyses.append({"name": name, "path": path, "created": _mtime_iso(path)})

    return {"stats": stats, "nilearn": nilearn, "group_analyses": group_analyses}

group_stats_detail

group_stats_detail(pm: PathManager, analysis_type: str, name: str) -> dict | None

One derivatives/ti-toolbox/stats/<type>/<name>/ run, read for the Results pane.

None when the run directory does not exist. status is "ok" once the run has written something besides its log, and "empty" when it has not -- the state the maintainer hit, where a failed 2-vs-1 group comparison left a bare .log and the pane could say nothing about it. reason is then the log's own ERROR line, so the UI reports why instead of showing an empty file list.

Source code in tit/catalog.py

def group_stats_detail(pm: PathManager, analysis_type: str, name: str) -> dict | None:
    """One ``derivatives/ti-toolbox/stats/<type>/<name>/`` run, read for the Results pane.

    ``None`` when the run directory does not exist. ``status`` is ``"ok"`` once the run has
    written something besides its log, and ``"empty"`` when it has not -- the state the
    maintainer hit, where a failed 2-vs-1 group comparison left a bare ``.log`` and the pane
    could say nothing about it. ``reason`` is then the log's own ERROR line, so the UI
    reports why instead of showing an empty file list.
    """
    if "/" in analysis_type or "/" in name or ".." in (analysis_type, name):
        return None
    run_dir = _jailed(pm, os.path.join(pm.ti_toolbox(), "stats", analysis_type, name))
    if run_dir is None or not os.path.isdir(run_dir):
        return None

    try:
        names = sorted(os.listdir(run_dir))
    except OSError:
        names = []

    artifacts: list[dict] = []
    for filename in names:
        path = _jailed(pm, os.path.join(run_dir, filename))
        if path is None or not os.path.isfile(path):
            continue
        if filename.endswith(".nii.gz") or filename.endswith(".nii"):
            kind = "nifti"
        else:
            kind = _STATS_KIND_BY_EXT.get(os.path.splitext(filename)[1], "file")
        artifacts.append(
            {
                "path": path,
                "kind": kind,
                "label": _STATS_FILE_LABELS.get(
                    filename, os.path.splitext(filename)[0].replace("_", " ")
                ),
            }
        )
    # The run's own files first, in the order the labels table lists them; anything else after.
    order = list(_STATS_FILE_LABELS)
    artifacts.sort(
        key=lambda a: (
            (
                order.index(os.path.basename(a["path"]))
                if os.path.basename(a["path"]) in order
                else len(order)
            ),
            a["path"],
        )
    )

    log_path = _stats_log_path(run_dir)
    log_path = _jailed(pm, log_path) if log_path else None
    parsed = {
        "config": [],
        "results": [],
        "groups": [],
        "image_shape": None,
        "clusters": None,
        "error": None,
    }
    if log_path:
        try:
            with open(log_path, encoding="utf-8", errors="replace") as f:
                parsed = _parse_stats_log(f.read())
        except OSError:
            pass

    clusters = parsed["clusters"]
    csv_path = _jailed(pm, os.path.join(run_dir, "significant_clusters.csv"))
    if clusters is None and csv_path and os.path.isfile(csv_path):
        try:
            clusters = _read_csv_table(csv_path)
        except OSError:
            clusters = None

    produced = [a for a in artifacts if a["kind"] != "log"]
    status = "ok" if produced else "empty"
    reason = parsed["error"]
    if status == "empty" and not reason:
        reason = (
            "This run wrote no output files. Its log is the only record; open it for the "
            "last step it reached."
        )

    return {
        "type": analysis_type,
        "name": name,
        "path": run_dir,
        "created": _mtime_iso(run_dir),
        "status": status,
        "reason": reason,
        "config": parsed["config"],
        "results": parsed["results"],
        "groups": parsed["groups"],
        "image_shape": parsed["image_shape"],
        "clusters": clusters,
        "log": log_path,
        "artifacts": artifacts,
    }

read_notes

read_notes(pm: PathManager) -> dict

Quick Notes content (derivatives/ti-toolbox/notes.txt).

Source code in tit/catalog.py

def read_notes(pm: PathManager) -> dict:
    """Quick Notes content (``derivatives/ti-toolbox/notes.txt``)."""
    path = os.path.join(pm.ti_toolbox(), "notes.txt")
    if not _project_paths_safe(pm, path) or not os.path.isfile(path):
        return {"text": "", "updated_at": None}
    with open(path, encoding="utf-8") as f:
        text = f.read()
    return {"text": text, "updated_at": _mtime_iso(path)}

write_notes

write_notes(pm: PathManager, text: str) -> dict

Replace Quick Notes content, atomically.

Source code in tit/catalog.py

def write_notes(pm: PathManager, text: str) -> dict:
    """Replace Quick Notes content, atomically."""
    path = os.path.join(pm.ti_toolbox(), "notes.txt")
    tmp = f"{path}.tmp"
    if not _project_paths_safe(pm, path, tmp):
        raise ValueError("Notes paths must remain inside the project")
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(text)
    os.replace(tmp, path)
    return {"text": text, "updated_at": _mtime_iso(path)}

subject_info_matrix

subject_info_matrix(pm: PathManager) -> dict

Presence matrix (subject x data-stage) across the whole project.

Source code in tit/catalog.py

def subject_info_matrix(pm: PathManager) -> dict:
    """Presence matrix (subject x data-stage) across the whole project."""
    columns = [
        "subject",
        "raw",
        "fastsurfer",
        "freesurfer",
        "m2m",
        "dwi",
        "ct",
        "simulations",
    ]
    rows = []
    for sid in subject_ids(pm):
        rows.append(
            [
                sid,
                _project_isdir(pm, pm.bids_subject(sid)),
                _project_isdir(pm, pm.fastsurfer_subject(sid)),
                _project_isdir(pm, pm.freesurfer_subject(sid)),
                _project_isdir(pm, pm.m2m(sid)),
                _project_isdir(pm, pm.bids_dwi(sid)),
                _has_ct(pm, sid),
                len(_simulation_names(pm, sid)),
            ]
        )
    return {"columns": columns, "rows": rows}