Skip to main content

Native API Queries

Create a connection to ApertureDB​

from aperturedb import Connector
from aperturedb import NotebookHelpers as nh

# ApertureDB Server Info for establishing connection
db_host  = "aperturedb.local"                       # assuming local installation as provided
user     = "admin"                                  # requires authentication
password = input("Please insert your password: ")   # use the password provided for the instance or "admin" by default

db = Connector.Connector(db_host, user=user, password=password)
print("Connected to ApertureDB server...")

Find a specific video with a simple metadata query​

query = [{
    "FindVideo": {
        "constraints": {
            "guid": ["==", "f592d7ddd6a14b16a8e5ab3f02789fd"]
        },
        "blobs": True,
        "results": {
            "all_properties": True
        }
    }
}]

response, blobs = db.query(query)

print(db.get_last_response_str())
nh.display_video_mp4(blobs[0])

Count videos with given metadata constraint to understand distribution​

query = [
  {
    "FindVideo": {
      "constraints": {
        "date_captured": [">=", {"_date": "2022-08-20T14:29:22+00:00"}],
        "sequence": [">", 20]
      },
      "blobs": False, 
      "results": {
        "count":True
      }
    }
  }
]


response, blobs = db.query(query)

print("Count of videos meeting criteria: " + str(response[0]["FindVideo"]["count"]))

Find videos captured by a given camera in a specified store​

query = [{
    "FindEntity": {                     # Find a specific store from application metadata
        "_ref": 1,
        "with_class": "Store",
        "constraints": {
            "chain": ["==", "Aramark"],
            "location": ["==", "Mile high"]
        },
        "results": {
            "list": ["chain", "location", "store_id"]
        }
    }
}, {
    "FindEntity": {                     # Find all the Nest Cameras in the store found by previous query
        "_ref": 2,
        "with_class": "Camera",
        "is_connected_to" : {
            "ref": 1
        },
        "constraints": {
            "brand": ["==", "Google"],
            "model": ["==", "Nest"]
        },
        "results": {
            "list": ["brand", "model", "resolution"]
        }
    }
}, {
    "FindVideo": {                      # Find all the videos connected to the cameras that match above
        "_ref": 3,
        "is_connected_to": {
            "ref": 2
        },
        "blobs": True,
        "results": {
            "all_properties": True
        }
    }
}]

response, blobs = db.query(query)
print(db.get_last_response_str())
num_videos = response[2]["FindVideo"]["returned"]
for count in range(num_videos):
    nh.display_video_mp4(blobs[count])

Sample videos and pre-process the video clips for display​

query = [{
    "FindVideo": {
        "_ref": 1,
        "constraints": {        # More complex metadata constraints
            "date_captured": [">=", {"_date": "2021-08-30T14:29:22+00:00"}],
            "sequence": [">=", 12, "<", 20]
        },
        "results": {
            "count": True
        },
        "blobs": True,
        "operations": [{        # Original videos are longer. We just want to sample and prepare them
            "type": "resize",
            "height": 300,
            "width":  300
        },
        {
            "type": "interval", # Interval Operation
            "start": 0,         # Start from the 0th frame
            "stop": 100,        # Stop at the 100th frame
            "step": 5           # Pick 1 every 5 frames.
        }]
    }
}]

response, blobs = db.query(query)

print(db.get_last_response_str())

num_videos = response[0]["FindVideo"]["returned"]
for count in range(num_videos):
    nh.display_video_mp4(blobs[count])

Extract frames from videos filmed indoors, say for labeling or classification​

query = [{
    "FindEntity": {                     # Find all the outdoor locations from application metadata
        "_ref": 1,
        "with_class": "Location",
        "constraints": {
            "name": ["in", ["house", "restaurant", "auditorium"]]
        },
        "results": {
            "all_properties": True
        }
    }
}, {
    "FindVideo": {                      # Find all the videos connected to the locations that match above
        "_ref": 2,
        "is_connected_to": {
            "ref": 1
        },
        "blobs": False,
        "results": {
            "count": True
        }
    }
}, {                                    # Save a sample of frames from videos above to create a training set
    "ExtractFrames": {
        "video_ref": 2,
        "frame_spec": [1, 100, 200],
        "as_format": "png",
        "operations": [
            {
                "type": "resize",
                "height": 224,
                "width": 224
            }
        ],
    }
}]

response, blobs = db.query(query)
print(db.get_last_response_str())
nh.display(blobs)

Extract Frames from all videos using batch API e.g. for concurrent access when training​

query = [{
    "FindVideo": {
        "_ref": 1,
        "constraints": {
            "sequence": [">", 0]
        },
        "blobs": False,
        "results": {
            "count": True,
        }
    }
}, {
    "ExtractFrames": {
        "video_ref": 1,
        "frame_spec": [0],
        "operations": [
            {
                "type": "resize",
                "width": 224,
            }
        ],
        "batch": {
            "batch_size": 10,
            "batch_id": 3                   # if you have multiple workers, each can now work on different batches
        },
    }
}]

response, blobs = db.query(query)

print(db.get_last_response_str())

nh.display(blobs)