pybis.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
pybis.py

Work with openBIS from Python.

"""

import os
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

import time
from datetime import datetime
import json
import re
from urllib.parse import urlparse

import pandas as pd
from pandas import DataFrame, Series

import threading
from threading import Thread
from queue import Queue
DROPBOX_PLUGIN = "jupyter-uploader-api"

search_for_type = {
    "space":      "as.dto.space.search.SpaceSearchCriteria",
    "project":    "as.dto.project.search.ProjectSearchCriteria",
    "experiment": "as.dto.experiment.search.ExperimentSearchCriteria",
    "code":       "as.dto.common.search.CodeSearchCriteria",
    "sample_type":"as.dto.sample.search.SampleTypeSearchCriteria",
}

fetch_option = {
    "properties":  { "@type": "as.dto.property.fetchoptions.PropertyFetchOptions" },
    "tags":        { "@type": "as.dto.tag.fetchoptions.TagFetchOptions" },
    "project":     { "@type": "as.dto.project.fetchoptions.ProjectFetchOptions" },
    "experiment":  { "@type": "as.dto.experiment.fetchoptions.ExperimentFetchOptions" },
    "registrator": { "@type": "as.dto.person.fetchoptions.PersonFetchOptions" },
    "modifier":    { "@type": "as.dto.person.fetchoptions.PersonFetchOptions" },
    "physicalData":{ "@type": "as.dto.dataset.fetchoptions.PhysicalDataFetchOptions" },
}


def format_timestamp(ts):
    return datetime.fromtimestamp(round(ts/1000)).strftime('%Y-%m-%d %H:%M:%S')

def extract_code(obj):
    return obj['code']

def extract_identifier(ident):
    if not isinstance(ident, dict): 
        return str(ident)
    return ident['identifier']

def extract_permid(permid):
    if not isinstance(permid, dict):
        return str(permid)
    return permid['permId']

def extract_nested_permid(permid):
    if not isinstance(permid, dict):
        return str(permid)
    return permid['permId']['permId']

def extract_nested_identifier(ident):
    if not isinstance(ident, dict): 
        return str(ident)
    return ident['identifier']['identifier']

def extract_person(person):
    if not isinstance(person, dict):
        return str(person)
    if 'email' in person and person['email'] is not '':
        return "%s %s <%s>" % (person['firstName'], person['lastName'], person['email'])
    else:
        return "%s %s" % (person['firstName'], person['lastName'])

def extract_properties(prop):
    if isinstance(prop, dict):
        newline = "; "
        props = []
        for key in prop:
            props.append("%s: %s" % (key, prop[key]))
        return newline.join(props)


def _criteria_for_code(code, object_type):
    criteria = {
        "criteria": [
            {
                "fieldName": "code",
                "fieldType": "ATTRIBUTE",
                "fieldValue": {
                "value": code,
                "@type": "as.dto.common.search.StringEqualToValue"
                },
                "@type": "as.dto.common.search.CodeSearchCriteria"
            }
        ],
        "@type": search_for_type[object_type],
        "operator": "AND"
    }
    return criteria

class Openbis:
    """Interface for communicating with openBIS. A current version of openBIS is needed (at least version 16.05).

    """

    def __init__(self, url='https://localhost:8443', verify_certificates=True, token=None):
        """Initialize a new connection to an openBIS server.

        :param host:
        """

        url_obj = urlparse(url)
        if  url_obj.netloc is None:
            raise ValueError("please provide the url in this format: https://openbis.host.ch:8443")

        self.url_obj = url_obj
        self.url     = url_obj.geturl()
        self.port    = url_obj.port
        self.hostname = url_obj.hostname
        self.as_v3 = '/openbis/openbis/rmi-application-server-v3.json'
        self.as_v1 = '/openbis/openbis/rmi-general-information-v1.json'
        self.reg_v1 = '/openbis/openbis/rmi-query-v1.json'
        self.verify_certificates = verify_certificates
        self.token = token
        self.datastores = []
        self.spaces = None
        self.dataset_types = None
        self.sample_types = None
        self.files_in_wsp = []
        self.token_path = None

        # some default settings for working with samples etc.
        self.default_space = None
        self.default_project = None
        self.default_experiment = None
        self.default_sample_type = None

        # use an existing token, if available
        if self.token is None:
            self.token = self._get_cached_token()


    def _get_cached_token(self):
        """Read the token from the cache, and set the token ivar to it, if there, otherwise None.
        If the token is not valid anymore, delete it. 
        """
        token_path = self.gen_token_path()
        if not os.path.exists(token_path):
            return None
        try:
            with open(token_path) as f:
                token = f.read()
                if not self.is_token_valid(token):
                    os.remove(token_path)
                    return None
                else:
                    return token
        except FileNotFoundError:
            return None


    def gen_token_path(self, parent_folder=None):
        """generates a path to the token file.
        The token is usually saved in a file called
        ~/.pybis/hostname.token
        """
        if parent_folder is None:
            # save token under ~/.pybis folder
            parent_folder = os.path.join(
                os.path.expanduser("~"),
                '.pybis'
            )
        path = os.path.join(parent_folder, self.hostname + '.token')
        return path


    def save_token(self, token=None, parent_folder=None):
        """ saves the session token to the disk, usually here: ~/.pybis/hostname.token. When a new Openbis instance is created, it tries to read this saved token by default.
        """
        if token is None:
            token = self.token

        token_path = None;
        if parent_folder is None:
            token_path = self.gen_token_path()
        else:
            token_path = self.gen_token_path(parent_folder)

        # create the necessary directories, if they don't exist yet
        os.makedirs(os.path.dirname(token_path), exist_ok=True)
        with open(token_path, 'w') as f: