neu_vol.webknossos¶

Addressing a webknossos layer, its credentials, and the few REST calls behind it.

A layer is addressed as wk://DATASET/LAYER; the server and token come from a credentials file, never from the URL or a spec. The array view is backends/webknossos.py. Stdlib only (urllib): the official webknossos client pins packages this environment holds newer versions of, and only a few routes are needed.

Supervoxels are the default: a mapping spec field (an agglomerate view) is applied by the server on every read, and leaving it unset reads the raw over-segmentation, even when the viewer shows a default mapping.

Credentials, first hit wins: $WK_TOKEN (+ $WK_URL), then $NEU_WK_CREDENTIALS, else ~/.config/webknossos/credentials.toml:

# chmod 600 — this is a secret
url = "https://webknossos.org"
token = "..."              # https://webknossos.org/account/token
organization = "..."       # optional: narrows dataset-name lookup

Attributes¶

`DATASTORE_API_VERSION`
`DEFAULT_CREDENTIALS`
`ELEMENT_CLASSES`
`ENV_CREDENTIALS`
`SCHEME`
`TABLE_BATCH`
`TIMEOUT_S`
`WK_API_VERSION`
`log`

Functions¶

`agglomerate_table`(spec, mapping, n_ids, *[, batch]) → numpy.ndarray	`table[sv] = agglomerate id` for every supervoxel id `0 .. n_ids - 1`.
`api`(url) → str
`check_mapping`(spec, mapping) → None	Refuse a mapping the layer does not have, naming the ones it does.
`clear_cache`() → None	Drop memoized dataset records and version checks (tests, long-lived processes).
`credentials`() → dict[str, str \| None]	`{"url", "token", "organization"}`, from the environment or the credentials file.
`credentials_path`() → pathlib.Path
`data_api`(datastore) → str
`dataset_info`(dataset_id, *[, creds]) → dict	The server's record for one dataset (`GET /datasets/{id}`).
`dataset_summary`(dataset_id, *[, creds]) → dict	What a dataset is, for choosing one: its record's description, uploader and dates,
`decode_list_of_long`(data) → numpy.ndarray	The `items` of a `ListOfLong`, packed or unpacked, as uint64.
`default_mapping`(layer) → str \| None	The mapping the viewer applies to this layer by default, if any.
`encode_list_of_long`(ids) → bytes	`ListOfLong` bytes for these ids, unpacked (what the server's own encoder emits).
`is_url`(location) → bool	True if this location names a webknossos layer rather than a store.
`iso_date`(ms) → str	A webknossos timestamp (ms since the epoch) as a UTC date; `-` if absent.
`layer_base`(spec) → str	The datastore URL of a resolved spec's layer; every per-layer route hangs off it.
`layer_record`(info, layer) → dict
`list_datasets`([pattern, organization, creds]) → list[dict]	The active datasets this token can see, oldest first — ONE request.
`mappings`(spec, *[, creds]) → dict	`{"agglomerate": [...], "json": [...]}` — every mapping the layer can be read with.
`parse_url`(url) → dict[str, str]	`wk://DATASET/LAYER` -> `{"dataset", "layer"}`.
`resolve_dataset_id`(dataset, *[, creds]) → str	A dataset name (or id) -> its id. An ambiguous name raises rather than picking one,
`spec_url`(spec) → str	The `wk://` URL for a spec — for messages.

Module Contents¶

neu_vol.webknossos.agglomerate_table(spec: Mapping[str, Any], mapping: str, n_ids: int, *, batch: int = TABLE_BATCH) → numpy.ndarray¶

table[sv] = agglomerate id for every supervoxel id 0 .. n_ids - 1.

Dense, indexed by supervoxel id, so applying it is table[volume]. Id 0 is background and is not asked about; it maps to 0.

neu_vol.webknossos.api(url: str) → str¶

neu_vol.webknossos.check_mapping(spec: Mapping[str, Any], mapping: str) → None¶: Refuse a mapping the layer does not have, naming the ones it does.

neu_vol.webknossos.clear_cache() → None¶: Drop memoized dataset records and version checks (tests, long-lived processes).

neu_vol.webknossos.credentials() → dict[str, str | None]¶

{"url", "token", "organization"}, from the environment or the credentials file.

Raises with the file’s expected shape when nothing is configured: the first run on a new machine is exactly when that is the useful answer.

neu_vol.webknossos.credentials_path() → pathlib.Path¶

neu_vol.webknossos.data_api(datastore: str) → str¶

neu_vol.webknossos.dataset_info(dataset_id: str, *, creds: Mapping[str, Any] | None = None) → dict¶: The server’s record for one dataset (GET /datasets/{id}).

neu_vol.webknossos.dataset_summary(dataset_id: str, *, creds: Mapping[str, Any] | None = None) → dict¶

What a dataset is, for choosing one: its record’s description, uploader and dates, and per layer the box, mags, default view and every mapping it can be read with.

One request for the record plus two per segmentation layer (the mapping lists).

neu_vol.webknossos.decode_list_of_long(data: bytes) → numpy.ndarray¶: The items of a ListOfLong, packed or unpacked, as uint64.

neu_vol.webknossos.default_mapping(layer: Mapping[str, Any]) → str | None¶: The mapping the viewer applies to this layer by default, if any.

neu_vol.webknossos.encode_list_of_long(ids: numpy.ndarray) → bytes¶: ListOfLong bytes for these ids, unpacked (what the server’s own encoder emits).

neu_vol.webknossos.is_url(location: Any) → bool¶: True if this location names a webknossos layer rather than a store.

neu_vol.webknossos.iso_date(ms: Any) → str¶: A webknossos timestamp (ms since the epoch) as a UTC date; - if absent.

neu_vol.webknossos.layer_base(spec: Mapping[str, Any]) → str¶: The datastore URL of a resolved spec’s layer; every per-layer route hangs off it.

neu_vol.webknossos.layer_record(info: Mapping[str, Any], layer: str) → dict¶

neu_vol.webknossos.list_datasets(pattern: str | None = None, *, organization: str | None = None, creds: Mapping[str, Any] | None = None) → list[dict]¶

The active datasets this token can see, oldest first — ONE request.

The server’s list includes every public dataset on it, other organizations’ too (webknossos.org answers with hundreds), so organization filters to one. Its records carry the layer names, dates and size but no description; dataset_summary adds that at one request per dataset.

pattern matches the name case-insensitively: a glob if it has *?[, else a substring.

neu_vol.webknossos.mappings(spec: Mapping[str, Any], *, creds: Mapping[str, Any] | None = None) → dict¶

{"agglomerate": [...], "json": [...]} — every mapping the layer can be read with.

Both kinds are accepted by the data route’s mappingName; agglomerate files are what a vendor delivers as views at several thresholds, JSON mappings are uploaded tables.

neu_vol.webknossos.parse_url(url: str) → dict[str, str]¶

wk://DATASET/LAYER -> {"dataset", "layer"}.

DATASET is a dataset name or its 24-hex id. The server is not part of the URL: it comes from the credentials, which is where its token is too.

neu_vol.webknossos.resolve_dataset_id(dataset: str, *, creds: Mapping[str, Any] | None = None) → str¶: A dataset name (or id) -> its id. An ambiguous name raises rather than picking one, since names are not unique across organizations.

neu_vol.webknossos.spec_url(spec: Mapping[str, Any]) → str¶: The wk:// URL for a spec — for messages.

neu_vol.webknossos.DATASTORE_API_VERSION = 15¶

neu_vol.webknossos.DEFAULT_CREDENTIALS = '~/.config/webknossos/credentials.toml'¶

neu_vol.webknossos.ELEMENT_CLASSES¶

neu_vol.webknossos.ENV_CREDENTIALS = 'NEU_WK_CREDENTIALS'¶

neu_vol.webknossos.SCHEME = 'wk://'¶

neu_vol.webknossos.TABLE_BATCH = 100000¶

neu_vol.webknossos.TIMEOUT_S = 600¶

neu_vol.webknossos.WK_API_VERSION = 16¶

neu_vol.webknossos.log¶