Nsls merge

.gitignore

@@ -9,3 +9,4 @@
 build/
 dist/
 xicam.egg_info/
+xicam.egg-info/

client/__init__.py

@@ -1,11 +1,13 @@
+from __future__ import absolute_import
 import pysftp
 import paramiko
-from globus import GLOBUSError
-from spot import SPOTError
-from newt import NEWTError
-import ssh
+from .globus import GLOBUSError
+from .spot import SPOTError
+from .newt import NEWTError
+from .databrokerclient import DBError
+from . import ssh
 
-__all__ = ['newt', 'spot', 'globus', 'sftp', 'ssh']
+__all__ = ['newt', 'spot', 'globus', 'sftp', 'ssh', 'databrokerclient']
 
 # Exceptions raised by clients that we care to handle
 EXCEPTIONS = (pysftp.ConnectionException, paramiko.ssh_exception.BadAuthenticationType,

client/dask_active_executor.py

@@ -1 +1,2 @@
+from __future__ import unicode_literals
 active_executor = None

client/dask_io_loop.py

@@ -1,4 +1,4 @@
-import distributed 
+import distributed
 from tornado.ioloop import IOLoop
 from threading import Thread
 from distributed import Scheduler, Worker, Executor
@@ -9,7 +9,8 @@
 
 __ioloop__ = IOLoop()
 
-class DaskLoop():
+
+class DaskLoop(object):
   def __init__(self):
     self.loop = __ioloop__
     self.t = Thread(target=self.loop.start)

client/dask_local_scheduler.py

@@ -2,7 +2,7 @@
 from pipeline import msg
 
 
-class LocalScheduler():
+class LocalScheduler(object):
     def __init__(self, dloop):
         """
           Create a local scheduler and worker to run commands

client/dask_remote_scheduler.py

@@ -141,7 +141,7 @@ def serve(self):
 # client.load_system_host_keys()
 # client.connect('edison.nersc.gov', username="hkrishna")
 
-class RemoteScheduler():
+class RemoteScheduler(object):
     """
        Create a remote executor
     """

client/databrokerclient.py

@@ -0,0 +1,53 @@
+from xicam import config
+
+
+class DataBrokerClient(object):  # replace with databroker client
+    def __init__(self, host, **kwargs):
+        import os
+        import numpy as np
+        # TODO: handlers will need to be registered
+        #
+        # from suitcase.als832 import ALSHDF5Handler, ALSHDF5SinoHandler
+        # from suitcase.als733 import ALSEDFHandler
+        super(DataBrokerClient, self).__init__()
+        self.host = host
+        self.port = 27017
+
+        from databroker import Broker
+
+        # TODO: implement this case when the name of the .yml file is passed
+        # db = Broker.named(str(host).lower())
+
+        db_config = {'description': 'Configuration to access the data from mongodb',
+                     'metadatastore': {
+                         'module': 'databroker.headersource.mongo',
+                         'class': 'MDS',
+                         'config': {
+                             'host': self.host,
+                             'port': self.port,
+                             'database': config.settings['Databroker MetaDataStore Name'],
+                             'timezone': 'US/Eastern'}},
+                     'assets': {
+                         'module': 'databroker.assets.mongo',
+                         'class': 'Registry',
+                         'config': {
+                             'host': self.host,
+                             'port': self.port,
+                             'database': config.settings['Databroker FileStore Name']}}
+                     }
+
+        db = Broker.from_config(db_config)
+        self.db = db
+
+    def __getitem__(self, item):
+        return self.db[item]
+
+    def __getattr__(self, item):
+        return self.db.__getattr__(item)
+
+    def __call__(self, *args, **kwargs):
+        return self.db(*args, **kwargs)
+
+
+class DBError(Exception):
+    pass

client/globus.py

@@ -177,16 +177,16 @@ def determine_local_endpoint(self):
             os.utime(test, None)
 
         for endpoint in user_endpoints:
-                params = {'path': unicode(test)}
-                r = self.get(self.TRANSFER_URL + '/endpoint/' + quote(endpoint) + '/ls',
-                             headers=self.authentication, params=params)
-                try:
-                    file_response = r.json()
-                    if 'is a file' in file_response['message']:
-                        os.remove(test)
-                        return endpoint
-                except KeyError:
-                    pass
+            params = {'path': str(test)}
+            r = self.get(self.TRANSFER_URL + '/endpoint/' + quote(endpoint) + '/ls',
+                         headers=self.authentication, params=params)
+            try:
+                file_response = r.json()
+                if 'is a file' in file_response['message']:
+                    os.remove(test)
+                    return endpoint
+            except KeyError:
+                pass
 
         os.remove(test)
         return None
@@ -256,11 +256,11 @@ def transfer_file(self, src_endpoint, src_path, dst_enpoint, dst_path=None):
             transfer_submission = json.load(json_file)
 
         transfer_submission["submission_id"] = submission_id["value"]
-        transfer_submission["source_endpoint"] = unicode(src_endpoint)
-        transfer_submission["destination_endpoint"] = unicode(dst_enpoint)
-        transfer_submission["DATA"][0]["source_path"] = unicode(src_path)
-        transfer_submission["DATA"][0]["destination_path"] = unicode(dst_path)
-        transfer_submission["label"] = unicode('transfered from SPEW client')
+        transfer_submission["source_endpoint"] = str(src_endpoint)
+        transfer_submission["destination_endpoint"] = str(dst_enpoint)
+        transfer_submission["DATA"][0]["source_path"] = str(src_path)
+        transfer_submission["DATA"][0]["destination_path"] = str(dst_path)
+        transfer_submission["label"] = str('transfered from SPEW client')
 
         r = self.post(self.TRANSFER_URL + '/transfer', json=transfer_submission, headers=self.authentication)
         transfer_result = self.check_response(r)
@@ -296,9 +296,9 @@ def delete_file(self, endpoint, fpath):
             delete_submission = json.load(json_file)
 
         delete_submission["submission_id"] = submission_id["value"]
-        delete_submission["endpoint"] = unicode(endpoint)
-        delete_submission["DATA"][0]["path"] = unicode(fpath)
-        delete_submission["label"] = unicode('deleted from SPEW client')
+        delete_submission["endpoint"] = str(endpoint)
+        delete_submission["DATA"][0]["path"] = str(fpath)
+        delete_submission["label"] = str('deleted from SPEW client')
 
         r = self.post(self.TRANSFER_URL + '/delete', json=delete_submission,
                       headers=self.authentication)
@@ -386,7 +386,7 @@ def transfer_generator(self, src_endpoint, src_path, dst_enpoint, dst_path=None)
         while status['status'] == 'ACTIVE':
             time.sleep(3)
             status = self.get_task_status(r['task_id'])
-            yield float(status['bytes_transferred'])/float(size)
+            yield old_div(float(status['bytes_transferred']), float(size))
 
 
 class GLOBUSError(Exception):

client/newt.py

@@ -206,7 +206,7 @@ def _check_scratch_path(self, path, system):
         try:
             contents = self.get_dir_contents(path, system)
             for item in contents:
-                if 'user' in item.keys():
+                if 'user' in list(item.keys()):
                     if item['user'] == self.username:
                         return True
         except HTTPError:
@@ -234,7 +234,7 @@ def get_home_dir(self, system):
                          self.username)
             contents = self.get_dir_contents(root_home, system)
             for item in contents:
-                    if 'user' in item.keys():
+                if 'user' in list(item.keys()):
                         if item['user'] == self.username:
                             home_dir = root_home
                             break
@@ -437,7 +437,7 @@ def download_file_generator(self, path, system, save_path=None, chunk_size=64*10
                 if chunk:  # filter out keep-alive new chunks
                     f.write(chunk)
                     f.flush()
-                    downloaded += len(chunk)/file_size
+                    downloaded += old_div(len(chunk), file_size)
                     yield downloaded
         r.close()
 

client/sftp.py

@@ -1,3 +1,4 @@
+
 import pysftp
 
 

client/spot.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 import os
 from time import sleep
-from StringIO import StringIO
+from io import StringIO
 from PIL import Image
 import numpy as np
 from client.newt import NewtClient
@@ -73,10 +73,10 @@ def search(self, query, **kwargs):
                           'end_station': ['bl832', 'bl733']}
         generic_kwargs = ['limitnum', 'skipnum']
         for key in kwargs:
-            if key not in allowed_kwargs.keys() and key not in generic_kwargs:
+            if key not in list(allowed_kwargs.keys()) and key not in generic_kwargs:
                 raise ValueError('%s keyword not in allowed keywords %s' %
-                                 (key, list(allowed_kwargs.keys() +
-                                  generic_kwargs)))
+                                 (key, list(list(allowed_kwargs.keys()) +
+                                            generic_kwargs)))
             elif key in allowed_kwargs:
                 if kwargs[key] not in allowed_kwargs[key]:
                     raise ValueError('%s keyword value must be on of %s' %
@@ -461,7 +461,7 @@ def download_dataset_generator(self, dataset, stage, save_path=None, chunk_size=
                 if chunk:  # filter out keep-alive new chunks
                     f.write(chunk)
                     f.flush()
-                    downloaded += len(chunk)/file_size
+                    downloaded += old_div(len(chunk), file_size)
                     yield downloaded
         r.close()
 
@@ -495,7 +495,7 @@ class SPOTError(Exception):
 
 if __name__ == '__main__':
     import time
-    from StringIO import StringIO
+    from io import StringIO
     from PIL import Image
     from matplotlib.pyplot import imshow, show, figure
     s = SpotClient()
@@ -505,11 +505,11 @@ class SPOTError(Exception):
     # print 'Time: ', time.time() - t
     t = time.time()
     arr = s.get_image_as('20160630_054009_prefire_3_0amp_scan7', 'raw', ext='tif', index=0)
-    print arr.shape
+    print(arr.shape)
     imshow(arr)
     show()
     # for i in range(3):
     #     figure(i)
     #     imshow(arr[:, :, i])
     # show()
-    print 'Time: ', time.time() - t
+    print('Time: ', time.time() - t)

client/ssh.py

@@ -1,3 +1,4 @@
+from __future__ import unicode_literals
 import paramiko