Dataset¶

def FlyWire(release: str = "public", **kwargs):
    """FlyWire, the whole-brain FAFB reconstruction.

        fw = connecto.FlyWire()                  # public release, mat 783
        fw = connecto.FlyWire("production")      # live, needs permissions
    """
    from ..backends import build

    spec = {"public": FLYWIRE, "production": FLYWIRE_PRODUCTION}.get(release)
    if spec is None:
        raise ValueError(f"`release` must be 'public' or 'production', got {release!r}.")
    return build(spec, **kwargs)

def BANC(**kwargs):
    """BANC. Defaults to neuPrint; pass ``backend="cave"`` for the CAVE copy."""
    from ..backends import build

    return build(BANC_SPEC, **kwargs)

def FANC(**kwargs):
    """FANC, the female adult nerve cord. Needs the ``FANC_edit`` group."""
    from ..backends import build

    return build(FANC_SPEC, **kwargs)

def MICrONS(**kwargs):
    """MICrONS minnie65, mouse visual cortex."""
    from ..backends import build

    return build(MICRONS_SPEC, **kwargs)

def Aedes(**kwargs):
    """The Aedes aegypti mosquito brain, via CAVE."""
    from ..backends import build

    return build(AEDES_SPEC, **kwargs)

def probe_cave(datastack: str, **overrides) -> DatasetSpec:
    """Build a DatasetSpec by asking the CAVE server what it has."""
    from caveclient import CAVEclient

    from ..auth import get_token

    client = CAVEclient(datastack, auth_token=get_token("cave").token)
    info = client.info.get_datastack_info()

    voxel = (
        float(info.get("viewer_resolution_x") or 1),
        float(info.get("viewer_resolution_y") or 1),
        float(info.get("viewer_resolution_z") or 1),
    )

    tables = set(client.materialize.get_tables())

    synapse_table = info.get("synapse_table")
    nucleus = next((t for t in sorted(tables) if _NUCLEUS_RE.search(t)), None)
    proofread = next((t for t in sorted(tables) if _PROOFREAD_RE.search(t)), None)
    ann_tables = [t for t in sorted(tables) if _ANNOTATION_RE.search(t)]

    caps = {Cap.SEGMENTATION, Cap.CHUNKEDGRAPH, Cap.MESHES, Cap.NEUROGLANCER}
    if synapse_table:
        caps |= {Cap.CONNECTIVITY, Cap.SYNAPSES}
    if ann_tables:
        caps.add(Cap.ANNOTATIONS)
    if nucleus:
        caps.add(Cap.SOMAS)
    if proofread:
        caps.add(Cap.PROOFREADING)
    try:
        if client.l2cache.has_cache():
            caps |= {Cap.L2CACHE, Cap.SKELETONS}
    except Exception:  # noqa: BLE001 - absence of an L2 cache is not an error
        pass

    logger.info(
        "Probed %s: synapses=%s nucleus=%s proofreading=%s annotations=%s caps=%s",
        datastack, synapse_table, nucleus, proofread, ann_tables,
        sorted(str(c) for c in caps),
    )

    spec = DatasetSpec(
        name=datastack,
        label=info.get("description") or datastack,
        backends=(
            BackendSpec(
                "cave", datastack,
                synapse_table=synapse_table,
                nucleus_table=nucleus,
                proofreading_table=proofread,
            ),
        ),
        annotation_sources=tuple(
            AnnotationSource(t, "cave_table", t, id_column="pt_root_id")
            for t in ann_tables
        ),
        voxel_size=voxel,
        segmentation_source=info.get("segmentation_source"),
        capabilities=frozenset(caps),
    )
    return spec.evolve(**overrides) if overrides else spec

def probe_neuprint(dataset: str, server: str = "neuprint.janelia.org", **overrides) -> DatasetSpec:
    """Build a DatasetSpec by asking a neuPrint server what it has."""
    from neuprint import Client

    from ..auth import get_token
    from ..backends.neuprint.versions import available

    name = dataset.split(":")[0]
    token = get_token("neuprint", server=server).token

    versions = available(server, name, token)
    full = dataset if dataset in versions else versions[-1]

    client = Client(server, dataset=full, token=token)
    meta = client.meta
    voxel = tuple(meta.get("voxelSize") or (1, 1, 1))

    caps = {
        Cap.ANNOTATIONS, Cap.CONNECTIVITY, Cap.SYNAPSES, Cap.SYNAPSE_SCORES,
        Cap.ROI_CONN, Cap.ROIS, Cap.SKELETONS, Cap.SOMAS,
    }
    if meta.get("neuroglancerMeta"):
        caps |= {Cap.MESHES, Cap.NEUROGLANCER}

    logger.info("Probed %s/%s: voxel=%s rois=%s", server, full, voxel, len(client.primary_rois))

    spec = DatasetSpec(
        name=name,
        label=name,
        backends=(BackendSpec("neuprint", f"{server}/{full}"),),
        annotation_sources=(AnnotationSource("neuprint", "neuprint", id_column="bodyId"),),
        voxel_size=voxel,
        capabilities=frozenset(caps),
    )
    return spec.evolve(**overrides) if overrides else spec

def CAVE(datastack: str, *, version=None, annotations="auto", **overrides):
    """Any CAVE datastack, without writing a class.

    Spec fields (`fields`, `side_map`, `capabilities`, ...) can be passed as
    keyword arguments to refine what was probed.
    """
    from ..backends import build

    ds_kwargs = {"version": version, "annotations": annotations}
    spec = probe_cave(datastack, **overrides)
    return build(spec, **ds_kwargs)

def NeuPrint(dataset: str, *, server: str = "neuprint.janelia.org", version=None,
             annotations="auto", **overrides):
    """Any neuPrint dataset, without writing a class."""
    from ..backends import build

    spec = probe_neuprint(dataset, server=server, **overrides)
    return build(spec, version=version, annotations=annotations)