"""Async tools for interacting with Vuforia APIs."""

import asyncio

import base64

import calendar # noqa: TC003

import json

import time

from http import HTTPMethod, HTTPStatus

from typing import Self

from beartype import BeartypeConf, beartype

from vws._async_vws_request import async_target_api_request

from vws._image_utils import ImageType as _ImageType

from vws._image_utils import get_image_data as _get_image_data

from vws._reco_counts import (

reco_counts_report_body,

reco_counts_report_path,

report_from_download_response,

)

from vws.exceptions.base_exceptions import VWSError

from vws.exceptions.custom_exceptions import (

RecoCountsReportNotReadyError,

RecoCountsReportTimeoutError,

ServerError,

TargetProcessingTimeoutError,

)

from vws.exceptions.vws_exceptions import TooManyRequestsError

from vws.reports import (

DatabaseSummaryReport,

RecoCountsReport,

RecoCountsReportRequest,

TargetStatusAndRecord,

TargetStatuses,

TargetSummaryReport,

)

from vws.response import Response # noqa: TC001

from vws.transports import AsyncHTTPXTransport, AsyncTransport

@beartype(conf=BeartypeConf(is_pep484_tower=True))

class AsyncVWS:

"""An async interface to Vuforia Web Services APIs."""

def __init__(

self,

*,

server_access_key: str,

server_secret_key: str,

base_vws_url: str = "https://vws.vuforia.com",

database_id: str | None = None,

request_timeout_seconds: float | tuple[float, float] = 30.0,

transport: AsyncTransport | None = None,

) -> None:

"""

Args:

server_access_key: A VWS server access key.

server_secret_key: A VWS server secret key.

base_vws_url: The base URL for the VWS API.

database_id: The ID of the database which the

given keys belong to. This is shown in the

target manager. It is needed only by

:meth:`request_database_reco_counts_report`.

request_timeout_seconds: The timeout for each

HTTP request. This can be a float to set both

the connect and read timeouts, or a

(connect, read) tuple.

transport: The async HTTP transport to use for

requests. Defaults to

``AsyncHTTPXTransport()``.

"""

self._server_access_key = server_access_key

self._server_secret_key = server_secret_key

self._base_vws_url = base_vws_url

self._database_id = database_id

self._request_timeout_seconds = request_timeout_seconds

self._transport = (

transport if transport is not None else AsyncHTTPXTransport()

)

async def aclose(self) -> None:

"""Close the underlying transport if it supports closing."""

await self._transport.aclose()

async def __aenter__(self) -> Self:

"""Enter the async context manager."""

return self

async def __aexit__(self, *_args: object) -> None:

"""Exit the async context manager and close the transport."""

await self.aclose()

async def make_request(

self,

*,

method: str,

data: bytes,

request_path: str,

expected_result_code: str,

content_type: str,

extra_headers: dict[str, str] | None = None,

) -> Response:

"""Make an async request to the Vuforia Target API.

Args:

method: The HTTP method which will be used in

the request.

data: The request body which will be used in the

request.

request_path: The path to the endpoint which

will be used in the request.

expected_result_code: See

"VWS API Result Codes" on

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api.

content_type: The content type of the request.

extra_headers: Additional headers to include in

the request.

Returns:

The response to the request.

Raises:

~vws.exceptions.custom_exceptions.ServerError:

There is an error with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError:

Vuforia is rate limiting access.

json.JSONDecodeError: The server did not respond

with valid JSON. This may happen if the

server address is not a valid Vuforia server.

"""

response = await async_target_api_request(

content_type=content_type,

server_access_key=self._server_access_key,

server_secret_key=self._server_secret_key,

method=method,

data=data,

request_path=request_path,

base_vws_url=self._base_vws_url,

request_timeout_seconds=self._request_timeout_seconds,

extra_headers=extra_headers or {},

transport=self._transport,

)

if (

response.status_code == HTTPStatus.TOO_MANY_REQUESTS

): # pragma: no cover

# The Vuforia API returns a 429 response with no JSON body.

raise TooManyRequestsError(response=response)

if (

response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR

): # pragma: no cover

raise ServerError(response=response)

result_code = json.loads(s=response.text)["result_code"]

if result_code == expected_result_code:

return response

raise VWSError.from_result_code(

result_code=result_code,

response=response,

)

async def add_target(

self,

*,

name: str,

width: float,

image: _ImageType,

application_metadata: str | None,

active_flag: bool,

) -> str:

"""Add a target to a Vuforia Web Services database.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#add

for parameter details.

Args:

name: The name of the target.

width: The width of the target.

image: The image of the target.

active_flag: Whether or not the target is active for query.

application_metadata: The application metadata of the target.

This must be base64 encoded, for example by using::

base64.b64encode('input_string').decode('ascii')

Returns:

The target ID of the new target.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.BadImageError: There is a problem

with the given image. For example, it must be a JPEG or PNG

file in the grayscale or RGB color space.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.MetadataTooLargeError: The given

metadata is too large. The maximum size is 1 MB of data when

Base64 encoded.

~vws.exceptions.vws_exceptions.ImageTooLargeError: The given image

is too large.

~vws.exceptions.vws_exceptions.TargetNameExistError: A target with

the given ``name`` already exists.

~vws.exceptions.vws_exceptions.ProjectInactiveError: The project is

inactive.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers. This has been seen to happen when the

given name includes a bad character.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

image_data = _get_image_data(image=image)

image_data_encoded = base64.b64encode(s=image_data).decode(

encoding="ascii",

)

data = {

"name": name,

"width": width,

"image": image_data_encoded,

"active_flag": active_flag,

"application_metadata": application_metadata,

}

content = json.dumps(obj=data).encode(encoding="utf-8")

response = await self.make_request(

method=HTTPMethod.POST,

data=content,

request_path="/targets",

expected_result_code="TargetCreated",

content_type="application/json",

)

return str(object=json.loads(s=response.text)["target_id"])

async def get_target_record(self, target_id: str) -> TargetStatusAndRecord:

"""Get a given target's target record from the Target

Management System.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#target-record.

Args:

target_id: The ID of the target to get details of.

Returns:

Response details of a target from Vuforia.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.UnknownTargetError: The given target

ID does not match a target in the database.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.GET,

data=b"",

request_path=f"/targets/{target_id}",

expected_result_code="Success",

content_type="application/json",

)

result_data = json.loads(s=response.text)

return TargetStatusAndRecord.from_response_dict(

response_dict=result_data,

)

async def wait_for_target_processed(

self,

*,

target_id: str,

seconds_between_requests: float = 0.2,

timeout_seconds: float = 60 * 5,

) -> None:

"""Wait up to five minutes (arbitrary) for a target to

get past the processing stage.

Args:

target_id: The ID of the target to wait for.

seconds_between_requests: The number of seconds to

wait between requests made while polling the

target status.

We wait 0.2 seconds by default, rather than

less, than that to decrease the number of calls

made to the API, to decrease the likelihood of

hitting the request quota.

timeout_seconds: The maximum number of seconds to

wait for the target to be processed.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.custom_exceptions.TargetProcessingTimeoutError: The

target remained in the processing stage for more than

``timeout_seconds`` seconds.

~vws.exceptions.vws_exceptions.UnknownTargetError: The given target

ID does not match a target in the database.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

start_time = asyncio.get_event_loop().time()

while True:

report = await self.get_target_summary_report(

target_id=target_id,

)

if report.status != TargetStatuses.PROCESSING:

# Guard against the target still being seen as

# processing by other endpoints due to eventual

# consistency.

await asyncio.sleep(

delay=seconds_between_requests,

)

return

elapsed_time = asyncio.get_event_loop().time() - start_time

if elapsed_time > timeout_seconds: # pragma: no cover

raise TargetProcessingTimeoutError

await asyncio.sleep(

delay=seconds_between_requests,

)

async def list_targets(self) -> list[str]:

"""List target IDs.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#details-list.

Returns:

The IDs of all targets in the database.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.GET,

data=b"",

request_path="/targets",

expected_result_code="Success",

content_type="application/json",

)

return list(json.loads(s=response.text)["results"])

async def get_target_summary_report(

self, target_id: str

) -> TargetSummaryReport:

"""Get a summary report for a target.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#summary-report.

Args:

target_id: The ID of the target to get a summary

report for.

Returns:

Details of the target.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.UnknownTargetError: The given target

ID does not match a target in the database.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.GET,

data=b"",

request_path=f"/summary/{target_id}",

expected_result_code="Success",

content_type="application/json",

)

result_data = dict(json.loads(s=response.text))

return TargetSummaryReport.from_response_dict(

response_dict=result_data,

)

async def get_database_summary_report(

self,

) -> DatabaseSummaryReport:

"""Get a summary report for the database.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#summary-report.

Returns:

Details of the database.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.GET,

data=b"",

request_path="/summary",

expected_result_code="Success",

content_type="application/json",

)

response_data = dict(json.loads(s=response.text))

return DatabaseSummaryReport.from_response_dict(

response_dict=response_data,

)

async def request_database_reco_counts_report(

self,

*,

year: int,

month: calendar.Month,

) -> RecoCountsReportRequest:

"""Request a per-target recognition count report for the database.

Vuforia generates the report in the background, so the report is not

available to download immediately. Use

:meth:`wait_for_reco_counts_report` to wait for it.

Args:

year: The year to get recognition counts for.

month: The month of the year to get recognition counts for.

Vuforia accepts only the current month and the previous

month. A month taken from a :class:`datetime.datetime` needs

wrapping, as in ``calendar.Month(value=now.month)``.

Returns:

The URL to download the report from, and the transaction ID of

the request.

Raises:

~vws.exceptions.custom_exceptions.DatabaseIdNotSetError: No

``database_id`` was given to the client.

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct, or the client's ``database_id`` is

not the ID of the database which the client's keys belong to.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given year and month are not

the current month or the previous month.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.POST,

data=reco_counts_report_body(year=year, month=month),

request_path=reco_counts_report_path(

database_id=self._database_id,

),

expected_result_code="Success",

content_type="application/json",

)

response_data = dict(json.loads(s=response.text))

return RecoCountsReportRequest.from_response_dict(

response_dict=response_data,

)

async def download_reco_counts_report(

self,

*,

presigned_url: str,

) -> RecoCountsReport:

"""Download a requested reco counts report.

The report's URL is not part of the VWS API, so this request is not

authorized with the client's keys.

Args:

presigned_url: The URL of the report, as given by

:meth:`request_database_reco_counts_report`.

Returns:

The downloaded report.

Raises:

~vws.exceptions.custom_exceptions.RecoCountsReportNotReadyError:

Vuforia has not finished generating the report.

~vws.exceptions.custom_exceptions.RecoCountsReportDownloadError:

The report could not be downloaded. For example, the report's

URL may have expired.

"""

response = await self._transport(

method=HTTPMethod.GET,

url=presigned_url,

headers={},

data=b"",

request_timeout=self._request_timeout_seconds,

)

return report_from_download_response(response=response)

async def wait_for_reco_counts_report(

self,

*,

presigned_url: str,

seconds_between_requests: float = 0.2,

timeout_seconds: float = 60 * 5,

) -> RecoCountsReport:

"""Wait for a requested reco counts report to be generated, then

download it.

Args:

presigned_url: The URL of the report, as given by

:meth:`request_database_reco_counts_report`.

seconds_between_requests: The number of seconds to wait between

requests made while polling the report's URL.

timeout_seconds: The maximum number of seconds to wait for the

report to be generated.

Returns:

The downloaded report.

Raises:

~vws.exceptions.custom_exceptions.RecoCountsReportTimeoutError:

The report was not generated within ``timeout_seconds``

seconds.

~vws.exceptions.custom_exceptions.RecoCountsReportDownloadError:

The report could not be downloaded. For example, the report's

URL may have expired.

"""

start_time = time.monotonic()

while True:

try:

return await self.download_reco_counts_report(

presigned_url=presigned_url,

)

except RecoCountsReportNotReadyError:

elapsed_time = time.monotonic() - start_time

if elapsed_time > timeout_seconds:

raise RecoCountsReportTimeoutError from None

await asyncio.sleep(delay=seconds_between_requests)

async def delete_target(self, target_id: str) -> None:

"""Delete a given target.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#delete.

Args:

target_id: The ID of the target to delete.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.UnknownTargetError: The given target

ID does not match a target in the database.

~vws.exceptions.vws_exceptions.TargetStatusProcessingError: The

given target is in the processing state.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

await self.make_request(

method=HTTPMethod.DELETE,

data=b"",

request_path=f"/targets/{target_id}",

expected_result_code="Success",

content_type="application/json",

)

async def get_duplicate_targets(self, target_id: str) -> list[str]:

"""Get targets which may be considered duplicates of a

given target.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#check.

Args:

target_id: The ID of the target to delete.

Returns:

The target IDs of duplicate targets.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.UnknownTargetError: The given target

ID does not match a target in the database.

~vws.exceptions.vws_exceptions.ProjectInactiveError: The project is

inactive.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

response = await self.make_request(

method=HTTPMethod.GET,

data=b"",

request_path=f"/duplicates/{target_id}",

expected_result_code="Success",

content_type="application/json",

)

return list(

json.loads(s=response.text)["similar_targets"],

)

async def update_target(

self,

*,

target_id: str,

name: str | None = None,

width: float | None = None,

image: _ImageType | None = None,

active_flag: bool | None = None,

application_metadata: str | None = None,

) -> None:

"""Update a target in a Vuforia Web Services database.

See

https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api#update

for parameter details.

Args:

target_id: The ID of the target to update.

name: The name of the target.

width: The width of the target.

image: The image of the target.

active_flag: Whether or not the target is active

for query.

application_metadata: The application metadata of

the target.

This must be base64 encoded, for example by

using::

base64.b64encode('input_string').decode('ascii')

Giving ``None`` will not change the application

metadata.

Raises:

~vws.exceptions.vws_exceptions.AuthenticationFailureError: The

secret key is not correct.

~vws.exceptions.vws_exceptions.BadImageError: There is a problem

with the given image. For example, it must be a JPEG or PNG

file in the grayscale or RGB color space.

~vws.exceptions.vws_exceptions.FailError: There was an error with

the request. For example, the given access key does not match a

known database.

~vws.exceptions.vws_exceptions.MetadataTooLargeError: The given

metadata is too large. The maximum size is 1 MB of data when

Base64 encoded.

~vws.exceptions.vws_exceptions.ImageTooLargeError: The given image

is too large.

~vws.exceptions.vws_exceptions.TargetNameExistError: A target with

the given ``name`` already exists.

~vws.exceptions.vws_exceptions.ProjectInactiveError: The project is

inactive.

~vws.exceptions.vws_exceptions.RequestTimeTooSkewedError: There is

an error with the time sent to Vuforia.

~vws.exceptions.custom_exceptions.ServerError: There is an error

with Vuforia's servers.

~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is

rate limiting access.

"""

data: dict[str, str | bool | float | int] = {}

if name is not None:

data["name"] = name

if width is not None:

data["width"] = width

if image is not None:

image_data = _get_image_data(image=image)

image_data_encoded = base64.b64encode(

s=image_data,

).decode(encoding="ascii")

data["image"] = image_data_encoded

if active_flag is not None:

data["active_flag"] = active_flag

if application_metadata is not None:

data["application_metadata"] = application_metadata

content = json.dumps(obj=data).encode(encoding="utf-8")

await self.make_request(

method=HTTPMethod.PUT,

data=content,

request_path=f"/targets/{target_id}",

expected_result_code="Success",

content_type="application/json",

)