# -*- coding: utf-8 -*-
from __future__ import division, unicode_literals, absolute_import
import json
import webbrowser
from abc import ABCMeta, abstractmethod
from warnings import warn
from six import string_types, add_metaclass
from enum import IntEnum, Enum
import requests
import numpy as np
[docs]class ConnectorRelationType(Enum):
SYNAPTIC = "Synaptic"
GAP_JUNCTION = "Gap junction"
TIGHT_JUNCTION = "Tight junction"
DESMOSOME = "Desmosome"
ABUTTING = "Abutting"
ATTACHMENT = "Attachment"
SPATIAL = "Spatial"
OTHER = ""
[docs] @classmethod
def from_relation(cls, relation):
return {
ConnectorRelation.presynaptic_to: cls.SYNAPTIC,
ConnectorRelation.postsynaptic_to: cls.SYNAPTIC,
ConnectorRelation.gapjunction_with: cls.GAP_JUNCTION,
ConnectorRelation.tightjunction_with: cls.TIGHT_JUNCTION,
ConnectorRelation.desmosome_with: cls.DESMOSOME,
ConnectorRelation.abutting: cls.ABUTTING,
ConnectorRelation.attached_to: cls.ATTACHMENT,
ConnectorRelation.close_to: cls.SPATIAL,
ConnectorRelation.other: cls.OTHER
}[relation]
[docs]class ConnectorRelation(Enum):
"""Enum describing the link between a treenode and connector, i.e. the treenode is ____ to the connector.
The enum's ``name`` is CATMAID's concept of "relation name":
what is returned in the ``relation`` field of the <pid>/connectors/types/ response.
The enum's ``value`` is the ``name`` field of the <pid>/connectors/types/ response.
The mappings from relation name to relation ID are project-specific and must be fetched from CATMAID.
"""
other = ""
presynaptic_to = "Presynaptic"
postsynaptic_to = "Postsynaptic"
gapjunction_with = "Gap junction"
tightjunction_with = "Tight junction"
desmosome_with = "Desmosome"
abutting = "Abutting"
attached_to = "Attachment"
close_to = "Close to"
@property
def type(self):
return ConnectorRelationType.from_relation(self)
@property
def is_synaptic(self):
return self.type == ConnectorRelationType.SYNAPTIC
def __str__(self):
return self.value
[docs]class StackOrientation(IntEnum):
"""Can be iterated over or indexed like the lower-case string representation of the orientation"""
XY = 0
XZ = 1
ZY = 2
def __str__(self):
return self.name.lower()
[docs] @classmethod
def from_str(cls, s):
return {o.name: o for o in StackOrientation}[s.upper()]
[docs] @classmethod
def from_value(cls, value, default='xy'):
"""Convert an int, str or StackOrientation into a StackOrientation.
A NoneType ``value`` will use the default orientation."""
if value is None:
value = default
if isinstance(value, string_types):
return cls.from_str(value)
elif isinstance(value, int):
return cls(value)
else:
raise TypeError("Cannot create a StackOrientation from {}".format(type(value).__name__))
def __iter__(self):
return iter(str(self))
def __getitem__(self, item):
return str(self)[item]
def __contains__(self, item):
return item in str(self)
[docs]def make_url(base_url, *args):
"""
Given any number of URL components, join them as if they were a path regardless of trailing and prepending slashes
Examples
--------
>>> make_url('google.com', 'mail')
'google.com/mail'
>>> make_url('google.com/', '/mail')
'google.com/mail'
"""
for arg in args:
arg_str = str(arg)
joiner = '' if base_url.endswith('/') else '/'
relative = arg_str[1:] if arg_str.startswith('/') else arg_str
base_url = requests.compat.urljoin(base_url + joiner, relative)
return base_url
[docs]class WrappedCatmaidException(Exception):
exception_keys = frozenset(('traceback', 'error', 'type'))
spacer = ' '
def __init__(self, message, response):
"""
Exception wrapping a django error which results in a JSON response being returned containing information
about that error.
Parameters
----------
response : requests.Response
Response containing JSON-formatted error from Django
"""
super(WrappedCatmaidException, self).__init__(message)
self.msg = message
data = response.json()
self.traceback = data['traceback']
self.type = data['type']
self.error = data['error']
def __str__(self):
return '\n'.join([
super(WrappedCatmaidException, self).__str__(),
self.spacer + 'Response contained traceback (most recent call last):'
] + [
self.spacer + line for line in self.traceback.split('\n')
] + [
'{}{}: {}'.format(self.spacer, self.type, self.error)
]
)
[docs] @classmethod
def raise_on_error(cls, response):
if response.headers.get('content-type') == 'application/json':
data = response.json()
if isinstance(data, dict) and cls.exception_keys.issubset(data):
raise cls('Received error response from {}'.format(response.url), response)
[docs]@add_metaclass(ABCMeta)
class AbstractCatmaidClient(object):
"""
Abstract parent class for CatmaidClient and CatmaidClientApplications.
Users should not subclass this; it is provided purely as a convenience for type checking.
"""
[docs] def get(self, relative_url, params=None, raw=False, **kwargs):
"""
Get data from a running instance of CATMAID.
Parameters
----------
relative_url : str or tuple of str
URL to send the request to, relative to the base_url. If a tuple is passed, its elements will be joined
with '/'.
params: dict or str, optional
JSON-like key/value data to be included in the get URL (defaults to empty)
raw: bool, optional
Whether to return the response as a string regardless of its content-type (by default, JSON responses will
be parsed)
kwargs
Extra keyword arguments to pass to `requests.Session.get()`
Returns
-------
dict or str
Data returned from CATMAID: type depends on the 'raw' parameter.
"""
return self.fetch(relative_url, method='GET', data=params, raw=raw, **kwargs)
[docs] def post(self, relative_url, data=None, raw=False, **kwargs):
"""
Post data to a running instance of CATMAID.
Parameters
----------
relative_url : str or tuple of str
URL to send the request to, relative to the base_url. If a tuple is passed, its elements will be joined
with '/'.
data: dict or str, optional
JSON-like key/value data to be included in the request as a payload (defaults to empty)
raw: bool, optional
Whether to return the response as a string regardless of its content-type (by default, JSON responses will
be parsed)
kwargs
Extra keyword arguments to pass to `requests.Session.post()`
Returns
-------
dict or str
Data returned from CATMAID: type depends on the 'raw' parameter.
"""
return self.fetch(relative_url, method='POST', data=data, raw=raw, **kwargs)
[docs] @abstractmethod
def fetch(self, relative_url, method='GET', data=None, raw=False, **kwargs):
pass
[docs]class CatmaidClient(AbstractCatmaidClient):
"""
Python object handling authentication, request pooling etc. for requests made to a CATMAID server.
Users creating their own interface should not subclass this, but instead subclass CatmaidClientApplication, which
wraps a CatmaidClient object. This composition approach eases testing and sharing CatmaidClient instances among
different interfaces.
"""
def __init__(self, base_url, token=None, auth_name=None, auth_pass=None, project_id=None):
"""
Instantiate CatmaidClient object for handling requests to a CATMAID server.
Parameters
----------
base_url : str
URL at which CATMAID server is running
token : str
API token as assigned by CATMAID server
auth_name : str
HTTP auth username
auth_pass : str
HTTP auth password
project_id : int
(Optional)
"""
self.base_url = base_url
self._session = requests.Session()
if auth_name is not None and auth_pass is not None:
self.set_http_auth(auth_name, auth_pass)
if token is not None:
self.set_api_token(token)
self.project_id = project_id
[docs] def set_http_auth(self, username, password):
"""
Set HTTP authorization for CatmaidClient in place.
Parameters
----------
username : str
HTTP authorization username
password : str
HTTP authorization password
Returns
-------
CatmaidClient
Reference to the same, now-authenticated CatmaidClient instance
"""
self._session.auth = (username, password)
return self
[docs] def set_api_token(self, token):
"""
Set CatmaidClient to use the given API token in place.
Parameters
----------
token : str
API token associated with your CATMAID account
Returns
-------
CatmaidClient
Reference to the same, now-authenticated CatmaidClient instance
"""
self._session.headers['X-Authorization'] = 'Token ' + token
return self
def _make_request_url(self, arg):
"""
Create an absolute request URL for the CATMAID server.
Parameters
----------
arg : str or tuple of str
Relative URL (to the base_url). If a tuple is passed, its elements will be joined with '/'.
Returns
-------
str
"""
if isinstance(arg, string_types):
return make_url(self.base_url, arg)
else:
return make_url(self.base_url, *arg)
[docs] @classmethod
def from_json(cls, credentials):
"""
Return a CatmaidClient instance with credentials matching those in a JSON file. Should have the property
`base_url` as a minimum.
If HTTP authentication is required, should have the properties `auth_name` and `auth_pass`.
If you intend to use an authorized CATMAID account (required for some endpoints), should have the property
`token`.
Can optionally include the property `project_id`.
Parameters
----------
credentials : str or dict
Path to the JSON credentials file, or a dict representing the object
Returns
-------
CatmaidClient
Instance of the API, authenticated with the encoded credentials
"""
if not isinstance(credentials, dict):
with open(str(credentials)) as f:
credentials = json.load(f)
return cls(
credentials['base_url'],
credentials.get('token'),
credentials.get('auth_name'),
credentials.get('auth_pass'),
credentials.get('project_id')
)
[docs] def fetch(self, relative_url, method='GET', data=None, raw=False, **kwargs):
"""
Interact with the CATMAID server in a manner very similar to the javascript CATMAID.fetch API.
Parameters
----------
relative_url : str or tuple of str
URL to send the request to, relative to the base_url. If a tuple is passed, its elements will be joined
with '/'.
method: {'GET', 'POST'}, optional
HTTP method to use (the default is 'GET')
data: dict or str, optional
JSON-like key/value data to be included in the request as a payload (defaults to empty)
raw: bool, optional
Whether to return the response as a string regardless of its content-type (by default, JSON responses will
be parsed)
kwargs
Extra keyword arguments to pass to `requests.Session.get/post()`, depending on `method`
Returns
-------
dict or list or str
Data returned from CATMAID. JSON responses will be parsed unless `raw` is `True`; all other responses
will be returned as strings.
"""
url = self._make_request_url(relative_url)
data = data or dict()
if method.upper() == 'GET':
response = self._session.get(url, params=data, **kwargs)
elif method.upper() == 'POST':
response = self._session.post(url, data=data, **kwargs)
else:
raise ValueError('Unknown HTTP method {}'.format(repr(method)))
response.raise_for_status()
WrappedCatmaidException.raise_on_error(response)
if response.headers['content-type'] == 'application/json' and not raw:
return response.json()
else:
return response.text
[docs]def get_typed(d, key, constructor=None, default=None):
"""
like dict.get, but if the response/default is not None, pass it to the given constructor.
Parameters
----------
d : dict
key : hashable
constructor : callable
default
Returns
-------
"""
response = d.get(key, default)
if constructor is None or response is None:
return response
else:
return constructor(response)
[docs]class CatmaidUrl(object):
tracing_tool_name = 'tracingtool'
def __init__(
self, base_url, project_id, stack_group_id=None, stack_id=None, scale=0, x=None, y=None, z=None,
tool=None, active_skeleton_id=None, active_node_id=None
):
self.base_url = base_url
self.project_id = project_id
self.default_scale = scale
self.stack_group = None
self.stack_group_scale = None
self.stacks = []
self.set_stack_group(stack_group_id, scale)
if stack_id is not None:
self.add_stack(stack_id, scale)
self.x = x
self.y = y
self.z = z
self.tool = tool
self.active_skeleton_id = active_skeleton_id
self.active_node_id = active_node_id
[docs] @classmethod
def from_catmaid(
cls, catmaid_client, stack_group_id=None, stack_id=None, scale=0, x=None, y=None, z=None,
tool=None, active_skeleton_id=None, active_node_id=None
):
"""
Instantiate CatmaidUrl based on a CATMAID interface instance.
Parameters
----------
catmaid_client : CatmaidClient or catpy.applications.base.CatmaidClientApplication
stack_group_id : int
stack_id : int
scale : float
x : float
x coordinate in project (real) space
y : float
y coordinate in project (real) space
z : float
z coordinate in project (real) space
tool : str
active_skeleton_id : int
active_node_id : int
Returns
-------
CatmaidUrl
"""
return cls(catmaid_client.base_url, catmaid_client.project_id, stack_group_id, stack_id, scale, x, y, z,
tool, active_skeleton_id, active_node_id)
[docs] @classmethod
def from_url(cls, url):
"""
Instantiate CatmaidUrl based on a URL pulled from a running CATMAID instance.
Parameters
----------
url : str
Returns
-------
CatmaidUrl
"""
base_url, args = url.split('/?')
d = dict(item.split('=') for item in args.split('&'))
kwargs = dict(
project_id=get_typed(d, 'pid', int), scale=None,
x=get_typed(d, 'xp', float), y=get_typed(d, 'yp', float), z=get_typed(d, 'zp', float),
tool=d.get('tool'),
active_skeleton_id=get_typed(d, 'active_skeleton_id', int),
active_node_id=get_typed(d, 'active_node_id', int)
)
obj = cls(base_url, **kwargs)
obj.set_stack_group(stack_group_id=int(d.get('sg')), scale=float(d.get('sgs')))
stacks = dict()
scales = dict()
for key, value in d.items():
if key.startswith('sid'):
try:
stacks[int(key[3:])] = int(value)
except ValueError:
pass
elif key.startswith('s'):
try:
scales[int(key[1:])] = int(value)
except ValueError:
pass
for idx, sid in sorted(stacks.items(), key=lambda x: (x[1], x[0])):
obj.add_stack(sid, scales.get(idx))
if obj.default_scale is None:
obj.default_scale = 0
return obj
[docs] def add_stack(self, stack_id, scale=None):
"""
Parameters
----------
stack_id : int
scale : float
Returns
-------
CatmaidUrl
A reference to itself, for chaining
"""
# todo? fetch stack ID from stack name
self.stacks.append((stack_id, scale))
if self.default_scale is None:
self.default_scale = scale
return self
[docs] def set_stack_group(self, stack_group_id, scale=None):
"""
Parameters
----------
stack_group_id : int
scale : float
Returns
-------
CatmaidUrl
A reference to itself, for chaining
"""
# todo? fetch stacks from stack group
self.stack_group = stack_group_id
self.stack_group_scale = scale
if self.default_scale is None:
self.default_scale = scale
return self
def _terminate_base_url(self):
url = self.base_url
if url.endswith('/'):
url += '?'
if not url.endswith('/?'):
url += '/?'
return url
def __str__(self):
elements = ['pid={}'.format(self.project_id)]
coords = ['{}p={}'.format(dim, float(getattr(self, dim))) for dim in 'xyz' if getattr(self, dim) is not None]
if len(coords) == 3:
elements.extend(coords)
elif coords:
warn('Only {} of 3 coordinates found, ignoring'.format(len(coords)))
if self.tool:
elements.append('tool=' + self.tool)
if self.tool == 'tracingtool':
elements.append('active_node_id={}'.format(self.active_node_id))
elements.append('active_skeleton_id={}'.format(self.active_skeleton_id))
if self.stack_group is not None:
elements.append('sg={}'.format(self.stack_group))
elements.append(
'sgs={}'.format(
float(self.stack_group_scale) if self.stack_group_scale is not None else float(self.default_scale)
)
)
if not self.stacks:
warn('No stacks added found, URL may be invalid')
for idx, (stack_id, scale) in enumerate(self.stacks):
elements.append('sid{}={}'.format(idx, stack_id))
elements.append('s{}={}'.format(idx, float(scale) if scale is not None else float(self.default_scale)))
return self._terminate_base_url() + '&'.join(elements)
[docs] def open(self):
webbrowser.open(str(self), new=2)
