# Copyright (C) 2017 Philipp Schlegel
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
"""This module contains neuron and neuronlist classes returned and accepted
by many functions within pymaid. CatmaidNeuron and CatmaidNeuronList objects
also provided quick access to many other PyMaid functions.
Examples
--------
>>> # Get a bunch of neurons from CATMAID server as CatmaidNeuronList
>>> nl = pymaid.get_neuron('annotation:uPN right')
>>> # CatmaidNeuronLists work in, many ways, like pandas DataFrames
>>> nl.head()
neuron_name skeleton_id n_nodes n_connectors \
0 PN glomerulus VA6 017 DB 16 12721 1878
1 PN glomerulus VL2a 22000 JMR 21999 10740 1687
2 PN glomerulus VC1 22133 BH 22132 8446 1664
3 PN putative glomerulus VC3m 22278 AA 22277 6228 674
4 PN glomerulus DL2v 22423 JMR 22422 4610 384
n_branch_nodes n_end_nodes open_ends cable_length review_status soma
0 773 822 280 2863.743284 NA True
1 505 537 194 2412.045343 NA True
2 508 548 110 1977.235899 NA True
3 232 243 100 1221.985849 NA True
4 191 206 93 1172.948499 NA True
>>> # Plot neurons
>>> nl.plot3d()
>>> # Neurons in a list can be accessed by index, ...
>>> nl[0]
>>> # ... by skeleton ID, ...
>>> nl.skid[16]
>>> # ... or by attributes
>>> nl[nl.cable_length > 2000]
>>> # Each neuron has a bunch of useful attributes
>>> print(nl[0].skeleton_id, nl[0].soma, nl[0].n_open_ends)
>>> # Attributes can also be accessed for the entire neuronslist
>>> nl.skeleton_id
:class:`~pymaid.CatmaidNeuron` and :class:`~pymaid.CatmaidNeuronList`
also allow quick access to other PyMaid functions:
>>> # This ...
>>> pymaid.reroot_neuron(nl[0], nl[0].soma, inplace=True)
>>> # ... is essentially equivalent to this
>>> nl[0].reroot(nl[0].soma)
>>> # Similarly, CatmaidNeurons do on-demand data fetching for you:
>>> # So instead of this ...
>>> an = pymaid.get_annotations(nl[0])
>>> # ..., you can do just this:
>>> an = nl[0].annotations
"""
import datetime
import functools
import hashlib
import json
import navis
import numbers
import numpy as np
import pandas as pd
import matplotlib.colors as mcl
import navis.config as nsconfig
from . import (fetch, utils, config, client, cache)
try:
import trimesh
except ImportError:
trimesh = None
__all__ = ['CatmaidNeuron', 'CatmaidNeuronList', 'Dotprops', 'Volume']
# Set up logging
logger = config.get_logger(__name__)
def inject_connection(func):
"""Raise error if no local or global connection."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
self = args[0]
remote_instance = kwargs.get('remote_instance')
if not remote_instance:
if not self._remote_instance:
raise Exception(f"{func.__name__}: Need CatmaidInstance to "
"fetch data. Either pass to function as "
"`remote_instance` or use neuron's "
"`set_remote_instance()` method.")
remote_instance = self._remote_instance
kwargs['remote_instance'] = remote_instance
return func(*args, **kwargs)
return wrapper
[docs]class CatmaidNeuron(navis.TreeNeuron):
"""Catmaid neuron object holding neuron data (nodes, connectors, name,
etc) and providing quick access to various PyMaid functions.
CatmaidNeuron can be minimally constructed from just a skeleton ID
and a CatmaidInstance. Other parameters (nodes, connectors, neuron name,
annotations, etc.) will then be retrieved from the server 'on-demand'.
The easiest way to construct a CatmaidNeuron is by using
:func:`~pymaid.get_neuron`.
Manually, a complete CatmaidNeuron can be constructed from a pandas
DataFrame (df) containing: df.nodes, df.connectors, df.skeleton_id,
df.neuron_name, df.tags
Using a CatmaidNeuron to initialise a CatmaidNeuron will automatically
make a copy.
Attributes
----------
skeleton_id : str
This neuron's skeleton ID.
neuron_name : str
This neuron's name.
nodes : ``pandas.DataFrame``
Contains complete node table.
connectors : ``pandas.DataFrame``
Contains complete connector table.
presynapses : ``pandas.DataFrame``
All presynaptic connectors.
postsynapses : ``pandas.DataFrame``
All postsynaptic connectors.
gap_junctions : ``pandas.DataFrame``
All gap junction connectors.
date_retrieved : ``datetime`` object
Timestamp of data retrieval.
tags : dict
Node tags.
connector_tags : dict
Connector tags.
annotations : list
This neuron's annotations.
graph : ``network.DiGraph``
Graph representation of this neuron.
igraph : ``igraph.Graph``
iGraph representation of this neuron. Returns ``None``
if igraph library not installed.
review_status : int
This neuron's review status.
n_connectors : int
Total number of synapses.
n_presynapses : int
Total number of presynaptic sites.
n_postsynapses : int
Total number of presynaptic sites.
n_branch_nodes : int
Number of branch nodes.
n_end_nodes : int
Number of end nodes.
n_open_ends : int
Number of open end nodes = leaf nodes that are not
tagged with either: ``ends``, ``not a branch``,
``uncertain end``, ``soma`` or
``uncertain continuation``.
cable_length : float
Cable length in micrometers [um].
segments : list of lists
Node IDs making up linear segments. Maximizes
segment lengths (similar to CATMAID's review widget).
small_segments : list of lists
Node IDs making up linear segments between
end/branch points.
soma : node ID of soma
Returns ``None`` if no soma or 'NA' if data not
available.
root : numpy.array
Node ID(s) of root.
color : tuple
Color of neuron. Used for e.g. export to json.
partners : pd.DataFrame
Connectivity table of this neuron.
Examples
--------
>>> # Initialize a new neuron
>>> n = pymaid.CatmaidNeuron(123456)
>>> # Retrieve node data from server on-demand
>>> n.nodes
CatmaidNeuron - INFO - Retrieving skeleton data...
node_id parent_id creator_id x y z radius confidence
0 ...
>>> # Initialize with skeleton data
>>> n = pymaid.get_neuron(123456)
>>> # Get annotations from server
>>> n.annotations
['annotation1', 'annotation2']
>>> # Force update of annotations
>>> n.get_annotations()
"""
#: Minimum radius for soma detection. Set to ``None`` if no tag needed.
#: Default = 1 micron
soma_detection_radius = .5 * nsconfig.ureg.um
#: Soma radius (e.g. for plotting). If string, must be column in nodes
#: table. Default = 'radius'.
soma_radius = 'radius'
# Set default function for soma finding.
_soma = None
# Tag to detect soma
soma_detection_tag = 'soma'
#: Attributes to be used when comparing two neurons.
EQ_ATTRIBUTES = ['n_nodes', 'n_connectors', 'soma', 'root', 'skeleton_id',
'n_branches', 'n_leafs', 'cable_length', 'name']
#: Attributes used for neuron summary
SUMMARY_PROPS = ['type', 'name', 'n_nodes', 'n_connectors',
'n_branches', 'n_leafs', 'cable_length', 'soma', 'units']
# Default value for lazy loading
_lazy_loading = True
[docs] def __init__(self, x, remote_instance=None, units='nm', **metadata):
"""Initialize CatmaidNeuron.
Parameters
----------
x
Data to construct neuron from:
- `pandas.DataFrame` is expected to be SWC table
- `pandas.Series` is expected to have a DataFrame
as `.nodes` - additional properties will be
attached as meta data
- `str` is treated as SWC file name
- `BufferedIOBase` e.g. from `open(filename)`
- `networkx.DiGraph` parsed by `navis.nx2neuron`
remote_instance : CatmaidInstance, optional
Storing this makes it more convenient to retrieve
e.g. neuron annotations, review status, etc. If
not provided, will try using global CatmaidInstance.
units : str | pint.Units | pint.Quantity
Units for coordinates. Defaults to ``None`` (dimensionless).
Strings must be parsable by pint: e.g. "nm", "um",
"micrometer" or "8 nanometers".
metadata
Any additional data to attach to neuron.
"""
if (isinstance(x, str) and x.isnumeric()) or isinstance(x, numbers.Number):
# Initialize empty neuron
metadata.update({'skeleton_id': x})
super().__init__(None, units=units, **metadata)
else:
super().__init__(x, units=units, **metadata)
if remote_instance is None:
remote_instance = getattr(x, 'remote_instance', None)
remote_instance = utils._eval_remote_instance(remote_instance,
raise_error=False)
# These will be overriden if x is a CatmaidNeuron
self._remote_instance = remote_instance
self.date_retrieved = datetime.datetime.now().isoformat()
def __eq__(self, other):
"""Implement neuron comparison."""
# Deactivate lazy loading during comparison
self._lazy_loading = False
if isinstance(other, CatmaidNeuron):
other._lazy_loading = False
try:
res = super().__eq__(other)
except BaseException:
raise
finally:
self._lazy_loading = True
if isinstance(other, CatmaidNeuron):
other._lazy_loading = True
return res
def __mul__(self, other, *args, **kwargs):
"""Implement multiplication for coordinates (nodes, connectors)."""
# Exclude missing radii from multiplication
is_missing = self.nodes.radius == -1
n = super().__mul__(other, *args, **kwargs)
n.nodes.loc[is_missing, 'radius'] = -1
return n
def __truediv__(self, other, *args, **kwargs):
"""Implement division for coordinates (nodes, connectors)."""
# Exclude missing radii from division
is_missing = self.nodes.radius == -1
n = super().__truediv__(other, *args, **kwargs)
n.nodes.loc[is_missing, 'radius'] = -1
return n
def __hash__(self):
# DO NOT REMOVE THIS! When defining __eq__ in subclass, _hash__ will
# not be inherited and we have to explicitly define it
return super().__hash__()
@property
def annotations(self):
"""Neuron annotations."""
if not hasattr(self, '_annotations'):
self._annotations = self.get_annotations()
return self._annotations
@annotations.setter
def annotations(self, v):
if not isinstance(v, (list, np.ndarray)):
raise TypeError(f'Expected annotations as list or array, got {type(v)}')
self._annotations = v
@property
def core_md5(self) -> str:
"""MD5 of core information for the neuron.
Generated from ``nodes`` and ``connectors`` table.
Returns
-------
md5 : string
MD5 of node and connector table. None if no such data.
"""
data = []
if self.has_nodes:
data.append(self.nodes[['node_id', 'parent_id',
'x', 'y', 'z']].values)
if self.has_connectors:
data.append(self.connectors[['node_id',
'connector_id',
'x', 'y', 'z']].values)
if data:
data = np.ascontiguousarray(np.concatenate(data, axis=0).astype(float))
return hashlib.md5(data).hexdigest()
@property
def gap_junctions(self):
"""Table with gap junctions.
Requires a "type" column in connector table. Will look for type labels
that include "gap" or that equal 2 or "2".
"""
if not self.has_connectors:
raise ValueError('No connector table found.')
# Make an educated guess what gap junctions are
types = self.connectors['type'].unique()
gap = [t for t in types if 'gap' in str(t) or t in [2, "2"]]
if len(gap) == 0:
logger.debug(f'Unable to find gap junctions in types: {types}')
return self.connectors.iloc[0:0] # return empty DataFrame
elif len(gap) > 1:
raise ValueError(f'Found ambigous labels for gap junctions: {gap}')
return self.connectors[self.connectors['type'] == gap[0]]
@property
def name(self):
"""Neuron name."""
return self.neuron_name
@name.setter
def name(self, v):
"""Neuron name."""
self.neuron_name = v
@property
def neuron_name(self):
"""Neuron name (legacy - please use .name instead)."""
if not hasattr(self, '_name'):
if not self._lazy_loading:
return 'NA'
self._name = self.get_name()
return self._name
@neuron_name.setter
def neuron_name(self, v):
self._name = v
@property
def open_ends(self):
"""Node IDs of open ends."""
if self.has_nodes:
closed = set(self.tags.get('ends', [])
+ self.tags.get('uncertain end', [])
+ self.tags.get('uncertain continuation', [])
+ self.tags.get('not a branch', [])
+ self.tags.get('soma', []))
ends = self.ends
return ends[~ends.node_id.isin(closed)]
else:
logger.info('No skeleton data available. Use .get_skeleton() '
'to fetch.')
return 'NA'
@property
def partners(self):
"""Get connected partners."""
if not hasattr(self, '_partners'):
if not self._lazy_loading:
return 'NA'
self._partners = self.get_partners()
return self._partners
@property
def review_status(self):
"""Review status in percent."""
if not hasattr(self, '_review_status'):
if not self._lazy_loading:
return 'NA'
self._review_status = self.get_review()
return self._review_status
@property
def skeleton_id(self):
"""Skeleton ID."""
return self.id
@skeleton_id.setter
def skeleton_id(self, value):
if not isinstance(value, (str, numbers.Number)):
raise TypeError(f'Skeleton ID must be number or string, got {type(value)}')
self.id = str(value)
@property
def tags(self):
"""Node tags."""
if not hasattr(self, '_tags'):
if not self._lazy_loading:
return 'NA'
self.get_skeleton()
return self._tags
@tags.setter
def tags(self, v):
if not isinstance(v, dict):
raise TypeError(f'Expected tags as dict, got {type(v)}')
self._tags = v
@property
def connector_tags(self):
"""Connector tags."""
if not hasattr(self, '_connector_tags'):
if not self._lazy_loading:
return 'NA'
self.get_connector_tags()
return self._connector_tags
@connector_tags.setter
def connector_tags(self, v):
if not isinstance(v, dict):
raise TypeError(f'Expected connector tags as dict, got {type(v)}')
self._connector_tags = v
@property
def type(self) -> str:
"""Return type."""
return 'CatmaidNeuron'
def _get_nodes(self):
# This function redefines how the .nodes property is retrieved
if not hasattr(self, '_nodes'):
if not self._lazy_loading:
return 'NA'
self.get_skeleton()
return self._nodes
def _get_connectors(self):
# This function redefines how the .connectors property is retrieved
if not hasattr(self, '_connectors'):
if not self._lazy_loading:
return 'NA'
self.get_skeleton()
return self._connectors
[docs] def copy(self, deepcopy=False):
"""Return a copy of this neuron."""
# Use TreeNeuron's copy method
x = super().copy(deepcopy=deepcopy)
# Remote instance is excluded from copy -> otherwise we are *silently*
# creating a new CatmaidInstance that will be identical to the original
# but will have it's own cache!
x._remote_instance = self._remote_instance
return x
@inject_connection
def get_skeleton(self, remote_instance=None, **fetch_kwargs):
"""Get/update skeleton data for neuron.
Parameters
----------
**fetch_kwargs
Will be passed to :func:`pymaid.get_neuron` e.g. to get
the full node history use::
n.get_skeleton(with_history = True)
or to get abutting connectors::
n.get_skeleton(get_abutting = True)
See Also
--------
:func:`~pymaid.get_neuron`
Function called to get skeleton information
"""
func = cache.never_cache(fetch.get_neuron)
skeleton = func(self.skeleton_id,
remote_instance=remote_instance,
return_df=True,
fetch_kwargs=fetch_kwargs).iloc[0]
self._nodes = skeleton.nodes
self._connectors = skeleton.connectors
self._tags = skeleton.tags
self._name = skeleton.neuron_name
self.date_retrieved = datetime.datetime.now().isoformat()
# Delete outdated attributes
self._clear_temp_attr()
if 'type' not in self.nodes:
navis.classify_nodes(self)
return
@inject_connection
def get_connector_tags(self, remote_instance=None, **fetch_kwargs):
"""Fetch tags on connectors of a neuron.
After running, ``neuron.connector_tags`` will store a list of
tag->connector IDs mappings analogous to ``neuron.tags``.
"""
logger.debug('Retrieving connector tags...')
self._connector_tags = fetch.get_connector_tags(self,
remote_instance=remote_instance)
return
def _soma(self):
"""Search for soma and return node ID of soma.
Uses either a node tag or node radius or a combination of both
to identify the soma. This is set in the class attributes
``soma_detection_radius`` and ``soma_detection_tag``. The default
values for these are::
soma_detection_radius = 100
soma_detection_tag = 'soma'
Returns
-------
node_id
Returns node ID if soma was found, None if no soma.
"""
tn = self.nodes
if self.soma_detection_radius:
is_large = tn.radius.values * self.units >= self.soma_detection_radius
tn = tn[is_large]
if tn.empty:
return None
if self.soma_detection_tag:
if self.soma_detection_tag not in self.tags:
return None
else:
tn = tn[tn.node_id.isin(self.tags[self.soma_detection_tag])]
if tn.empty:
return None
return tn.node_id.values
@inject_connection
def get_partners(self, remote_instance=None):
"""Get connectivity table for this neuron."""
# Get partners
func = cache.never_cache(fetch.get_partners)
self._partners = func(self.skeleton_id, remote_instance=remote_instance)
return self.partners
@inject_connection
def get_review(self, remote_instance=None):
"""Get/Update review status for neuron."""
func = cache.never_cache(fetch.get_review)
self._review_status = func(self.skeleton_id,
remote_instance=remote_instance).loc[0, 'percent_reviewed']
return self._review_status
@inject_connection
def get_annotations(self, remote_instance=None):
"""Retrieve annotations for neuron."""
func = cache.never_cache(fetch.get_annotations)
self._annotations = func(self.skeleton_id,
remote_instance=remote_instance).get(str(self.skeleton_id), [])
return self.annotations
@inject_connection
def get_name(self, remote_instance=None):
"""Retrieve/update name of neuron."""
func = cache.never_cache(fetch.get_names)
self._name = func(self.skeleton_id,
remote_instance=remote_instance)[str(self.skeleton_id)]
return self.name
[docs] @inject_connection
def reload(self, remote_instance=None):
"""Reload neuron from server.
Currently only updates name, nodes, connectors and tags, not e.g.
annotations.
"""
func = cache.never_cache(fetch.get_neuron)
n = func(self.skeleton_id,
remote_instance=remote_instance)
self.__dict__.update(n.__dict__)
# Clear temporary attributes
self._clear_temp_attr()
def set_remote_instance(self, remote_instance=None, api_token=None,
server_url=None, http_user=None, http_password=None):
"""Assign remote_instance to neuron.
Provide either existing CatmaidInstance OR your credentials.
Parameters
----------
remote_instance : pymaid.CatmaidInstance, optional
server_url : str, optional
api_token : str, optional
http_user : str, optional
http_password : str, optional
See Also
--------
:class:`~pymaid.CatmaidInstance`
"""
if remote_instance:
self._remote_instance = remote_instance
elif server_url:
self._remote_instance = client.CatmaidInstance(server=server_url,
api_token=api_token,
http_user=http_user,
http_password=http_password
)
else:
raise ValueError('Provide either CatmaidInstance or credentials.')
[docs] def summary(self, add_props=None):
"""Get a summary of this neuron."""
# Suppress lazy loading during summary
self._lazy_loading = False
try:
res = super().summary()
except BaseException:
raise
finally:
self._lazy_loading = True
return res
def to_dataframe(self):
"""Turn this CatmaidNeuron into a pandas DataFrame with original CATMAID data.
Returns
-------
pandas DataFrame
neuron_name skeleton_id nodes connectors tags
0
"""
return pd.DataFrame([[self.neuron_name, self.skeleton_id,
self.nodes, self.connectors, self.tags]],
columns=['neuron_name', 'skeleton_id', 'nodes',
'connectors', 'tags'])
[docs]class CatmaidNeuronList(navis.NeuronList):
"""Compilation of :class:`~pymaid.CatmaidNeuron` that allow quick
access to neurons' attributes/functions. They are designed to work in many
ways much like a pandas.DataFrames by, for example, supporting ``.iloc[ ]``,
``.itertuples()``, ``.empty`` or ``.copy()``.
CatmaidNeuronList can be minimally constructed from just skeleton IDs.
Other parameters (nodes, connectors, neuron name, annotations, etc.)
will then be retrieved from the server 'on-demand'.
The easiest way to get a CatmaidNeuronList is by using
:func:`~pymaid.get_neuron` (see examples).
Attributes
----------
skeleton_id : array of str
name : array of str
nodes : ``pandas.DataFrame``
Merged node table.
connectors : ``pandas.DataFrame``
Merged connector table. This also works for
`presynapses`, `postsynapses` and `gap_junctions`.
tags : np.array of dict
Node tags.
annotations : np.array of list
partners : pd.DataFrame
Connectivity table for these neurons.
graph : np.array of ``networkx`` graph objects
igraph : np.array of ``igraph`` graph objects
review_status : np.array of int
n_connectors : np.array of int
n_presynapses : np.array of int
n_postsynapses : np.array of int
n_branch_nodes : np.array of int
n_end_nodes : np.array of int
n_open_ends : np.array of int
cable_length : np.array of float
Cable lengths in micrometers [um].
soma : np.array of node IDs
root : np.array of node IDs
n_cores : int
Number of cores to use. Default ``os.cpu_count()-1``.
use_threading : bool (default=True)
If True, will use parallel threads. Should be slightly
up to a lot faster depending on the numbers of cores.
Switch off if you experience performance issues.
Examples
--------
>>> # Initialize with just a Skeleton ID
>>> nl = pymaid.CatmaidNeuronList([123456, 45677])
>>> # Retrieve review status from server on-demand
>>> nl.review_status
array([90, 23])
>>> # Initialize with skeleton data
>>> nl = pymaid.get_neuron([123456, 45677])
>>> # Get annotations from server
>>> nl.annotations
[['annotation1', 'annotation2'], ['annotation3', 'annotation4']]
>>> Index using node count
>>> subset = nl[nl.n_nodes > 6000]
>>> # Get neuron by its skeleton ID
>>> n = nl.skid[123456]
>>> # Index by multiple skeleton ID
>>> subset = nl[['123456', '45677']]
>>> # Index by neuron name
>>> subset = nl['name1']
>>> # Index using annotation
>>> subset = nl['annotation:uPN right']
>>> # Concatenate lists
>>> nl += pymaid.get_neuron([912345])
"""
[docs] def __init__(self, x, make_copy=False, **kwargs):
if isinstance(x, pd.DataFrame):
# Break DataFrame into Series
x = [x.iloc[i] for i in range(x.shape[0])]
super().__init__(x, make_copy=make_copy, make_using=CatmaidNeuron, **kwargs)
# Legacy indexer
self.skid = self.idx
@property
def _remote_instance(self):
"""Find a single remote instance for all neurons in this NeuronList."""
all_instances = [n._remote_instance for n in self.neurons if hasattr(n, '_remote_instance')]
if len(set(all_instances)) > 1:
# Note that multiprocessing causes remote_instances to be pickled
# and thus not be the same anymore
logger.debug('Neurons are using multiple remote_instances! '
'Returning first entry.')
elif len(set(all_instances)) == 0:
logger.warning('No remote_instance found. Use '
'.set_remote_instance() to assign one to all '
'neurons.')
return None
return all_instances[0]
def copy(self, **kwargs):
"""Make copy of this neuronlist."""
c = super().copy(**kwargs)
c.skid = c.idx
return c
def reload(self):
"""Update neuron skeletons from server."""
self.get_skeletons(skip_existing=False)
def get_annotations(self, skip_existing=False):
"""Get/update annotations for neurons."""
if self.empty:
logger.warning('Unable to fetch annotations: CatmaidNeuronList is empty.')
return
if skip_existing:
to_update = [n.skeleton_id for n in self.neurons if 'annotations' not in n.__dict__]
else:
to_update = self.skeleton_id.tolist()
if to_update:
func = cache.never_cache(fetch.get_annotations)
annotations = func(to_update,
remote_instance=self._remote_instance)
for n in self.neurons:
n.annotations = annotations.get(str(n.skeleton_id), [])
def get_names(self, skip_existing=False):
"""Get/update neuron names."""
if self.empty:
logger.warning('Unable to fetch names: CatmaidNeuronList is empty.')
return
if skip_existing:
to_update = [n.skeleton_id for n in self.neurons if 'neuron_name' not in n.__dict__]
else:
to_update = self.skeleton_id.tolist()
if to_update:
func = cache.never_cache(fetch.get_names)
names = func(to_update, remote_instance=self._remote_instance)
for n in self.neurons:
if str(n.skeleton_id) in names:
n.neuron_name = names[str(n.skeleton_id)]
def get_skeletons(self, skip_existing=False):
"""Fill in/update skeleton data of neurons.
Updates ``.nodes``, ``.connectors``, ``.tags``, ``.date_retrieved`` and
``.neuron_name``. Will also generate new graph representation to match
nodes/connectors.
"""
if self.empty:
logger.warning('Unable to fetch skeletons: CatmaidNeuronList is empty.')
return
if skip_existing:
to_update = [n for n in self.neurons if 'nodes' not in n.__dict__]
else:
to_update = self.neurons
if to_update:
func = cache.never_cache(fetch.get_neuron)
skdata = func([n.skeleton_id for n in to_update],
remote_instance=self._remote_instance,
return_df=True).set_index('skeleton_id')
for n in config.tqdm(to_update,
desc='Processing neurons',
disable=config.pbar_hide,
leave=config.pbar_leave):
n.nodes = skdata.loc[str(n.skeleton_id), 'nodes']
n.connectors = skdata.loc[str(n.skeleton_id), 'connectors']
n.tags = skdata.loc[str(n.skeleton_id), 'tags']
n.neuron_name = skdata.loc[str(n.skeleton_id), 'neuron_name']
n.date_retrieved = datetime.datetime.now().isoformat()
# Delete and update attributes
n._clear_temp_attr()
def get_review(self, skip_existing=False):
"""Get/update review status."""
if self.empty:
logger.warning('CatmaidNeuronList is empty - no review status to fetch.')
return
if skip_existing:
to_update = [n.skeleton_id for n in self.neurons if 'review_status' not in n.__dict__]
else:
to_update = self.skeleton_id.tolist()
if to_update:
func = cache.never_cache(fetch.get_review)
rev = func(to_update, remote_instance=self._remote_instance)
rev.set_index('skeleton_id', inplace=True)
for n in self.neurons:
if str(n.skeleton_id) in rev:
n.review_status = rev.loc[str(n.skeleton_id),
'percent_reviewed']
else:
logger.warning('No review status found for neuron '
f'{n.skeleton_id}')
def set_remote_instance(self, remote_instance=None, server_url=None,
api_token=None, http_user=None, http_password=None):
"""Assign remote_instance to all neurons.
Provide either existing CatmaidInstance OR your credentials.
Parameters
----------
remote_instance : pymaid.CatmaidInstance, optional
server_url : str, optional
api_token : str, optional
http_user : str, optional
http_password : str, optional
"""
if not remote_instance and server_url and api_token:
remote_instance = client.CatmaidInstance(server_url,
api_token=api_token,
http_user=http_user,
http_password=http_password)
elif not remote_instance:
raise Exception('Provide either CatmaidInstance or credentials.')
for n in self.neurons:
n._remote_instance = remote_instance
[docs] def summary(self, N=None, add_props=[]):
"""Get summary over all neurons in this NeuronList.
Parameters
----------
N : int | slice, optional
If int, get only first N entries.
add_props : list, optional
Additional properties to add to summary. If attribute not
available will return 'NA'.
Returns
-------
pandas DataFrame
"""
if not self.empty:
# Fetch a union of all summary props (keep order)
all_props = [p for l in self.SUMMARY_PROPS for p in l]
props = np.unique(all_props)
props = sorted(props, key=lambda x: all_props.index(x))
else:
props = []
# Add skeleton ID
props = np.insert(props, 2, 'skeleton_id')
if add_props:
props = np.append(props, add_props)
if not isinstance(N, slice):
N = slice(N)
try:
for n in self.neurons:
n._lazy_loading = False
summary = pd.DataFrame(data=[[getattr(n, a, 'NA') for a in props]
for n in self.neurons[N]],
columns=props)
except BaseException:
raise
finally:
for n in self.neurons:
n._lazy_loading = True
return summary
[docs] def has_annotation(self, x, intersect=False, partial=False,
raise_not_found=True):
"""Filter neurons by their annotations.
Parameters
----------
x : str | list of str
Annotation(s) to filter for. Use tilde (~) as prefix
to look for neurons WITHOUT given annotation(s).
intersect : bool, optional
If True, neuron must have ALL positive annotations to
be included and ALL negative (~) annotations to be
excluded. If False, must have at least one positive
to be included and one of the negative annotations
to be excluded.
partial : bool, optional
If True, allow partial match of annotation.
raise_not_found : bool, optional
If True, will raise exception if no match is found.
If False, will simply return empty list.
Returns
-------
:class:`pymaid.CatmaidNeuronList`
Neurons that have given annotation(s).
Examples
--------
>>> # Get neurons that have "test1" annotation
>>> nl.has_annotation('test1')
>>> # Get neurons that have either "test1" or "test2"
>>> nl.has_annotation(['test1', 'test2'])
>>> # Get neurons that have BOTH "test1" and "test2"
>>> nl.has_annotation(['test1', 'test2'], intersect=True)
>>> # Get neurons that have "test1" but NOT "test2"
>>> nl.has_annotation(['test1', '~test2'])
"""
# This makes sure nobody accidentally forgets brackets around
# multiple annotations
for v in [intersect, partial, raise_not_found]:
if not isinstance(v, bool):
raise TypeError('Expected boolean, got {}'.format(type(v)))
inc, exc = utils._eval_conditions(x)
if not inc and not exc:
raise ValueError('Must provide at least a single annotation')
# Make sure we have annotations to begin with
self.get_annotations(skip_existing=True)
selection = []
for n in self.neurons:
if inc:
if not partial:
pos = [a in n.annotations for a in inc]
else:
pos = [any(a in b for b in n.annotations) for a in inc]
# Skip if any positive annotation is missing
if intersect and not all(pos):
continue
# Skip if none of the positive annotations are there
elif not intersect and not any(pos):
continue
if exc:
if not partial:
neg = [a in n.annotations for a in exc]
else:
neg = [any(a in b for b in n.annotations) for a in exc]
# Skip if all negative annotations are present
if intersect and all(neg):
continue
# Skip if any of the negative annotations are present
elif not intersect and any(neg):
continue
selection.append(n)
if not selection and raise_not_found:
raise ValueError('No neurons with matching annotation(s) found')
else:
return CatmaidNeuronList(selection, make_copy=self.copy_on_subset)
[docs] @classmethod
def from_selection(cls, fname):
"""Generate NeuronList from CATMAID JSON file."""
# Read data from file
with open(fname, 'r') as f:
data = json.load(f)
# Generate NeuronLost
nl = cls([e['skeleton_id'] for e in data])
# Parse colors
for k, e in enumerate(data):
nl[k].color = mcl.to_rgb(e['color'])
return nl
[docs] def to_selection(self, save_to='selection.json'):
"""Generate JSON file which can be loaded in CATMAID selection tables.
Uses neurons' ``.color`` attribute (if exists).
Parameters
----------
save_to : str | None, optional
Filename to save selection to. If ``None``, will
return the json data instead.
"""
colors = [getattr(n, 'color', config.default_color) for n in self.neurons]
data = [dict(skeleton_id=int(n.skeleton_id),
color=mcl.to_hex(c, keep_alpha=False),
opacity=1) for n, c in zip(self.neurons, colors)]
if save_to:
with open(save_to, 'w') as outfile:
json.dump(data, outfile)
logger.info('Selection saved as {}.'.format(save_to))
else:
return data
def to_dataframe(self):
"""Turn this CatmaidneuronList into a pandas DataFrame.
Returns
-------
pandas DataFrame
neuron_name skeleton_id nodes connectors tags
0
1
"""
return pd.DataFrame([[n.neuron_name, n.skeleton_id, n.nodes,
n.connectors, n.tags] for n in self.neurons],
columns=['neuron_name', 'skeleton_id', 'nodes',
'connectors', 'tags'])
[docs] def remove_duplicates(self, key='skeleton_id', inplace=False):
"""Remove duplicate neurons from list.
Parameters
----------
key : str | list, optional
Attribute(s) by which to identify duplicates. In case of
multiple, all attributes must match to flag a neuron as
duplicate.
inplace : bool, optional
If False will return a copy of the original with
duplicates removed.
"""
return super().remove_duplicates(key=key, inplace=inplace)
# This is for legacy but will be removed soon-ish
Dotprops = navis.Dotprops
Volume = navis.Volume