Merge pull request #564 from aperture-data/release-0.4.43

Release 0.4.43

Author: gsaluja9

aperturedb/ImageDownloader.py

@@ -144,16 +144,16 @@ def download_image(self, url, filename):
 
         self.times_arr.append(time.time() - start)
 
-    def worker(self, thid, generator, start, end):
+    def worker(self, thid, generator, start, end, run_event):
 
         for i in range(start, end):
 
             url, filename = generator[i]
 
             self.download_image(url, filename)
 
-            if thid == 0 and self.stats:
-                self.pb.update((i - start) / (end - start))
+            if self.stats:
+                self.pb.update(1)
 
     def print_stats(self):
 

aperturedb/ParallelQuery.py

@@ -214,7 +214,7 @@ def filter_per_group(group):
         self.times_arr.append(query_time)
         self.actual_stats.append(worker_stats)
 
-    def worker(self, thid: int, generator, start: int, end: int):
+    def worker(self, thid: int, generator, start: int, end: int, run_event) -> None:
         # A new connection will be created for each thread
         client = self.client.clone()
 
@@ -223,9 +223,11 @@ def worker(self, thid: int, generator, start: int, end: int):
         if (end - start) % self.batchsize > 0:
             total_batches += 1
 
-        logger.info(f"Worker {thid} executing {total_batches} batches")
+        logger.info(
+            f"Worker {thid} executing {total_batches} batches, {self.stats=}")
         for i in range(total_batches):
-
+            if not run_event.is_set():
+                break
             batch_start = start + i * self.batchsize
             batch_end = min(batch_start + self.batchsize, end)
 
@@ -238,8 +240,9 @@ def worker(self, thid: int, generator, start: int, end: int):
                     f"Worker {thid} failed to execute batch {i}: [{batch_start},{batch_end}]")
                 self.error_counter += 1
 
-            if thid == 0 and self.stats:
-                self.pb.update((i + 1) / total_batches)
+            if self.stats:
+                self.pb.update(self.batchsize)
+        logger.info(f"Worker {thid} executed {total_batches} batches")
 
     def get_objects_existed(self) -> int:
         return sum([stat["objects_existed"]
@@ -307,7 +310,7 @@ def query(self, generator, batchsize: int = 1, numthreads: int = 4, stats: bool
             logger.info(
                 f"Commands per query = {self.commands_per_query}, Blobs per query = {self.blobs_per_query}"
             )
-            self.run(generator, batchsize, numthreads, stats)
+            self.batched_run(generator, batchsize, numthreads, stats)
 
     def print_stats(self) -> None:
 

aperturedb/Parallelizer.py

@@ -1,9 +1,9 @@
 import math
 import time
-from threading import Thread
-
+import threading
 
-from aperturedb import ProgressBar
+from threading import Thread
+from tqdm import tqdm as tqdm
 
 
 class Parallelizer:
@@ -28,10 +28,7 @@ class Parallelizer:
     ```
     """
 
-    def __init__(self, progress_to_file=""):
-
-        self.pb_file = progress_to_file
-
+    def __init__(self):
         self._reset()
 
     def _reset(self, batchsize: int = 1, numthreads: int = 1):
@@ -46,17 +43,13 @@ def _reset(self, batchsize: int = 1, numthreads: int = 1):
         self.error_counter = 0
         self.actual_stats = []
 
-        if self.pb_file:
-            self.pb = ProgressBar.ProgressBar(self.pb_file)
-        else:
-            self.pb = ProgressBar.ProgressBar()
-
     def get_times(self):
 
         return self.times_arr
 
-    def run(self, generator, batchsize: int, numthreads: int, stats: bool):
-
+    def batched_run(self, generator, batchsize: int, numthreads: int, stats: bool):
+        run_event = threading.Event()
+        run_event.set()
         self._reset(batchsize, numthreads)
         self.stats = stats
         self.generator = generator
@@ -65,7 +58,8 @@ def run(self, generator, batchsize: int, numthreads: int, stats: bool):
             self.total_actions = generator.sample_count
         else:
             self.total_actions = len(generator)
-
+        self.pb = tqdm(total=self.total_actions, desc="Progress",
+                       unit="batches", unit_scale=True, dynamic_ncols=True)
         start_time = time.time()
 
         if self.total_actions < batchsize:
@@ -82,15 +76,22 @@ def run(self, generator, batchsize: int, numthreads: int, stats: bool):
                           self.total_actions)
 
             thread_add = Thread(target=self.worker,
-                                args=(i, generator, idx_start, idx_end))
+                                args=(i, generator, idx_start, idx_end, run_event))
             thread_arr.append(thread_add)
 
         a = [th.start() for th in thread_arr]
-        a = [th.join() for th in thread_arr]
+        try:
+            while run_event.is_set() and any([th.is_alive() for th in thread_arr]):
+                time.sleep(1)
+        except KeyboardInterrupt:
+            print("Interrupted ... Shutting down workers")
+        finally:
+            run_event.clear()
+            a = [th.join() for th in thread_arr]
 
         # Update progress bar to completion
         if self.stats:
-            self.pb.update(1)
+            self.pb.close()
 
         self.total_actions_time = time.time() - start_time
 

aperturedb/ProgressBar.py

@@ -4,7 +4,7 @@
 """
 from aperturedb.Query import QueryBuilder
 from aperturedb.CommonLibrary import execute_query
-from aperturedb import ProgressBar
+from tqdm import tqdm
 from aperturedb.Connector import Connector
 import logging
 import json
@@ -595,38 +595,42 @@ def _remove_objects(self, type, class_name, batch_size):
 
         total = count
 
-        pb = ProgressBar.ProgressBar()
+        pb = tqdm(total=total, desc=f"Removing {type} {class_name}",
+                  unit="elements", unit_scale=True, dynamic_ncols=True)
 
         find = "Find" + cmd
         dele = "Delete" + cmd
-        while count > 0:
-
-            q = [{
-                find: {
-                    "_ref": 1,
-                    "with_class": class_name,
-                    "results": {
-                        "limit": batch_size
+        try:
+            while count > 0:
+
+                q = [{
+                    find: {
+                        "_ref": 1,
+                        "with_class": class_name,
+                        "results": {
+                            "limit": batch_size
+                        }
                     }
-                }
-            }, {
-                dele: {
-                    "ref": 1
-                }
-            }]
+                }, {
+                    dele: {
+                        "ref": 1
+                    }
+                }]
 
-            try:
-                self.execute(q)
-            except:
-                return False
+                try:
+                    self.execute(q)
+                except:
+                    return False
 
-            self.print(
-                f"Delete transaction of {batch_size} elements took: {self.client.get_last_query_time()}")
+                self.print(
+                    f"Delete transaction of {batch_size} elements took: {self.client.get_last_query_time()}")
 
-            count -= batch_size
+                count -= batch_size
 
-            if self.verbose:
-                pb.update(abs((count - total) / total))
+                if self.verbose:
+                    pb.update(batch_size)
+        finally:
+            pb.close()
 
         return True
 

aperturedb/Utils.py

@@ -16,7 +16,7 @@
 
 logger = logging.getLogger(__name__)
 
-__version__ = "0.4.42"
+__version__ = "0.4.43"
 
 # set log level
 formatter = logging.Formatter(

aperturedb/__init__.py

@@ -36,7 +36,7 @@ def execute(command: CommandTypes,
 
     utils = Utils(create_connector())
     available_commands = {
-        CommandTypes.STATUS: lambda: print(utils),
+        CommandTypes.STATUS: lambda: print(utils.status()),
         CommandTypes.SUMMARY: utils.summary,
         CommandTypes.REMOVE_ALL: lambda: confirm(
             CommandTypes.REMOVE_ALL, force) and utils.remove_all_objects(),

aperturedb/cli/utilities.py

@@ -153,8 +153,10 @@ build_notebook_image(){
     mkdir -p docker/notebook/aperturedata
     cp -r aperturedb pyproject.toml LICENSE README.md docker/notebook/aperturedata
     LATEST_IMAGE=${DOCKER_REPOSITORY}/aperturedb-notebook${IMAGE_EXTENSION_LATEST}
+    CPU_IMAGE=${DOCKER_REPOSITORY}/aperturedb-notebook:cpu
     echo "Building image ${NOTEBOOK_IMAGE}"
     docker build -t ${NOTEBOOK_IMAGE} -t ${LATEST_IMAGE} -f docker/notebook/Dockerfile .
+    docker build -t ${CPU_IMAGE} -f docker/notebook/Dockerfile.cpu .
     if [ "${NO_PUSH}" != "true" ]
     then
         docker push --all-tags ${DOCKER_REPOSITORY}/aperturedb-notebook