pybis.py

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

"""
pybis.py

Work with openBIS using Python.

"""

from __future__ import print_function
from .utils import (
    extract_attr,
    extract_permid,
    extract_code,
    extract_deletion,
    extract_identifier,
    extract_nested_identifier,
    extract_nested_permid,
    extract_nested_permids,
    extract_identifiers,
    extract_property_assignments,
    extract_role_assignments,
    extract_person,
    extract_person_details,
    extract_id,
    extract_userId,
    is_number,
)
from datetime import datetime
import pandas as pd
from pandas import DataFrame, Series
from .semantic_annotation import SemanticAnnotation
from .tag import Tag
from .role_assignment import RoleAssignment
from .group import Group
from .person import Person
from .dataset import DataSet
from .sample import Sample
from .experiment import Experiment
from .project import Project
from .space import Space
from .things import Things
from .definitions import (
    openbis_definitions,
    get_definition_for_entity,
    fetch_option,
    get_fetchoption_for_entity,
    get_type_for_entity,
    get_method_for_entity,
    get_fetchoptions,
)
from .openbis_object import OpenBisObject, Transaction
from .vocabulary import Vocabulary, VocabularyTerm
from .entity_type import (
    EntityType,
    SampleType,
    DataSetType,
    MaterialType,
    ExperimentType,
)
from .utils import (
    parse_jackson,
    check_datatype,
    split_identifier,
    format_timestamp,
    is_identifier,
    is_permid,
    nvl,
    VERBOSE,
)
from . import data_set as pbds
from tabulate import tabulate
from texttable import Texttable
from collections import namedtuple, defaultdict
import zlib
from urllib.parse import urlparse, urljoin, quote
import re
import json
import time
import os
import random
import subprocess
import errno

import requests

import urllib3
import logging
import sys


# import the various openBIS entities

LOG_NONE = 0
LOG_SEVERE = 1
LOG_ERROR = 2
LOG_WARNING = 3
LOG_INFO = 4
LOG_ENTRY = 5
LOG_PARM = 6
LOG_DEBUG = 7

DEBUG_LEVEL = LOG_NONE

def now():
    return time.time_ns() // 1000000

def get_search_type_for_entity(entity, operator=None):
    """Returns a dictionary containing the correct search criteria type
    for a given entity.

    Example::
        get_search_type_for_entity('space')
        # returns:
        {'@type': 'as.dto.space.search.SpaceSearchCriteria'}
    """
    search_criteria = {
        "space": "as.dto.space.search.SpaceSearchCriteria",
        "userId": "as.dto.person.search.UserIdSearchCriteria",
        "email": "as.dto.person.search.EmailSearchCriteria",
        "firstName": "as.dto.person.search.FirstNameSearchCriteria",
        "lastName": "as.dto.person.search.LastNameSearchCriteria",
        "project": "as.dto.project.search.ProjectSearchCriteria",
        "experiment": "as.dto.experiment.search.ExperimentSearchCriteria",
        "experiment_type": "as.dto.experiment.search.ExperimentTypeSearchCriteria",
        "sample": "as.dto.sample.search.SampleSearchCriteria",
        "sample_type": "as.dto.sample.search.SampleTypeSearchCriteria",
        "dataset": "as.dto.dataset.search.DataSetSearchCriteria",
        "dataset_type": "as.dto.dataset.search.DataSetTypeSearchCriteria",
        "external_dms": "as.dto.externaldms.search.ExternalDmsSearchCriteria",
        "material": "as.dto.material.search.MaterialSearchCriteria",
        "material_type": "as.dto.material.search.MaterialTypeSearchCriteria",
        "vocabulary_term": "as.dto.vocabulary.search.VocabularyTermSearchCriteria",
        "tag": "as.dto.tag.search.TagSearchCriteria",
        "authorizationGroup": "as.dto.authorizationgroup.search.AuthorizationGroupSearchCriteria",
        "person": "as.dto.person.search.PersonSearchCriteria",
        "code": "as.dto.common.search.CodeSearchCriteria",
        "global": "as.dto.global.GlobalSearchObject",
        "plugin": "as.dto.plugin.search.PluginSearchCriteria",
        "propertyType": "as.dto.property.search.PropertyTypeSearchCriteria",
    }

    sc = {"@type": search_criteria[entity]}
    if operator is not None:
        sc["operator"] = operator

    return sc


def _type_for_id(ident, entity):
    ident = ident.strip()
    """Returns the data type for a given identifier/permId for use with the API call, e.g.
    {
        "identifier": "/DEFAULT/SAMPLE_NAME",
        "@type": "as.dto.sample.id.SampleIdentifier"
    }
    or
    {
        "permId": "20160817175233002-331",
        "@type": "as.dto.sample.id.SamplePermId"
    }
    """
    # Tags have strange permIds...
    if entity.lower() == "tag":
        if "/" in ident:
            if not ident.startswith("/"):
                ident = "/" + ident
            return {"permId": ident, "@type": "as.dto.tag.id.TagPermId"}
        else:
            return {"code": ident, "@type": "as.dto.tag.id.TagCode"}

    entities = {
        "sample": "Sample",
        "dataset": "DataSet",
        "experiment": "Experiment",
        "plugin": "Plugin",
        "space": "Space",
        "project": "Project",
        "semanticannotation": "SemanticAnnotation",
    }
    search_request = {}
    if entity.lower() in entities:
        entity_capitalize = entities[entity.lower()]
    else:
        entity_capitalize = entity.capitalize()

    if is_identifier(ident):
        # people tend to omit the / prefix of an identifier...
        if not ident.startswith("/"):
            ident = "/" + ident
        # ELN-LIMS style contains also experiment in sample identifer, i.e. /space/project/experiment/sample_code
        # we have to remove the experiment-code
        if ident.count("/") == 4:
            codes = ident.split("/")
            ident = "/".join([codes[0], codes[1], codes[2], codes[4]])

        search_request = {
            "identifier": ident.upper(),
            "@type": "as.dto.{}.id.{}Identifier".format(