crantpy.utils package

crantpy.utils package#

Subpackages#

crantpy.utils.cave package
- Submodules
- Module contents

Module contents#

exception crantpy.utils.FilteringError(message=None)[source]#

Bases: ValueError

Raised if a filtering operation fails.

Parameters:: message (str, optional) – The error message.
Return type:: None

exception crantpy.utils.NoMatchesError(message=None)[source]#

Bases: ValueError

Raised if no matches are found.

Parameters:: message (str, optional) – The error message.
Return type:: None

crantpy.utils.add_annotation_layer(annotations, scene, name=None, connected=False)[source]#

Add annotations as new layer to scene.

Parameters:

annotations (array or list) – Coordinates for annotations (in voxel space): - (N, 3): Point annotations at x/y/z coordinates - (N, 2, 3): Line segments with start and end points - (N, 4): Ellipsoids with x/y/z center and radius
scene (dict) – Scene to add annotation layer to.
name (str, optional) – Name for the annotation layer.
connected (bool, default False) – If True, point annotations will be connected as a path (TODO).

Returns:

Modified scene with annotation layer added.

Return type:

dict

Examples

>>> # Add point annotations
>>> points = np.array([[100, 200, 50], [150, 250, 60]])
>>> scene = add_annotation_layer(points, scene, name="my_points")
>>>
>>> # Add line annotations
>>> lines = np.array([
...     [[100, 200, 50], [150, 250, 60]],
...     [[150, 250, 60], [200, 300, 70]]
... ])
>>> scene = add_annotation_layer(lines, scene, name="my_lines")

crantpy.utils.add_skeleton_layer(skeleton, scene, name=None)[source]#

Add skeleton as line annotation layer to scene.

Parameters:

skeleton (TreeNeuron or DataFrame) – Neuron skeleton to add. Coordinates must be in nanometers. Will be automatically converted to voxel space.
scene (dict) – Scene to add skeleton layer to.
name (str, optional) – Name for the skeleton layer.

Returns:

Modified scene with skeleton layer added.

Return type:

dict

Examples

>>> skeleton = crt.viz.get_skeletons([720575940621039145])[0]
>>> scene = construct_scene()
>>> scene = add_skeleton_layer(skeleton, scene)

crantpy.utils.cached_per_id(cache_name, id_param='x', max_age=7200, result_id_column='old_id')[source]#

Decorator for caching function results on a per-ID basis.

This decorator caches results for individual IDs rather than entire function calls. When the function is called with a list of IDs, it will: 1. Check which IDs have valid cached results 2. Only call the function for uncached IDs 3. Merge cached and new results 4. Cache the new results

This is particularly useful for functions like update_ids() where we want to avoid re-computing results for IDs we’ve already processed.

Parameters:

cache_name (str) – Name of the global cache to use.
id_param (str, default 'x') – Name of the parameter containing the IDs to cache.
max_age (int, default MAXIMUM_CACHE_DURATION) – Maximum age of cached results in seconds.
result_id_column (str, default 'old_id') – Column name in the result DataFrame that contains the ID.

Returns:

The decorated function with per-ID caching capabilities.

Return type:

callable

Notes

The decorated function must return a pandas DataFrame
The ID parameter can be a list, array, or single ID
Cache entries are stored with timestamps for staleness checking
The function gains a clear_cache method to manually clear the cache

Examples

>>> @cached_per_id(cache_name="update_ids_cache", id_param="x")
>>> def update_ids(x, dataset=None):
>>>     # Process IDs
>>>     return pd.DataFrame({'old_id': x, 'new_id': x, 'changed': False})
>>>
>>> # First call - computes all IDs
>>> result1 = update_ids([1, 2, 3])
>>>
>>> # Second call - uses cached results for IDs 2 and 3
>>> result2 = update_ids([2, 3, 4])

crantpy.utils.cached_result(cache_name, max_age=7200, key_fn=None, should_cache_fn=None, validate_cache_fn=None)[source]#

Decorator for caching function results.

WARNING: This decorator is not thread-safe. It is recommended to use threading.Lock() to ensure thread safety when using this decorator in a multi-threaded environment.

This decorator provides a flexible caching mechanism for function results. It supports custom cache keys, validation, and conditional caching, making it suitable for a variety of use cases.

The cache stores entries in a dictionary structure: {

‘result’: original_function_result, ‘metadata’: {

‘_created_at’: timestamp

}

}

This approach avoids modifying the original result objects directly, ensuring compatibility with immutable types.

Parameters:

cache_name (str) – Name of the global cache to use. This is used to group cached results under a specific namespace.
max_age (int, default MAXIMUM_CACHE_DURATION) – Maximum age of cached result in seconds. Cached results older than this duration are considered stale and will be refreshed.
key_fn (callable, optional) – Function to generate a unique cache key based on the function’s arguments. Defaults to using the first positional argument or the ‘dataset’ keyword argument. If the function returns None, an error will be raised.
should_cache_fn (callable, optional) – Function to determine whether the result of the function should be cached. It takes the function result and arguments as input and returns a boolean.
validate_cache_fn (callable, optional) – Function to validate if a cached result is still valid beyond the age check. It takes the cached result and the function arguments as input and returns a boolean.

Returns:

The decorated function with caching capabilities.

Return type:

callable

Examples

>>> # Basic Caching:
>>> @cached_result(cache_name="example_cache")
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Custom Cache Key:
>>> @cached_result(cache_name="example_cache", key_fn=lambda x: f"key_{x}")
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Conditional Caching:
>>> @cached_result(cache_name="example_cache", should_cache_fn=lambda result, *args: result > 10)
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Cache Validation:
>>> def validate_cache(result, *args):
>>>     return result is not None

>>> @cached_result(cache_name="example_cache", validate_cache_fn=validate_cache)
>>> def expensive_function(x):
>>>     return x ** 2

Notes

The decorated function gains a clear_cache method to manually clear the cache for the specified cache_name.
The check_stale parameter can be used to skip staleness checks when calling the decorated function.

crantpy.utils.clear_all_caches()[source]#

Clears all caches.

Return type:: None

crantpy.utils.clear_cave_client_cache()[source]#

Clears the CAVE client cache.

Return type:: None

crantpy.utils.clear_cloudvolume_cache()[source]#

Clears the cloudvolume cache.

Return type:: None

crantpy.utils.clear_global_cache(cache_name)[source]#

Clear a named global cache.

Parameters:: cache_name (str) – Name of the cache to clear
Return type:: None

crantpy.utils.construct_scene(*, image=True, segmentation=True, brain_mesh=True, merge_biased_seg=False, nuclei=False, base_neuroglancer=False, layout='xy-3d', dataset=None)[source]#

Construct a basic neuroglancer scene for CRANT data.

Parameters:

image (bool, default True) – Whether to add the aligned EM image layer.
segmentation (bool, default True) – Whether to add the proofreadable segmentation layer.
brain_mesh (bool, default True) – Whether to add the brain mesh layer.
merge_biased_seg (bool, default False) – Whether to add the merge-biased segmentation layer (for proofreading).
nuclei (bool, default False) – Whether to add the nuclei segmentation layer.
base_neuroglancer (bool, default False) – Whether to use base neuroglancer (affects segmentation layer format).
layout (str, default "xy-3d") – Layout to show. Options: “3d”, “xy-3d”, “xy”, “4panel”.
dataset (str, optional) – Which dataset to use (“latest” or “sandbox”). If None, uses default.

Returns:

Neuroglancer scene dictionary with requested layers.

Return type:

dict

Examples

>>> # Create a minimal visualization scene
>>> scene = construct_scene(image=True, segmentation=True, brain_mesh=True)
>>>
>>> # Create a full proofreading scene
>>> scene = construct_scene(
...     image=True,
...     segmentation=True,
...     brain_mesh=True,
...     merge_biased_seg=True,
...     nuclei=True
... )

crantpy.utils.create_sql_query(table_name, fields, condition=None, limit=None, start=None)[source]#

Creates a SQL query to get the specified fields from the specified table.

Parameters:

table_name (str) – The name of the table to query.
fields (List[str]) – The list of field names to include in the query.
condition (str, optional) – The WHERE clause of the query.
limit (int, optional) – The maximum number of rows to return.
start (int, optional) – The number of rows to skip (OFFSET).

Returns:

The constructed SQL query string.

Return type:

str

crantpy.utils.decode_url(url, format='json')[source]#

Decode neuroglancer URL to extract information.

Parameters:

url (str or list of str) – Neuroglancer URL(s) to decode.
format (str, default "json") – Output format: - “json”: Full scene dictionary - “brief”: Dict with position, selected segments, and annotations - “dataframe”: DataFrame with segment IDs and their layers

Returns:

Decoded information in requested format.

Return type:

dict or DataFrame

Examples

>>> url = "https://spelunker.cave-explorer.org/#!{...}"
>>> info = decode_url(url, format='brief')
>>> print(info['selected'])  # List of selected segment IDs
>>> print(info['position'])  # [x, y, z] coordinates

crantpy.utils.encode_url(segments=None, annotations=None, coords=None, skeletons=None, skeleton_names=None, seg_colors=None, seg_groups=None, invis_segs=None, scene=None, base_neuroglancer=False, layout='xy-3d', open=False, to_clipboard=False, shorten=False, *, dataset=None)[source]#

Encode data as CRANT neuroglancer scene URL.

Parameters:

segments (int or list of int, optional) – Segment IDs (root IDs) to have selected in the scene.
annotations (array or dict, optional) – Coordinates for annotations: - (N, 3) array: Point annotations at x/y/z coordinates (in voxels) - dict: Multiple annotation layers {name: (N, 3) array}
coords ((3,) array, optional) – X, Y, Z coordinates (in voxels) to center the view on.
skeletons (TreeNeuron or NeuronList, optional) – Skeleton(s) to add as annotation layer(s). Must be in nanometers.
skeleton_names (str or list of str, optional) – Names for the skeleton(s) to display in the UI. If a single string is provided, it will be used for all skeletons. If a list is provided, its length must match the number of skeletons.
seg_colors (str, tuple, list, dict, or array, optional) – Colors for segments: - str or tuple: Single color for all segments - list: List of colors matching segments - dict: Mapping of segment IDs to colors - array: Labels that will be converted to colors
seg_groups (list or dict, optional) – Group segments into separate layers: - list: Group labels matching segments - dict: {group_name: [seg_id1, seg_id2, …]}
invis_segs (int or list, optional) – Segment IDs to select but keep invisible.
scene (dict or str, optional) – Existing scene to modify (as dict or URL string).
base_neuroglancer (bool, default False) – Whether to use base neuroglancer instead of CAVE Spelunker.
layout (str, default "xy-3d") – Layout to show. Options: “3d”, “xy-3d”, “xy”, “4panel”.
open (bool, default False) – If True, opens the URL in a web browser.
to_clipboard (bool, default False) – If True, copies the URL to clipboard (requires pyperclip).
shorten (bool, default False) – If True, creates a shortened URL (requires state server).
dataset (str, optional) – Which dataset to use. If None, uses default.

Returns:

Neuroglancer URL.

Return type:

str

Examples

>>> # Simple scene with segments
>>> url = encode_url(segments=[720575940621039145, 720575940621039146])
>>>
>>> # Scene with colored segments
>>> url = encode_url(
...     segments=[720575940621039145, 720575940621039146],
...     seg_colors={720575940621039145: 'red', 720575940621039146: 'blue'}
... )
>>>
>>> # Scene with skeleton and centered view
>>> import navis
>>> skeleton = crt.viz.get_skeletons([720575940621039145])[0]
>>> url = encode_url(
...     segments=[720575940621039145],
...     skeletons=skeleton,
...     coords=[24899, 14436, 3739]
... )

crantpy.utils.filter_df(df, column, value, regex=False, case=False, match_all=False, exact=True)[source]#

This function filters a df based on a column and a value. It can handle string, numeric, and list-containing columns.

Parameters:

(pd.DataFrame) (df)
(str) (column)
(Any) (value)
(bool) (exact)
(bool)
(bool) – if True, requires all filter values to be present in the cell’s list. If False, requires at least one filter value to be present. Defaults to False.
(bool)
df (pandas.DataFrame)
column (str)
value (Any)
regex (bool)
case (bool)
match_all (bool)
exact (bool)

Returns:

pd.DataFrame

Return type:

The filtered df.

crantpy.utils.generate_cave_token(save=False)[source]#

Generates a token for the CAVE client. If save is True, the token will be saved (overwriting any existing token).

Parameters:: save (bool, default False) – Whether to save the token after generation.
Return type:: None

crantpy.utils.get_cave_client(dataset=None, clear_cache=False, check_stale=True)[source]#

Returns a CAVE client instance. If a token is already set, it will be used for authentication. Otherwise, a new token will be generated.

Parameters:

clear_cache (bool, default False) – If True, bypasses the cache and fetches a new client.
check_stale (bool, default True) – If True, checks if the cached client is stale based on materialization and maximum cache duration.
dataset (str, optional) – The dataset to use. If not provided, uses the default dataset.

Returns:

A CAVE client instance authenticated with the token.

Return type:

CAVEclient

Raises:

ValueError – If no token is found after attempting to generate one.

crantpy.utils.get_cave_datastacks()[source]#

Get available CAVE datastacks.

Return type:: list

crantpy.utils.get_cloudvolume(dataset=None, clear_cache=False, check_stale=True, **kwargs)[source]#

Returns a cloudvolume instance.

Parameters:

dataset (str | None)
clear_cache (bool)
check_stale (bool)

Return type:

caveclient.CAVEclient

crantpy.utils.get_current_cave_token()[source]#

Retrieves the current token from the CAVE client.

Returns:: The current CAVE token.
Return type:: str
Raises:: ValueError – If no token is found.

crantpy.utils.get_dataset_segmentation_source(dataset)[source]#

Get segmentation source for given dataset.

Parameters:: dataset (str)
Return type:: str

crantpy.utils.get_datastack_segmentation_source(datastack)[source]#

Get segmentation source for given CAVE datastack.

Return type:: str

crantpy.utils.get_global_cache(cache_name)[source]#

Get a named global cache dictionary.

Parameters:: cache_name (str) – Name of the cache to retrieve
Returns:: The requested cache dictionary
Return type:: dict

crantpy.utils.inject_dataset(allowed=None, disallowed=None, param_name='dataset')[source]#

Inject current default dataset.

Parameters:

allowed (List[str] or str, optional) – List of allowed datasets or a single allowed dataset.
disallowed (List[str] or str, optional) – List of disallowed datasets or a single disallowed dataset.
param_name (str, default 'dataset') – Name of the parameter to inject the dataset into.

Returns:

Decorator function that injects the dataset.

Return type:

Callable

crantpy.utils.is_latest_roots(x, timestamp=None, dataset=None, progress=True, batch_size=100000, validate_ids=True, use_http_session=True)[source]#

Check if the given root IDs are the latest based on the timestamp.

Parameters:

x (IDs = str | int | np.int64) – The root IDs to check.
timestamp (Timestamp = str | int | np.int64 | datetime | np.datetime64 | pd.Timestamp) – The timestamp to compare against. Can also be “mat” for the latest materialization timestamp.
dataset (str, optional) – The dataset to use.
progress (bool, default True) – Whether to show progress bar for large batches.
batch_size (int, default 100_000) – Batch size for processing large numbers of IDs.
validate_ids (bool, default True) – Whether to validate root IDs before processing.
use_http_session (bool, default True) – Whether to use direct HTTP session for better performance.

Returns:

A boolean array indicating whether each root ID is the latest.

Return type:

np.ndarray

Examples

>>> from crantpy.utils.cave.helpers import is_latest_roots
>>> is_latest_roots([123456789, 987654321])
array([ True, False])

>>> # Check against latest materialization
>>> is_latest_roots([123456789], timestamp="mat")
array([ True])

crantpy.utils.is_valid_root(x, dataset=None, raise_exc=False)[source]#

Check if ID is (potentially) valid root ID.

Parameters:

x (IDs = str | int | np.int64) – The root IDs to check.
dataset (str, optional) – The dataset to use.
raise_exc (bool, default False) – Whether to raise an exception if invalid IDs are found.

Returns:

A boolean array indicating whether each root ID is valid.

Return type:

np.ndarray

Raises:

ValueError – If raise_exc is True and invalid IDs are found.

crantpy.utils.is_valid_supervoxel(x, dataset=None, raise_exc=False)[source]#

Check if ID is (potentially) valid supervoxel ID.

Parameters:

x (IDs = str | int | np.int64) – The supervoxel IDs to check.
dataset (str, optional) – The dataset to use.
raise_exc (bool, default False) – Whether to raise an exception if invalid IDs are found.

Returns:

If x is a single ID, returns bool. If x is iterable, returns array.

Return type:

bool or np.ndarray

Raises:

ValueError – If raise_exc is True and invalid IDs are found.

crantpy.utils package

Contents

crantpy.utils package#

Subpackages#

Submodules#

Module contents#