Roboflow Inference

"""Smart Tools starter: one Roboflow object-detection image. Not production-certified."""
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('predictions'), list):
        raise ValueError('unexpected_schema')
    items = []
    for item in raw['predictions']:
        if not isinstance(item, dict) or not isinstance(item.get('class'), str):
            raise ValueError('unexpected_schema')
        score = item.get('confidence')
        if type(score) not in (int, float) or not math.isfinite(score) or not 0 <= score <= 1:
            raise ValueError('unexpected_schema')
        box = {key: item.get(key) for key in ('x', 'y', 'width', 'height')}
        if any(type(n) not in (int, float) or not math.isfinite(n) for n in box.values()):
            raise ValueError('unexpected_schema')
        if box['width'] < 0 or box['height'] < 0:
            raise ValueError('unexpected_schema')
        items.append({'label': item['class'], 'score': score, 'box_center_pixels': box})
    return items


def run():
    start = monotonic()
    result = {'schema_version': '1.0.0', 'provider': 'Roboflow', 'status': 'error',
              'model_version': None, 'data': None, 'error': None,
              'captured_at': datetime.now(timezone.utc).isoformat()}
    try:
        key, model = os.environ['ROBOFLOW_API_KEY'], os.environ['ROBOFLOW_MODEL_ID']
        filename = Path(os.environ['IMAGE_PATH'])
        if not key or '/' not in model or not filename.is_file():
            raise ValueError('configuration')
        from inference_sdk import InferenceHTTPClient, InferenceConfiguration
        result['sdk_version'] = version('inference-sdk')
        result['model_version'] = model
        # Hosted endpoint only; image leaves this machine. Header auth needs server >=1.5.
        client = InferenceHTTPClient(api_url='https://serverless.roboflow.com', api_key=key)
        client.configure(InferenceConfiguration(api_key_transport='header'))
        result['data'] = normalize(client.infer(str(filename), model_id=model))
        result['status'] = 'ok'
    except (KeyError, FileNotFoundError):
        result['error'] = 'configuration'
    except ValueError:
        result['error'] = 'configuration_or_schema'
    except ImportError:
        result['error'] = 'dependency_missing'
    except Exception:
        # Provider exceptions may contain request URLs or credentials: never echo them.
        result['error'] = 'provider_or_transport_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)