WIP: Feature/Analysis 2d graphs and nodeports package (#114)

added notebooks for #3 use case for displaying 2D graphs
enhanced simcore-sdk/nodeports package to connect to S3, postgres, download from S3 and compatible with the docker swarm.

Author: sanderegg

.env-devel

@@ -1,3 +1,4 @@
+POSTGRES_ENDPOINT=postgres:5432
 POSTGRES_USER=simcore
 POSTGRES_PASSWORD=simcore
 POSTGRES_DB=simcoredb

Makefile

@@ -4,7 +4,7 @@
 
 PY_FILES = $(strip $(shell find services packages -iname '*.py'))
 
-export PYTHONPATH=${PWD}/packages/s3wrapper/src
+export PYTHONPATH=${CURDIR}/packages/s3wrapper/src:${CURDIR}/packages/simcore-sdk/src
 
 build-devel:
 	docker-compose -f services/docker-compose.yml -f services/docker-compose.devel.yml build
@@ -26,19 +26,18 @@ up:
 
 up-swarm:
 	docker swarm init
-	docker-compose -f services/docker-compose.yml -f services/docker-compose.deploy.yml up
+	docker stack deploy -c services/docker-compose.yml -c services/docker-compose.deploy.yml workbench
 
 down:
 	docker-compose -f services/docker-compose.yml down
 	docker-compose -f services/docker-compose.yml -f services/docker-compose.devel.yml down
 
 down-swarm:
-	docker-compose -f services/docker-compose.yml -f services/docker-compose.deploy.yml down
+	docker stack rm workbench
 	docker swarm leave -f
 
 stack-up:
 	docker swarm init
-	POSTGRES_USER=simcore POSTGRES_PASSWORD=simcore POSTGRES_DB=simcoredb RABBITMQ_USER=simcore RABBITMQ_PASSWORD=simcore RABBITMQ_PROGRESS_CHANNEL=comp.backend.channels.progress RABBITMQ_LOG_CHANNEL=comp.backend.channels.log S3_ENDPOINT=minio:9000 S3_ACCESS_KEY=12345678  S3_SECRET_KEY=12345678 S3_BUCKET_NAME=simcore docker stack deploy -c services/docker-compose.yml -c services/docker-compose.deploy.yml osparc
 
 stack-down:
 	docker stack rm osparc

packages/s3wrapper/src/s3wrapper/s3_client.py

@@ -6,7 +6,6 @@
 from minio import Minio
 from minio.error import ResponseError
 
-logging.basicConfig(level=logging.DEBUG)
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -122,9 +121,9 @@ def get_metadata(self, bucket_name, object_name):
 
         return {}
 
-    def list_objects(self, bucket_name, recursive=False):
+    def list_objects(self, bucket_name, prefix=None, recursive=False):
         try:
-            return self.client.list_objects(bucket_name, recursive=recursive)
+            return self.client.list_objects(bucket_name, prefix=prefix, recursive=recursive)
         except ResponseError as _err:
             logging.exception("Could not list objects")
 
@@ -152,7 +151,7 @@ def exists_object(self, bucket_name, object_name, recursive=False):
         ''' This seems to be pretty heavy, should be used with care
         '''
         try:
-            objects = self.list_objects(bucket_name, recursive)
+            objects = self.list_objects(bucket_name, recursive=recursive)
             for obj in objects:
                 if obj.object_name == object_name:
                     return True

packages/s3wrapper/tests/test_s3_client.py

@@ -104,7 +104,7 @@ def test_file_upload_download(s3_client, bucket, text_files):
     filepath = text_files(1)[0]
     object_name = "1"
     assert s3_client.upload_file(bucket, object_name, filepath)
-    filepath2 = filepath + "."
+    filepath2 = filepath + ".rec"
     assert s3_client.download_file(bucket, object_name, filepath2)
     assert filecmp.cmp(filepath2, filepath)
 
@@ -173,7 +173,7 @@ def test_presigned_put(s3_client, bucket, text_files):
         with urllib.request.urlopen(req) as _f:
             pass
 
-    filepath2 = filepath + "."
+    filepath2 = filepath + ".rec"
     assert s3_client.download_file(bucket, object_name, filepath2)
     assert filecmp.cmp(filepath2, filepath)
 
@@ -233,3 +233,28 @@ def test_object_exists(s3_client, bucket, text_files):
     assert s3_client.upload_file(bucket, object_name, file2)
     assert not s3_client.exists_object(bucket, object_name, False)
     assert s3_client.exists_object(bucket, object_name, True)
+
+def test_list_objects(s3_client, bucket, text_files):
+    files = text_files(2)
+    file1 = files[0]
+    file2 = files[1]
+    object_name = "level1/level2/1"
+    assert s3_client.upload_file(bucket, object_name, file1)
+    object_name = "level2/level2/2"
+    assert s3_client.upload_file(bucket, object_name, file2)
+    
+    listed_objects = s3_client.list_objects(bucket)    
+    for s3_obj in listed_objects:
+        assert s3_obj.object_name == "level1/" or s3_obj.object_name == "level2/"
+    
+    listed_objects = s3_client.list_objects(bucket, prefix="level1")
+    for s3_obj in listed_objects:
+        assert s3_obj.object_name == "level1/"
+
+    listed_objects = s3_client.list_objects(bucket, prefix="level1", recursive=True)
+    for s3_obj in listed_objects:
+        assert s3_obj.object_name == "level1/level2/1"
+    
+    listed_objects = s3_client.list_objects(bucket, recursive=True)
+    for s3_obj in listed_objects:
+        assert s3_obj.object_name == "level1/level2/1" or s3_obj.object_name == "level2/level2/2"

packages/simcore-sdk/requirements.txt

@@ -1,2 +1,4 @@
 psycopg2==2.7.4
 sqlalchemy==1.2.5
+networkx==2.1
+tenacity==4.12.0

packages/simcore-sdk/src/simcore_sdk/config/connection_config.json

@@ -0,0 +1,31 @@
+{
+    "version":"0.1",
+    "inputs": [
+        {
+            "key": "in_1",
+            "label": "computational data",
+            "desc": "these are computed data out of a pipeline",
+            "type": "file-url",
+            "value": "/home/jovyan/data/outputControllerOut.dat",
+            "timestamp": "2018-05-23T15:34:53.511Z"
+        },
+        {
+            "key": "in_5",
+            "label": "some number",
+            "desc": "numbering things",
+            "type": "int",
+            "value": "666",
+            "timestamp": "2018-05-23T15:34:53.511Z"
+        }
+    ],
+    "outputs": [
+        {
+            "key": "out_1",
+            "label": "some boolean output",
+            "desc": "could be true or false...",
+            "type": "bool",
+            "value": "null",
+            "timestamp": "2018-05-23T15:34:53.511Z"
+        }
+    ]
+}

packages/simcore-sdk/src/simcore_sdk/config/db.py

@@ -3,7 +3,7 @@
 """
 from os import environ as env
 
-POSTGRES_URL = "postgres:5432"
+POSTGRES_URL = env.get("POSTGRES_ENDPOINT", "postgres:5432")
 POSTGRES_USER = env.get("POSTGRES_USER", "simcore")
 POSTGRES_PW = env.get("POSTGRES_PASSWORD", "simcore")
 POSTGRES_DB = env.get("POSTGRES_DB", "simcoredb")

packages/simcore-sdk/src/simcore_sdk/nodeports/__init__.py

@@ -0,0 +1,11 @@
+#pylint: disable=C0111
+#pylint: disable=C0103
+# import os
+# from simcore_sdk.nodeports import config
+# from simcore_sdk.nodeports import io
+# from simcore_sdk.nodeports import serialization
+
+# _CONFIG = config.CONFIG[os.environ.get("simcoreapi_CONFIG", "default")]
+# _IO = io.IO(config=_CONFIG)
+# # create initial Simcore object
+# PORTS = serialization.create_from_json(_IO, auto_read=True, auto_write=True)

packages/simcore-sdk/src/simcore_sdk/nodeports/_item.py

@@ -0,0 +1,86 @@
+"""This module contains an item representing a node port"""
+
+import logging
+import collections
+import datetime
+from simcore_sdk.nodeports import exceptions
+from simcore_sdk.nodeports import config
+from simcore_sdk.nodeports import filemanager
+
+_LOGGER = logging.getLogger(__name__)
+
+_DataItem = collections.namedtuple("_DataItem", config.DATA_ITEM_KEYS)
+
+class DataItem(_DataItem):
+    """This class encapsulate a Data Item and provide accessors functions"""
+    def __new__(cls, **kwargs):        
+        _LOGGER.debug("Creating new data item with %s", kwargs)
+        self = super(DataItem, cls).__new__(cls, **kwargs)
+        self.new_data_cb = None
+        self.get_node_from_uuid_cb = None
+        return self
+
+    def get(self):
+        """returns the data converted to the underlying type.
+
+            Can throw InvalidPtrotocolError if the underling type is unknown.
+            Can throw ValueError if the conversion fails.
+            returns the converted value or None if no value is defined
+        """
+        _LOGGER.debug("Getting data item")
+        if self.type not in config.TYPE_TO_PYTHON_TYPE_MAP:
+            raise exceptions.InvalidProtocolError(self.type)
+        if self.value == "null":
+            _LOGGER.debug("Got empty data item")
+            return None
+        _LOGGER.debug("Got data item with value %s", self.value)
+
+        if isinstance(self.value, str) and self.value.startswith("link."):
+            link = self.value.split(".")
+            if len(link) != 3:
+                raise exceptions.InvalidProtocolError(self.value, "Invalid link definition: " + str(self.value))
+            other_node_uuid = link[1]
+            other_port_key = link[2]
+
+            if self.type in config.TYPE_TO_S3_FILE_LIST:
+                # try to fetch from S3 as a file
+                _LOGGER.debug("Fetch file from S3 %s", self.value)
+                return filemanager.download_file_from_S3(node_uuid=other_node_uuid, 
+                                                        node_key=other_port_key, 
+                                                        file_name=self.key)
+            elif self.type in config.TYPE_TO_S3_FOLDER_LIST:
+                # try to fetch from S3 as a folder
+                _LOGGER.debug("Fetch folder from S3 %s", self.value)
+                return filemanager.download_folder_from_s3(node_uuid=other_node_uuid, 
+                                                            node_key=other_port_key, 
+                                                            folder_name=self.key)
+            else:
+                # try to fetch link from database node
+                _LOGGER.debug("Fetch value from other node %s", self.value)
+                if not self.get_node_from_uuid_cb:
+                    raise exceptions.NodeportsException("callback to get other node information is not set")
+
+                other_nodeports = self.get_node_from_uuid_cb(other_node_uuid) #pylint: disable=not-callable
+                _LOGGER.debug("Received node from DB %s, now returning value", other_nodeports)
+                return other_nodeports.get(other_port_key)
+
+
+        return config.TYPE_TO_PYTHON_TYPE_MAP[self.type](self.value)
+
+    def set(self, value):
+        """sets the data to the underlying port
+        
+        Arguments:
+            value {any type} -- must be convertible to a string, or an exception will be thrown.
+        """
+        _LOGGER.info("Setting data item with value %s", value)
+        # let's create a new data
+        data_dct = self._asdict()
+        new_value = str(value)
+        if new_value != data_dct["value"]:
+            data_dct["value"] = str(value)
+            data_dct["timestamp"] = datetime.datetime.utcnow().isoformat()
+            new_data = DataItem(**data_dct)
+            if self.new_data_cb:
+                _LOGGER.debug("calling new data callback")
+                self.new_data_cb(new_data) #pylint: disable=not-callable

packages/simcore-sdk/src/simcore_sdk/nodeports/_itemslist.py

@@ -0,0 +1,137 @@
+"""This module defines a container for port descriptions"""
+
+# pylint: disable=too-many-ancestors
+import logging
+from collections import MutableSequence
+from simcore_sdk.nodeports import exceptions
+from simcore_sdk.nodeports._item import DataItem
+
+_LOGGER = logging.getLogger(__name__)
+
+class DataItemsList(MutableSequence): 
+    """This class contains a list of Data Items."""
+
+    def __init__(self, data=None, read_only=False, change_cb=None, get_node_from_node_uuid_cb=None):
+        _LOGGER.debug("Creating DataItemsList with %s", data)
+        if data is None:
+            data = []
+        
+        data_keys = set()
+        for item in data:
+            if not isinstance(item, DataItem):
+                raise TypeError
+            data_keys.add(item.key)
+        # check uniqueness... we could use directly a set for this as well
+        if len(data_keys) != len(data):
+            raise exceptions.InvalidProtocolError(data)
+        self.__lst = data
+        self.read_only = read_only
+
+        self.__change_notifier = change_cb
+        self.__get_node_from_node_uuid_cb = get_node_from_node_uuid_cb
+        self.__assign_change_notifier_to_data()        
+    
+    @property
+    def change_notifier(self):
+        """Callback function to be set if client code wants notifications when
+        an item is modified or replaced"""
+        return self.__change_notifier
+    
+    @change_notifier.setter
+    def change_notifier(self, value):
+        self.__change_notifier = value
+        self.__assign_change_notifier_to_data()
+
+    @property
+    def get_node_from_node_uuid_cb(self):
+        return self.__get_node_from_node_uuid_cb
+
+    @get_node_from_node_uuid_cb.setter
+    def get_node_from_node_uuid_cb(self, value):
+        self.__get_node_from_node_uuid_cb = value
+        self.__assign_change_notifier_to_data()
+
+    def __setitem__(self, index, value):
+        _LOGGER.debug("Setting item %s with %s", index, value)
+        if self.read_only:
+            raise exceptions.ReadOnlyError(self)
+        if not isinstance(value, DataItem):
+            raise TypeError
+        if isinstance(index, str):
+            # it might be a key            
+            index = self.__find_index_from_key(index)
+        if not index < len(self.__lst):
+            raise exceptions.UnboundPortError(index)    
+        # check for uniqueness
+        stored_index = self.__find_index_from_key(value.key)
+        if stored_index != index:
+            # not the same key not allowed
+            raise exceptions.InvalidProtocolError(value._asdict())
+        self.__lst[index] = value
+        self.__assign_change_notifier_to_data()
+        self.__notify_client()
+        
+
+    def __notify_client(self):
+        if self.change_notifier and callable(self.change_notifier):
+            self.change_notifier() #pylint: disable=not-callable
+
+    def __getitem__(self, index):
+        _LOGGER.debug("Getting item %s", index)
+        if isinstance(index, str):
+            # it might be a key
+            index = self.__find_index_from_key(index)            
+        if index < len(self.__lst):
+            return self.__lst[index]
+        raise exceptions.UnboundPortError(index)
+
+    def __len__(self):        
+        return len(self.__lst)
+
+    def __delitem__(self, index):
+        _LOGGER.debug("Deleting item %s", index)
+        if self.read_only:
+            raise exceptions.ReadOnlyError(self)
+        del self.__lst[index]
+
+    def insert(self, index, value):
+        _LOGGER.debug("Inserting item %s at %s", value, index)
+        if self.read_only:
+            raise exceptions.ReadOnlyError(self)
+        if not isinstance(value, DataItem):
+            raise TypeError
+        if self.__find_index_from_key(value.key) < len(self.__lst):
+            # the key already exists
+            raise exceptions.InvalidProtocolError(value._asdict())
+        self.__lst.insert(index, value)
+        self.__assign_change_notifier_to_data()
+        self.__notify_client()
+
+    def __find_index_from_key(self, item_key):
+        indices = [index for index in range(0, len(self.__lst)) if self.__lst[index].key == item_key]
+        if not indices:
+            return len(self.__lst)            
+        if len(indices) > 1:
+            raise exceptions.InvalidProtocolError(indices)
+        return indices[0]
+
+    def __assign_change_notifier_to_data(self):
+        for data in self.__lst:
+            data.new_data_cb = self.__item_value_updated_cb
+            data.get_node_from_uuid_cb = self.get_node_from_node_uuid_cb
+
+    def __item_value_updated_cb(self, new_data_item):
+        # a new item shall replace the current one
+        self.__replace_item(new_data_item)
+        self.__notify_client()
+
+    def __replace_item(self, new_data_item):
+        item_index = self.__find_index_from_key(new_data_item.key)
+        self.__lst[item_index] = new_data_item
+
+    @classmethod
+    def __check_uniqueness(cls, dataitems):
+        data_keys = set([item.key for item in dataitems])
+        if len(dataitems) != len(data_keys):
+            raise exceptions.InvalidProtocolError(dataitems)
+