Merge pull request #38 from aperture-data/release-0.1.1

luisremis · web-flow · commit 31f7492e21a7 · 2022-02-03T12:17:17.000-08:00
Release 0.1.1
diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
@@ -40,9 +40,13 @@ jobs:
           password: ${{ secrets.DOCKER_PASS }}
 
       - name: Run Tests
+        env:
+          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+          AWS_DEFAULT_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
         run: |
           pip3 install .
-          pip3 install ipython torch torchvision
+          pip3 install ipython torch torchvision boto3
           cd test
           bash run_test.sh
 
diff --git a/aperturedb/CSVParser.py b/aperturedb/CSVParser.py
@@ -47,7 +47,9 @@ def parse_properties(self, df, idx):
                     prop = key[len("date:"):] # remove prefix
                     properties[prop] = {"_date": self.df.loc[idx, key]}
                 else:
-                    properties[key] = self.df.loc[idx, key]
+                    value = self.df.loc[idx, key]
+                    if value == value:  # skips nan values
+                        properties[key] = value
 
         return properties
 
diff --git a/aperturedb/ImageLoader.py b/aperturedb/ImageLoader.py
@@ -1,5 +1,8 @@
 import math
 import time
+import requests
+import os
+import boto3
 from threading import Thread
 
 import numpy as np
@@ -9,10 +12,12 @@
 from aperturedb import ParallelLoader
 from aperturedb import CSVParser
 
-HEADER_PATH = "filename"
-PROPERTIES  = "properties"
-CONSTRAINTS = "constraints"
-IMG_FORMAT  = "format"
+HEADER_PATH   = "filename"
+HEADER_URL    = "url"
+HEADER_S3_URL = "s3_url"
+PROPERTIES    = "properties"
+CONSTRAINTS   = "constraints"
+IMG_FORMAT    = "format"
 
 class ImageGeneratorCSV(CSVParser.CSVParser):
 
@@ -21,6 +26,11 @@ class ImageGeneratorCSV(CSVParser.CSVParser):
         Expects a csv file with the following columns (format optional):
 
             filename,PROP_NAME_1, ... PROP_NAME_N,constraint_PROP1,format
+            OR
+            url,PROP_NAME_1, ... PROP_NAME_N,constraint_PROP1,format
+            OR
+            s3_url,PROP_NAME_1, ... PROP_NAME_N,constraint_PROP1,format
+            ...
 
         Example csv file:
         filename,id,label,constaint_id,format
@@ -29,7 +39,7 @@ class ImageGeneratorCSV(CSVParser.CSVParser):
         ...
     '''
 
-    def __init__(self, filename, check_image=True):
+    def __init__(self, filename, check_image=True, n_download_retries=3):
 
         super().__init__(filename)
 
@@ -40,13 +50,31 @@ def __init__(self, filename, check_image=True):
         self.props_keys       = [x for x in self.props_keys if x != IMG_FORMAT]
         self.constraints_keys = [x for x in self.header[1:] if x.startswith(CSVParser.CONTRAINTS_PREFIX) ]
 
+        self.source_type      = self.header[0]
+        if self.source_type not in [ HEADER_PATH, HEADER_URL, HEADER_S3_URL ]:
+            print("Source not recognized: " + self.source_type)
+            raise Exception("Error loading image: " + filename )
+
+        self.n_download_retries = n_download_retries
+
     # TODO: we can add support for slicing here.
     def __getitem__(self, idx):
 
-        filename   = self.df.loc[idx, HEADER_PATH]
         data = {}
 
-        img_ok, img = self.load_image(filename)
+        img_ok = True
+        img = None
+
+        if self.source_type == HEADER_PATH:
+            image_path   = self.df.loc[idx, HEADER_PATH]
+            img_ok, img  = self.load_image(image_path)
+        elif self.source_type == HEADER_URL:
+            image_path   = self.df.loc[idx, HEADER_URL]
+            img_ok, img  = self.load_url(image_path)
+        elif self.source_type == HEADER_S3_URL:
+            image_path   = self.df.loc[idx, HEADER_S3_URL]
+            img_ok, img  = self.load_s3_url(image_path)
+
         if not img_ok:
             print("Error loading image: " + filename )
             raise Exception("Error loading image: " + filename )
@@ -67,12 +95,12 @@ def __getitem__(self, idx):
         return data
 
     def load_image(self, filename):
-
         if self.check_image:
             try:
                 a = cv2.imread(filename)
                 if a.size <= 0:
                     print("IMAGE SIZE ERROR:", filename)
+                    return false, None
             except:
                 print("IMAGE ERROR:", filename)
 
@@ -83,14 +111,73 @@ def load_image(self, filename):
             return True, buff
         except:
             print("IMAGE ERROR:", filename)
+        return False, None
+
+    def check_image_buffer(self, img):
+        try:
+            decoded_img = cv2.imdecode(img, cv2.IMREAD_COLOR)
+
+            # Check image is correct
+            decoded_img = decoded_img if decoded_img is not None else img
+
+            return True
+        except:
+            return False
+
+    def load_url(self, url):
+        retries = 0
+        while True:
+            imgdata = requests.get(url)
+            if imgdata.ok:
+                imgbuffer = np.frombuffer(imgdata.content, dtype='uint8')
+                if self.check_image and not self.check_image_buffer(imgbuffer):
+                    print("IMAGE ERROR: ", url)
+                    return False, None
+
+                return imgdata.ok, imgdata.content
+            else:
+                if retries >= self.n_download_retries:
+                    break
+                print("WARNING: Retrying object:", url)
+                retries += 1
+                time.sleep(2)
+
+        return False, None
+
+    def load_s3_url(self, s3_url):
+        retries = 0
+
+        # The connections by boto3 cause ResourceWarning. Known
+        # issue: https://github.com/boto/boto3/issues/454
+        s3 = boto3.client('s3')
+
+        while True:
+            try:
+                bucket_name = s3_url.split("/")[2]
+                object_name = s3_url.split("s3://" + bucket_name + "/")[-1]
+                s3_response_object = s3.get_object(Bucket=bucket_name, Key=object_name)
+                img = s3_response_object['Body'].read()
+                imgbuffer = np.frombuffer(img, dtype='uint8')
+                if self.check_image and not self.check_image_buffer(imgbuffer):
+                    print("IMAGE ERROR: ", s3_url)
+                    return False, None
+
+                return True, img
+            except:
+                if retries >= self.n_download_retries:
+                    break
+                print("WARNING: Retrying object:", s3_url)
+                retries += 1
+                time.sleep(2)
 
+        print("S3 ERROR:", s3_url)
         return False, None
 
     def validate(self):
 
         self.header = list(self.df.columns.values)
 
-        if self.header[0] != HEADER_PATH:
+        if self.header[0] not in [ HEADER_PATH, HEADER_URL, HEADER_S3_URL ]:
             raise Exception("Error with CSV file field: filename. Must be first field")
 
 class ImageLoader(ParallelLoader.ParallelLoader):
diff --git a/aperturedb/ParallelLoader.py b/aperturedb/ParallelLoader.py
@@ -85,7 +85,11 @@ def ingest(self, generator, batchsize=1, numthreads=1, stats=False):
 
         start_time = time.time()
 
-        elements_per_thread = math.ceil(self.total_elements / self.numthreads)
+        if self.total_elements < batchsize:
+            elements_per_thread = self.total_elements
+            self.numthreads = 1
+        else:
+            elements_per_thread = math.ceil(self.total_elements / self.numthreads)
 
         thread_arr = []
         for i in range(self.numthreads):
diff --git a/aperturedb/PyTorchDataset.py b/aperturedb/PyTorchDataset.py
@@ -10,6 +10,8 @@
 from torch.utils import data
 from torchvision import transforms
 
+DEFAULT_BATCH_SIZE = 50
+
 class ApertureDBDatasetConstraints(data.Dataset):
 
     # initialise function of class
@@ -39,16 +41,19 @@ class ApertureDBDataset(data.Dataset):
     # initialise function of class
     def __init__(self, db, query, label_prop=None):
 
-        self.db = db
+        self.db = db.create_new_connection()
         self.query = query
         self.find_image_idx = None
         self.total_elements = 0
-        self.batch_size     = 100
+        self.batch_size     = DEFAULT_BATCH_SIZE
         self.batch_images   = []
         self.batch_start    = 0
         self.batch_end      = 0
         self.label_prop     = label_prop
 
+        self.prev_requested   = -1
+        self.sequence_counter = DEFAULT_BATCH_SIZE
+
         for i in range(len(query)):
 
             name = list(query[i].keys())[0]
@@ -76,6 +81,18 @@ def __init__(self, db, query, label_prop=None):
 
     def __getitem__(self, index):
 
+        if index == self.prev_requested + 1:
+            self.sequence_counter += 1
+        else:
+            self.sequence_counter = 0
+
+        if self.sequence_counter >= DEFAULT_BATCH_SIZE:
+            self.batch_size = DEFAULT_BATCH_SIZE
+        else:
+            self.batch_size = 1
+
+        self.prev_requested = index
+
         if index >= self.total_elements:
             raise StopIteration
 
@@ -116,7 +133,21 @@ def get_batch(self, index):
         query[self.find_image_idx]["FindImage"]["batch"] = qbatch
 
         try:
-            r,b = self.db.query(query)
+
+            # This is to handle potential issues with
+            # disconnection/timeout and SSL context on multiprocessing
+            connection_ok = False
+            try:
+                r,b = self.db.query(query)
+                connection_ok = True
+            except:
+                # Connection failed, we retry just once to re-connect
+                self.db = self.db.create_new_connection()
+
+            if not connection_ok:
+                # Connection failed, we have reconnected, we try again.
+                r,b = self.db.query(query)
+
             if len(b) == 0:
                 print("index:", index)
                 raise Exception("No results returned from ApertureDB")
diff --git a/setup.py b/setup.py
@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="aperturedb",
-    version="0.1.0",
+    version="0.1.1",
     description="ApertureDB Client Module",
     install_requires=['vdms', 'scikit-image', 'image',
                       'opencv-python', 'numpy', 'matplotlib', 'pandas'],
diff --git a/test/aperturedb/config.json b/test/aperturedb/config.json
@@ -11,9 +11,6 @@
 
     "pmgd_disk_sync_option": "never_sync",
 
-    // Serialize graph access
-    "serialize_graph_access": true,
-
     "create_parameters": {
         "pmgd_num_allocators": 32,
         "pmgd_journal_size": 1024
diff --git a/test/dbinfo.py b/test/dbinfo.py
@@ -1,4 +1,6 @@
 # This file containts information on to access the server
 
-DB_HOST="localhost"
-DB_PORT=55555
+DB_HOST     = "localhost"
+DB_PORT     = 55555
+DB_USER     = "admin"
+DB_PASSWORD = "admin"
diff --git a/test/generateInput.py b/test/generateInput.py
@@ -99,6 +99,52 @@ def generate_images_csv(multiplier):
 
     return df
 
+def generate_http_images_csv(ip_file_csv):
+
+    images    = pd.read_csv(ip_file_csv, sep=",", header=None)
+
+    ids      = [int(1000000000* random.random()) for i in range(len(images))]
+    age      = [int(100* random.random())       for i in range(len(images))]
+    height   = [float(200* random.random())     for i in range(len(images))]
+    license  = [x for x in range(len(images))]
+
+    df = pd.DataFrame()
+    df['url']      = images
+    df["urlid"]    = ids
+    df['license']  = license
+    df["age"]      = age
+    df["height"]   = height
+    df["constraint_urlid"] = ids
+
+    df = df.sort_values("urlid")
+
+    df.to_csv("input/http_images.adb.csv", index=False)
+
+    return df
+
+def generate_s3_images_csv(ip_file_csv):
+
+    images    = pd.read_csv(ip_file_csv, sep=",", header=None)
+
+    ids      = [int(1000000000* random.random()) for i in range(len(images))]
+    age      = [int(100* random.random())       for i in range(len(images))]
+    height   = [float(200* random.random())     for i in range(len(images))]
+    license  = [x for x in range(len(images))]
+
+    df = pd.DataFrame()
+    df['s3_url']   = images
+    df["id"]       = ids
+    df['license']  = license
+    df["age"]      = age
+    df["height"]   = height
+    df["constraint_id"] = ids
+
+    df = df.sort_values("id")
+
+    df.to_csv("input/s3_images.adb.csv", index=False)
+
+    return df
+
 def generate_connections_csv(persons, images):
 
     connections  = list(product(images["id"][::100], persons["id"][::100]))
@@ -183,6 +229,8 @@ def main(params):
     persons = generate_person_csv(params.multiplier)
     blobs   = generate_blobs_csv()
     images  = generate_images_csv(int(params.multiplier/2))
+    s3_imgs = generate_http_images_csv("input/sample_http_urls.csv")
+    s3_imgs = generate_s3_images_csv("input/sample_s3_urls.csv")
     connect = generate_connections_csv(persons, images)
     bboxes  = generate_bboxes_csv(images)
 
diff --git a/test/input/sample_http_urls.csv b/test/input/sample_http_urls.csv
@@ -0,0 +1,10 @@
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/1002318269_97db6e0975.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/10201275523_3e6ea67c7f.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/2297552664_1ee0e8855d.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/4140939180_07aeded917.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/4436463882_b96a3d9df9.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/4572998878_658b45226f.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/6985418911_df7747990d.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/7289030198_1f1ba44113.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/9329902958_0bc80ce58a.jpg
+https://aperturedata-public.s3.us-west-2.amazonaws.com/sample_images/9506922316_c19019e38f.jpg
diff --git a/test/input/sample_s3_urls.csv b/test/input/sample_s3_urls.csv
@@ -0,0 +1,10 @@
+s3://aperturedata-public/sample_images/4436463882_b96a3d9df9.jpg
+s3://aperturedata-public/sample_images/9329902958_0bc80ce58a.jpg
+s3://aperturedata-public/sample_images/2297552664_1ee0e8855d.jpg
+s3://aperturedata-public/sample_images/4140939180_07aeded917.jpg
+s3://aperturedata-public/sample_images/6985418911_df7747990d.jpg
+s3://aperturedata-public/sample_images/10201275523_3e6ea67c7f.jpg
+s3://aperturedata-public/sample_images/7289030198_1f1ba44113.jpg
+s3://aperturedata-public/sample_images/1002318269_97db6e0975.jpg
+s3://aperturedata-public/sample_images/9506922316_c19019e38f.jpg
+s3://aperturedata-public/sample_images/4572998878_658b45226f.jpg
diff --git a/test/run_test.sh b/test/run_test.sh
diff --git a/test/test_Loaders.py b/test/test_Loaders.py
diff --git a/test/test_torch_connector.py b/test/test_torch_connector.py
