docs updated and minor bugs fixed

Author: stephenoffer

.idea/workspace.xml

@@ -17,8 +17,8 @@ def main():
                 print("Removing: {}".format(path))
                 os.system('rm -rf {}'.format(path))
 
-    print("Removing pype.egg-info")
-    os.system('rm -rf {}'.format(os.path.join(dirpath,'pype.egg-info')))
+    print("Removing dKeras.egg-info")
+    os.system('rm -rf {}'.format(os.path.join(dirpath,'dKeras.egg-info')))
 
 
 if __name__ == "__main__":

cleanup.py

@@ -85,39 +85,45 @@ def main():
         print("Time elapsed: {}\nFPS: {}".format(elapsed, n_data / elapsed))
 
     elif test_type == 1:
-        if use_search:
-            results = {}
-            best_time = np.inf
-            best_n_workers = -1
+        if model_name != 'all':
+            models = [model_names[model_name]]
+        else:
+            models = [model_names[m] for m in model_names.keys()]
 
-            for n in search_pool:
-                model = dKeras(model_names[model_name], wait_for_workers=True,
-                               n_workers=n)
-                print("Workers are ready")
+        for m in models:
+            if use_search:
+                results = {}
+                best_time = np.inf
+                best_n_workers = -1
 
-                start_time = time.time()
-                preds = model.predict(test_data)
-                elapsed = time.time() - start_time
+                for n in search_pool:
+                    model = dKeras(m, wait_for_workers=True,
+                                   n_workers=n)
+                    print("Workers are ready")
 
-                time.sleep(3)
+                    start_time = time.time()
+                    preds = model.predict(test_data)
+                    elapsed = time.time() - start_time
 
-                if elapsed < best_time:
-                    best_time = elapsed
-                    best_n_workers = n
+                    time.sleep(3)
 
-                results[str(n)] = elapsed
-                model.close()
+                    if elapsed < best_time:
+                        best_time = elapsed
+                        best_n_workers = n
 
-            print('{}\nN\tElapsed Time'.format('=' * 80))
+                    results[str(n)] = elapsed
+                    model.close()
 
-            for k in results.keys():
-                print("{}\t{}".format(k, results[k]))
+                print('{}\nN\tElapsed Time'.format('=' * 80))
 
-            print("{}\nTests completed:\n\tBest N workers: {}\t FPS: {}".format(
-                '=' * 80, best_n_workers, n_data / best_time))
-        else:
-            model = dKeras(model_names[model_name], wait_for_workers=True,
-                           n_workers=n_workers)
+                for k in results.keys():
+                    print("{}\t{}".format(k, results[k]))
+
+                print("{}\nTests completed:\n\tBest N workers: {}\t FPS: {}".format(
+                    '=' * 80, best_n_workers, n_data / best_time))
+            else:
+                model = dKeras(model_names[model_name], wait_for_workers=True,
+                               n_workers=n_workers)
 
             start_time = time.time()
             preds = model.predict(test_data)

dKeras.egg-info/PKG-INFO

@@ -9,7 +9,7 @@
 # Remote Worker Functions Configuration
 # -----------------------------------------------------------------------------
 DEFAULT_N_WORKERS = 4
-WORKER_WAIT_TIME = 1e-4
+WORKER_WAIT_TIME = 5e-3
 WORKER_INFERENCE_BATCH_SIZE = 250
 
 # -----------------------------------------------------------------------------

dKeras.egg-info/SOURCES.txt

@@ -18,26 +18,78 @@
 
 
 class dKeras(object):
+    """Distributed Keras Model Wrapper.
+
+        It will automatically set up remote
+        workers and data servers for data parallelism algorithms. Using
+        the same notation as a regular Keras model, it makes distributing a
+        Keras model simple.
+
+        .. code-block:: python
+
+            from tensorflow.keras.applications import ResNet50
+            from dkeras import dKeras
+
+            model = dKeras(ResNet50)
+            preds = model.predict(data)
+
+        Arguments:
+            model: Un-initialized Keras model
+            verbose: Verbose setting boolean variable. Default is False.
+            weights: Weights arg for prebuilt models, example: ResNet50(
+                weights='imagenet'). Default is None.
+            n_workers: Integer number of worker processes. If left None,
+                then it will automatically find the an estimate of the optimal
+                number of workers. Default is None.
+            init_ray: Boolean arg for whether to initialize Ray within
+                the model initialization. Default is False.
+            rm_existing_ray: Boolean arg for whether to remove any
+                existing Ray clusters. Default is True.
+            rm_local_model: Boolean arg for whether to remove the local
+                copy of the Keras model for memory conservation. Default is
+                False.
+            wait_for_workers: Boolean arg for whether to wait for all of
+                the worker processes to initialize and connect to the data
+                server.
+            redis_address: In the case of initializing Ray inside of
+                model initialization, the redis address is required for
+                connecting to existing Ray clusters.
+            n_cpus_per_worker: The integer number of CPUs per worker
+                processes. If left None, it will allocate automatically. The
+                default is None.
+            n_gpus_per_worker: The integer or float number of GPUs per
+                worker processes. If left None, it will allocate
+                automatically. The default is None.
+            n_cpus_per_server: The integer number of CPUs per data
+                server. If left None, it will allocate automatically. The
+                default is None.
+            """
 
     def __init__(self,
                  model,
-                 weights=None,
-                 n_workers=None,
-                 init_ray=True,
-                 rm_existing_ray=False,
-                 rm_local_model=True,
-                 wait_for_workers=False,
-                 redis_address=None):
-        """
-
-        :param model:
-        :param n_workers:
-        """
+                 verbose: bool = True,
+                 weights: list = None,
+                 n_workers: int = None,
+                 init_ray: bool = True,
+                 rm_existing_ray: bool = False,
+                 rm_local_model: bool = True,
+                 wait_for_workers: bool = False,
+                 redis_address: str = None,
+                 n_cpus_per_worker: int = None,
+                 n_gpus_per_worker: int = None,
+                 n_cpus_per_server: int = None):
+
+        config.N_CPUS_PER_SERVER = n_cpus_per_server
+        config.N_CPUS_PER_WORKER = n_cpus_per_worker
+        config.N_CPUS_PER_SERVER = n_gpus_per_worker
+        self.verbose = verbose
         if init_ray:
             if ray.is_initialized():
                 if rm_existing_ray:
                     ray.shutdown()
                     ray.init()
+                else:
+                    ray.init(redis_address=redis_address)
             else:
                 ray.init()
 
@@ -91,7 +143,7 @@ def predict(self, data, distributed=True, stop_ray=False):
             n_data = len(data)
             if n_data % self.n_workers > 0:
                 self.data_server.set_batch_size.remote(
-                    int(n_data / self.n_workers)+1)
+                    int(n_data / self.n_workers) + 1)
             else:
                 self.data_server.set_batch_size.remote(
                     int(n_data / self.n_workers))