Merge pull request #1036 from adamtheturtle/simple-query

Add the initial query interface

Author: adamtheturtle

requirements.txt

@@ -1,2 +1,3 @@
 requests==2.22.0
 timeout-decorator==0.4.1
+urllib3==1.25.3

src/vws/__init__.py

@@ -3,9 +3,11 @@
 """
 
 from ._version import get_versions
+from .query import CloudRecoService
 from .vws import VWS
 
 __all__ = [
+    'CloudRecoService',
     'VWS',
 ]
 

src/vws/query.py

@@ -0,0 +1,81 @@
+"""
+Tools for interacting with the Vuforia Cloud Recognition Web APIs.
+"""
+
+import io
+from typing import Any, Dict, List
+from urllib.parse import urljoin
+
+import requests
+from urllib3.filepost import encode_multipart_formdata
+
+from ._authorization import authorization_header, rfc_1123_date
+
+
+class CloudRecoService:
+    """
+    An interface to the Vuforia Cloud Recognition Web APIs.
+    """
+
+    def __init__(
+        self,
+        client_access_key: str,
+        client_secret_key: str,
+    ) -> None:
+        """
+        Args:
+            client_access_key: A VWS client access key.
+            client_secret_key: A VWS client secret key.
+        """
+        self._client_access_key = client_access_key.encode()
+        self._client_secret_key = client_secret_key.encode()
+
+    def query(self, image: io.BytesIO) -> List[Dict[str, Any]]:
+        """
+        Use the Vuforia Web Query API to make an Image Recognition Query.
+
+        See
+        https://library.vuforia.com/articles/Solution/How-To-Perform-an-Image-Recognition-Query
+        for parameter details.
+
+        Args:
+            image: The image to make a query against.
+
+        Returns:
+            An ordered list of target details of matching targets.
+        """
+        image_content = image.getvalue()
+        body = {
+            'image': ('image.jpeg', image_content, 'image/jpeg'),
+        }
+        date = rfc_1123_date()
+        request_path = '/v1/query'
+        content, content_type_header = encode_multipart_formdata(body)
+        method = 'POST'
+
+        authorization_string = authorization_header(
+            access_key=self._client_access_key,
+            secret_key=self._client_secret_key,
+            method=method,
+            content=content,
+            # Note that this is not the actual Content-Type header value sent.
+            content_type='multipart/form-data',
+            date=date,
+            request_path=request_path,
+        )
+
+        headers = {
+            'Authorization': authorization_string,
+            'Date': date,
+            'Content-Type': content_type_header,
+        }
+
+        base_vwq_url = 'https://cloudreco.vuforia.com'
+        response = requests.request(
+            method=method,
+            url=urljoin(base=base_vwq_url, url=request_path),
+            headers=headers,
+            data=content,
+        )
+
+        return list(response.json()['results'])

tests/conftest.py

@@ -8,7 +8,7 @@
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
 
-from vws import VWS
+from vws import VWS, CloudRecoService
 
 pytest_plugins = [  # pylint: disable=invalid-name
     'tests.fixtures.images',
@@ -29,9 +29,21 @@ def _mock_database() -> Iterator[VuforiaDatabase]:
 @pytest.fixture()
 def vws_client(_mock_database: VuforiaDatabase) -> Iterator[VWS]:
     """
-    Yield a VWS client which connects to a mock.
+    Yield a VWS client which connects to a mock database.
     """
     yield VWS(
         server_access_key=_mock_database.server_access_key,
         server_secret_key=_mock_database.server_secret_key,
     )
+
+
+@pytest.fixture()
+def cloud_reco_client(_mock_database: VuforiaDatabase,
+                      ) -> Iterator[CloudRecoService]:
+    """
+    Yield a ``CloudRecoService`` client which connects to a mock database.
+    """
+    yield CloudRecoService(
+        client_access_key=_mock_database.client_access_key,
+        client_secret_key=_mock_database.client_secret_key,
+    )

tests/test_query.py

@@ -0,0 +1,73 @@
+"""
+Tests for the ``CloudRecoService`` querying functionality.
+"""
+
+import io
+import uuid
+
+from vws import VWS, CloudRecoService
+
+
+class TestQuery:
+    """
+    Tests for making image queries.
+    """
+
+    def test_no_matches(
+        self,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        An empty list is returned if there are no matches.
+        """
+        result = cloud_reco_client.query(image=high_quality_image)
+        assert result == []
+
+    def test_match(
+        self,
+        vws_client: VWS,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        Details of matching targets are returned.
+        """
+        target_id = vws_client.add_target(
+            name='x',
+            width=1,
+            image=high_quality_image,
+        )
+        vws_client.wait_for_target_processed(target_id=target_id)
+        [matching_target] = cloud_reco_client.query(image=high_quality_image)
+        assert matching_target['target_id'] == target_id
+
+
+class TestMaxNumResults:
+    """
+    Tests for the ``max_num_results`` parameter of ``query``.
+    """
+
+    def test_default(
+        self,
+        vws_client: VWS,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        By default the maximum number of results is 1.
+        """
+        target_id = vws_client.add_target(
+            name=uuid.uuid4().hex,
+            width=1,
+            image=high_quality_image,
+        )
+        target_id_2 = vws_client.add_target(
+            name=uuid.uuid4().hex,
+            width=1,
+            image=high_quality_image,
+        )
+        vws_client.wait_for_target_processed(target_id=target_id)
+        vws_client.wait_for_target_processed(target_id=target_id_2)
+        matches = cloud_reco_client.query(image=high_quality_image)
+        assert len(matches) == 1