candidate_catalog

tit.opt.candidate_catalog

Bounded, project-confined reads of optimization candidate records.

list_candidates

list_candidates(pm, subject: str, kind: str, run: str, offset: int = 0, limit: int = 100, sort: str = 'roi_mean', descending: bool = True) -> dict

Return a deterministic page, with unavailable metrics sorted last.

Source code in tit/opt/candidate_catalog.py

def list_candidates(
    pm,
    subject: str,
    kind: str,
    run: str,
    offset: int = 0,
    limit: int = 100,
    sort: str = "roi_mean",
    descending: bool = True,
) -> dict:
    """Return a deterministic page, with unavailable metrics sorted last."""
    root = Path(pm.project_dir)
    directory = run_directory(pm, subject, kind, run)
    rows = _rows(root, directory, subject, kind)
    if sort not in {*_FIELDS, "objective"}:
        raise ValueError("Unknown sort metric.")

    def key(row):
        value = (
            row.get("objective") if sort == "objective" else row["metrics"].get(sort)
        )
        return (
            value is None,
            (-value if descending else value) if value is not None else 0,
            row["id"],
        )

    if offset < 0 or not 1 <= limit <= 500:
        raise ValueError("Invalid candidate page.")
    rows.sort(key=key)
    page = rows[offset : offset + limit]
    geometries = _geometries(root, page)
    return {
        "candidates": [_public(row, geometries.get(row["id"])) for row in page],
        "total": len(rows),
        "legacy": kind == "flex"
        and not any(directory.rglob("candidate_manifest.json"))
        and not any(directory.rglob("candidates.csv")),
    }

candidate_detail

candidate_detail(pm, subject: str, kind: str, run: str, candidate_id: str) -> dict

Resolve a saved candidate into the standard editable simulation config.

Source code in tit/opt/candidate_catalog.py

def candidate_detail(pm, subject: str, kind: str, run: str, candidate_id: str) -> dict:
    """Resolve a saved candidate into the standard editable simulation config."""
    from tit.config_io import serialize_config
    from tit.sim.config import Montage, SimulationConfig

    root = Path(pm.project_dir)
    directory = run_directory(pm, subject, kind, run)
    rows = _rows(root, directory, subject, kind)
    row = next((r for r in rows if r["id"] == candidate_id), None)
    if row is None:
        raise FileNotFoundError("Candidate not found.")
    name = (
        "candidate_"
        + hashlib.sha256(f"{kind}/{run}/{candidate_id}".encode()).hexdigest()[:16]
    )
    provenance = {
        "kind": kind,
        "run": run,
        "candidate_id": candidate_id,
        "metrics": "optimization estimates",
    }
    config = row["_manifest"].get("config", {})
    geometry = None
    if kind == "flex":
        identity = row["_manifest"].get("head_mesh_identity")
        if identity is None:
            row["replay_note"] = (
                "Head mesh identity was not recorded; unchanged mesh physics cannot be verified for this history."
            )
            provenance["head_mesh_identity"] = "unavailable in legacy history"
        else:
            from tit.mesh_identity import verify_subject_mesh

            if not isinstance(identity, dict) or not isinstance(
                identity.get("path"), str
            ):
                raise ValueError("Candidate head mesh identity is malformed.")
            verify_subject_mesh(pm, subject, identity.get("sha256"))
            provenance["head_mesh_sha256"] = identity["sha256"]
            provenance["head_mesh_source_path"] = identity["path"]
        geometry = _geometries(root, [row])[candidate_id]
        electrodes = []
        currents = []
        for channel in geometry["electrodes"]:
            if len(channel) != 2:
                raise ValueError(
                    "Simulation replay currently requires two electrodes per carrier."
                )
            ordered = sorted(channel, key=lambda e: -e["current_A"])
            positive, negative = (float(e["current_A"]) for e in ordered)
            if not (
                math.isfinite(positive)
                and math.isfinite(negative)
                and positive > 0 > negative
                and abs(positive + negative) < 1e-10
            ):
                raise ValueError(
                    "Candidate carrier currents are invalid or unbalanced."
                )
            currents.append(positive * 1000)
            electrodes.extend(ordered)
        if len(currents) != 2:
            raise ValueError("Flex replay requires exactly two carriers.")
        spec = electrodes[0]
        if any(
            any(cell[key] != spec[key] for key in ("shape", "dimensions", "thickness"))
            for cell in electrodes
        ):
            raise ValueError(
                "Mixed electrode geometries cannot be replayed in the standard Simulator."
            )
        poses = [cell["posmat"] for cell in electrodes]
        positions = [[pose[i][3] for i in range(3)] for pose in poses]
        montage = Montage(
            name,
            Montage.Mode.FLEX_FREE,
            [positions[i : i + 2] for i in range(0, len(positions), 2)],
            electrode_poses=poses,
            provenance=provenance,
        )
        simulation = SimulationConfig(
            subject,
            [montage],
            conductivity=config.get("anisotropy_type", "scalar"),
            intensities=currents,
            electrode_shape=spec["shape"],
            electrode_dimensions=spec["dimensions"],
            gel_thickness=spec["thickness"][0],
            rubber_thickness=spec["thickness"][1],
            aniso_maxratio=config.get("aniso_maxratio", 10),
            aniso_maxcond=config.get("aniso_maxcond", 2),
        )
    else:
        if not all(value is not None and value > 0 for value in row["currents_mA"]):
            raise ValueError("Saved montage does not contain usable carrier currents.")
        montage = Montage(
            name,
            Montage.Mode.NET,
            row["pairs"],
            eeg_net=row["eeg_net"],
            provenance=provenance,
        )
        provenance["geometry"] = "Simulator defaults; not recorded in Ex history"
        simulation = SimulationConfig(
            subject, [montage], intensities=row["currents_mA"]
        )
    return {
        "candidate": _public(row, geometry),
        "simulation_config": serialize_config(simulation),
    }

candidate_mapping

candidate_mapping(pm, subject: str, run: str, candidate_id: str, eeg_net: str) -> dict

Map this candidate to distinct cap sites without changing recorded history.

Distances are Euclidean millimetres, not scalp geodesics. Uses the same assignment as ordinary Flex-result mapping, with candidate-specific poses.

Source code in tit/opt/candidate_catalog.py

def candidate_mapping(
    pm, subject: str, run: str, candidate_id: str, eeg_net: str
) -> dict:
    """Map this candidate to distinct cap sites without changing recorded history.

    Distances are Euclidean millimetres, not scalp geodesics. Uses the same
    assignment as ordinary Flex-result mapping, with candidate-specific poses.
    """
    import numpy as np

    from tit.tools.map_electrodes import map_electrodes_to_net, read_csv_positions

    if not eeg_net or Path(eeg_net).name != eeg_net or "\\" in eeg_net:
        raise ValueError("EEG net must be a filename.")
    detail = candidate_detail(pm, subject, "flex", run, candidate_id)
    pairs = detail["simulation_config"]["montages"][0]["electrode_pairs"]
    positions = np.asarray([point for pair in pairs for point in pair], dtype=float)
    root = Path(pm.project_dir)
    net_path = _file(root, Path(pm.eeg_positions(subject)) / eeg_net)
    net_positions, labels = read_csv_positions(str(net_path))
    if (
        net_positions.shape != (len(labels), 3)
        or len(labels) < len(positions)
        or len(set(labels)) != len(labels)
        or not np.isfinite(net_positions).all()
    ):
        raise ValueError("EEG net needs enough distinct, finite electrode sites.")
    indices = [[i // 2, i % 2] for i in range(len(positions))]
    mapped = map_electrodes_to_net(positions, net_positions, labels, indices)
    mapped_labels = mapped["mapped_labels"]
    if len(mapped_labels) != len(positions):
        raise ValueError("Could not assign every candidate electrode to the cap.")
    return {
        "eeg_net": eeg_net,
        "pairs": [mapped_labels[i : i + 2] for i in range(0, len(mapped_labels), 2)],
        "optimized_positions": mapped["optimized_positions"],
        "mapped_positions": mapped["mapped_positions"],
        "distances": [float(distance) for distance in mapped["distances"]],
    }