Import many SWC files into

import_skeletons.py

#!/usr/bin/env python3
"""Script for uploading multiple neurons to CATMAID from SWC files.

Assumes all of your SWC files are in a directory, and called 'my_neuron_name.swc'.

Requirements:
  python >= 3.6
  catpy (pip install catpy)
  tqdm (pip install tqdm)
"""
import os
from typing import List, Dict, Union, Any, Optional
from pathlib import Path

from catpy.applications import CatmaidClientApplication
from tqdm import tqdm

MY_SWC_DIR = "path/to/directory"  # directory containing SWC files
ANNOTATIONS = ["imported"]  # list of strings to annotate uploaded skeletons with
MY_CREDS = "path/to/credentials.json"  # path to credentials file, with a format like


# {
#   "base_url": "https://catmaid.servername.com/catmaid_subdirectory/",
#   "project_id": 1,
#   "token": "your_catmaid_token", # see https://catmaid.readthedocs.io/en/stable/api.html#api-token; must have "import" permission
#   "auth_name": "http_auth_username",  # delete if you don't use HTTP auth
#   "auth_pass": "http_auth_password"   # as above
# }


class SkeletonImporter(CatmaidClientApplication):
    def upload_swc(self, fpath: Union[str, os.PathLike], **kwargs) -> Dict[str, Any]:
        """Takes path to SWC file, plus POST parameters described in the skeletons/import endpoint.
        Returns response dict as described.

        https://github.com/catmaid/CATMAID/blob/master/django/applications/catmaid/control/skeleton.py#L2376
        """
        fname = os.path.basename(fpath)
        with open(fpath, 'rb') as f:
            result = self.post((self.project_id, "skeletons", "import"), data=kwargs,
                               files={fname: f})
        return result

    def annotate_skeletons(self, skids: List[int], annotations: List[str],
                           meta: Optional[List[str]] = None):
        """Annotate many skeletons with many annotations.

        Those annotations can optionally be annotated by passing a list of meta-annotations.
        """
        data = {"annotations": annotations, "skeleton_ids": skids}
        if meta:
            data["meta_annotations"] = meta

        response = self.post((self.project_id, "annotations", "add"), data)
        return response

    def import_and_annotate(self, swc_dir, annotations=None):
        skeleton_ids = []
        results = []
        for fpath in tqdm(list(Path(swc_dir).glob('*.swc'))):
            result = self.upload_swc(fpath, name=fpath.stem)
            skeleton_ids.append(result["skeleton_id"])
            results.append(result)

        if annotations:
            self.annotate_skeletons(skeleton_ids, annotations)

        print(f"Imported {len(skeleton_ids)} skeletons")
        return results


if __name__ == "__main__":
    imp = SkeletonImporter.from_json(MY_CREDS)
    imp.import_and_annotate(MY_SWC_DIR, ANNOTATIONS)