Google Cloud Vision API

"""Smart Tools: one cloud label request. Offline-tested; live integration untested."""
import json
import math
import os
import sys
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from time import monotonic


class ProviderFailure(Exception):
    pass


def normalize(raw):
    if not isinstance(raw, dict) or type(raw.get('error_code')) is not int:
        raise ValueError('unexpected_schema')
    if raw['error_code'] != 0:
        raise ProviderFailure()
    labels = raw.get('labels')
    if not isinstance(labels, list):
        raise ValueError('unexpected_schema')
    result = []
    for item in labels:
        if not isinstance(item, dict):
            raise ValueError('unexpected_schema')
        label, score = item.get('label'), item.get('score')
        if not isinstance(label, str) or not label.strip():
            raise ValueError('unexpected_schema')
        if type(score) not in (int, float) or not math.isfinite(score) or not 0 <= score <= 1:
            raise ValueError('unexpected_schema')
        result.append({'label': label, 'score': score})
    return result


def run():
    start = monotonic()
    result = {'schema_version': '1.0.0', 'provider': 'Google Cloud Vision',
              'status': 'error', 'model_version': None, 'feature': 'LABEL_DETECTION',
              'endpoint': 'vision.googleapis.com', 'data': None, 'error': None,
              'captured_at': datetime.now(timezone.utc).isoformat()}
    try:
        project = os.environ['GOOGLE_CLOUD_QUOTA_PROJECT'].strip()
        if not project:
            raise ValueError('configuration')
        with Path(os.environ['IMAGE_PATH']).open('rb') as image:
            content = image.read(5 * 1024 * 1024 + 1)
        # Local 5 MiB policy for this starter; not a quoted provider maximum.
        if not content or len(content) > 5 * 1024 * 1024:
            raise ValueError('starter_image_limit')
        import google.auth
        from google.auth.exceptions import DefaultCredentialsError, RefreshError
        from google.api_core import exceptions as api_errors
        from google.cloud import vision_v1
        result['sdk_version'] = version('google-cloud-vision')
        try:
            credentials, _ = google.auth.default(
                scopes=['https://www.googleapis.com/auth/cloud-platform'],
                quota_project_id=project)
            request = {'requests': [{'image': {'content': content},
                       'features': [{'type_': vision_v1.Feature.Type.LABEL_DETECTION,
                                     'max_results': 10}]}]}
            # Bytes leave this machine. No automatic RPC retry; cloud model revision is not exposed here.
            with vision_v1.ImageAnnotatorClient(
                    credentials=credentials,
                    client_options={'api_endpoint': 'vision.googleapis.com'}) as client:
                batch = client.batch_annotate_images(request=request, retry=None, timeout=30)
            if len(batch.responses) != 1:
                raise ValueError('unexpected_schema')
            response = batch.responses[0]
            raw = {'error_code': response.error.code,
                   'labels': [{'label': x.description, 'score': x.score}
                              for x in response.label_annotations]}
            result['data'] = normalize(raw)
            result['status'] = 'ok'
        except (DefaultCredentialsError, RefreshError):
            result['error'] = 'authentication'
        except (api_errors.Unauthenticated, api_errors.PermissionDenied):
            result['error'] = 'authentication_or_permission'
        except api_errors.ResourceExhausted:
            result['error'] = 'quota_or_capacity'
        except api_errors.DeadlineExceeded:
            result['error'] = 'deadline_exceeded'
        except api_errors.InvalidArgument:
            result['error'] = 'invalid_request'
        except api_errors.GoogleAPICallError:
            result['error'] = 'provider_or_transport_failure'
    except (KeyError, OSError):
        result['error'] = 'configuration'
    except ImportError:
        result['error'] = 'dependency_missing'
    except ProviderFailure:
        result['error'] = 'provider_image_error'
    except (ValueError, TypeError, AttributeError):
        result['error'] = 'configuration_or_schema'
    except Exception:
        # Never print provider exception strings, credential paths or image contents.
        result['error'] = 'unexpected_failure'
    result['elapsed_ms'] = round((monotonic() - start) * 1000)
    return result


if __name__ == '__main__':
    output = run()
    print(json.dumps(output, ensure_ascii=False, allow_nan=False))
    sys.exit(0 if output['status'] == 'ok' else 1)