Amazon Rekognition

"""One authorized JPEG/PNG sent to AWS DetectLabels; no automatic retry."""
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


def normalize(raw):
    if not isinstance(raw, dict) or not isinstance(raw.get('Labels'), list):
        raise ValueError('unexpected_schema')
    output = []
    for item in raw['Labels']:
        if not isinstance(item, dict) or not isinstance(item.get('Name'), str):
            raise ValueError('unexpected_schema')
        confidence = item.get('Confidence')
        if type(confidence) not in (int, float) or not math.isfinite(confidence) or not 0 <= confidence <= 100:
            raise ValueError('unexpected_schema')
        output.append({'label': item['Name'], 'score': confidence / 100})
    model = raw.get('LabelModelVersion')
    if model is not None and not isinstance(model, str):
        raise ValueError('unexpected_schema')
    return output, model


def create_client(region):
    import boto3
    from botocore.config import Config
    settings = Config(connect_timeout=5, read_timeout=30,
                      retries={'mode': 'standard', 'total_max_attempts': 1})
    # Standard AWS credentials chain; no access keys in source code.
    return boto3.client('rekognition', region_name=region, config=settings)


def classify_error(exc):
    code = getattr(exc, 'response', {}).get('Error', {}).get('Code')
    if code in ('AccessDeniedException', 'UnrecognizedClientException', 'InvalidSignatureException', 'ExpiredTokenException'):
        return 'authorization_failure'
    if code in ('ThrottlingException', 'ProvisionedThroughputExceededException'):
        return 'throttled'
    if code in ('InvalidImageFormatException', 'ImageTooLargeException', 'InvalidParameterException'):
        return 'provider_input_failure'
    if type(exc).__name__ in ('NoCredentialsError', 'PartialCredentialsError', 'ProfileNotFound'):
        return 'credentials_missing'
    if type(exc).__name__ in ('ReadTimeoutError', 'ConnectTimeoutError', 'EndpointConnectionError'):
        return 'transport_failure'
    return 'provider_or_transport_failure'


def run(client_factory=create_client):
    start = monotonic()
    result = {'schema_version': '1.0.0', 'provider': 'Amazon Rekognition', 'status': 'error',
              'model_version': None, 'data': None, 'error': None,
              'captured_at': datetime.now(timezone.utc).isoformat()}
    try:
        region = os.environ['AWS_REGION'].strip()
        filename = Path(os.environ['IMAGE_PATH'])
        if not region or filename.suffix.lower() not in ('.jpg', '.jpeg', '.png'):
            raise ValueError('configuration')
        # Bounded read: this starter's cap is 4 MiB, not a claimed service limit.
        with filename.open('rb') as image:
            content = image.read(4 * 1024 * 1024 + 1)
        if not content or len(content) > 4 * 1024 * 1024:
            raise ValueError('input_size')
        client = client_factory(region)
        try:
            raw = client.detect_labels(Image={'Bytes': content}, MaxLabels=20,
                                       MinConfidence=70, Features=['GENERAL_LABELS'])
            result['data'], result['model_version'] = normalize(raw)
        finally:
            client.close()
        result['region'] = region
        result['status'] = 'ok'
    except (KeyError, OSError):
        result['error'] = 'configuration'
    except (ValueError, TypeError):
        result['error'] = 'configuration_or_schema'
    except ImportError:
        result['error'] = 'dependency_missing'
    except Exception as exc:
        result['error'] = classify_error(exc)
    if result['status'] != 'ok':
        result['data'] = None
    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)