easy-local-alpr/alpr_api.py

import json
import os
import sys
import threading
import time
from time import sleep

import ultimateAlprSdk
from PIL import Image
from flask import Flask, request, jsonify, render_template

counter = 0

"""
Hi there!

This script is a REST API server that uses the ultimateALPR SDK to process images and return the license plate
information. The server is created using Flask and the ultimateALPR SDK is used to process the images.

See the README.md file for more information on how to run this script.
"""

# Defines the default JSON configuration. More information at https://www.doubango.org/SDKs/anpr/docs/Configuration_options.html
JSON_CONFIG = {
    "debug_level": "info",
    "debug_write_input_image_enabled": False,
    "debug_internal_data_path": ".",

    "num_threads": -1,
    "gpgpu_enabled": True,
    "max_latency": -1,

    "klass_vcr_gamma": 1.5,

    "detect_roi": [0, 0, 0, 0],
    "detect_minscore": 0.35,

    "car_noplate_detect_min_score": 0.8,

    "pyramidal_search_enabled": True,
    "pyramidal_search_sensitivity": 0.38,  # default 0.28
    "pyramidal_search_minscore": 0.8,
    "pyramidal_search_min_image_size_inpixels": 800,

    "recogn_rectify_enabled": True,  # heavy on cpu
    "recogn_minscore": 0.4,
    "recogn_score_type": "min"
}

IMAGE_TYPES_MAPPING = {
    'RGB': ultimateAlprSdk.ULTALPR_SDK_IMAGE_TYPE_RGB24,
    'RGBA': ultimateAlprSdk.ULTALPR_SDK_IMAGE_TYPE_RGBA32,
    'L': ultimateAlprSdk.ULTALPR_SDK_IMAGE_TYPE_Y
}


def load_engine():
    bundle_dir = getattr(sys, '_MEIPASS', os.path.abspath(os.path.dirname(__file__)))

    JSON_CONFIG["assets_folder"] = os.path.join(bundle_dir, "assets")
    JSON_CONFIG.update({
        "charset": "latin",
        "car_noplate_detect_enabled": False,
        "ienv_enabled": False,
        "openvino_enabled": False,
        "openvino_device": "GPU",
        "npu_enabled": False,
        "klass_lpci_enabled": False,
        "klass_vcr_enabled": False,
        "klass_vmmr_enabled": False,
        "klass_vbsr_enabled": False,
        "license_token_file": "",
        "license_token_data": ""
    })

    result = ultimateAlprSdk.UltAlprSdkEngine_init(json.dumps(JSON_CONFIG))
    if not result.isOK():
        raise RuntimeError("Init failed: %s" % result.phrase())

    while counter < 3000:
        sleep(1)

    unload_engine()
    load_engine()


def unload_engine():
    result = ultimateAlprSdk.UltAlprSdkEngine_deInit()
    if not result.isOK():
        raise RuntimeError("DeInit failed: %s" % result.phrase())


def process_image(image: Image) -> str:
    global counter
    counter += 1

    width, height = image.size

    if image.mode in IMAGE_TYPES_MAPPING:
        image_type = IMAGE_TYPES_MAPPING[image.mode]
    else:
        raise ValueError("Invalid mode: %s" % image.mode)

    result = ultimateAlprSdk.UltAlprSdkEngine_process(
        image_type,
        image.tobytes(),
        width,
        height,
        0,  # stride
        1  # exifOrientation
    )
    if not result.isOK():
        raise RuntimeError("Process failed: %s" % result.phrase())
    else:
        return result.json()


def create_rest_server_flask():
    app = Flask(__name__)

    @app.route('/v1/image/alpr', methods=['POST'])
    def alpr():
        interference = time.time()

        if 'upload' not in request.files:
            return jsonify({'error': 'No image found'})

        image = request.files['upload']
        if image.filename == '':
            return jsonify({'error': 'No selected file'})

        image = Image.open(image)
        result = process_image(image)
        result = convert_to_cpai_compatible(result)

        if not result['predictions']:
            print("No plate found in the image, attempting to split the image")

            predictions_found = []

            width, height = image.size
            cell_width = width // 3
            cell_height = height // 3
            cells_to_process = [2, 4, 5, 6, 8, 9]

            for cell_index in range(1, 10):
                row = (cell_index - 1) // 3
                col = (cell_index - 1) % 3
                left = col * cell_width
                upper = row * cell_height
                right = left + cell_width
                lower = upper + cell_height

                if cell_index in cells_to_process:
                    cell_image = image.crop((left, upper, right, lower))
                    result_cell = json.loads(process_image(cell_image))

                    if 'plates' in result_cell:
                        for plate in result_cell['plates']:
                            warpedBox = plate['warpedBox']
                            x_coords = warpedBox[0::2]
                            y_coords = warpedBox[1::2]
                            x_min = min(x_coords) + left
                            x_max = max(x_coords) + left
                            y_min = min(y_coords) + upper
                            y_max = max(y_coords) + upper

                            predictions_found.append({
                                'confidence': plate['confidences'][0] / 100,
                                'label': "Plate: " + plate['text'],
                                'plate': plate['text'],
                                'x_min': x_min,
                                'x_max': x_max,
                                'y_min': y_min,
                                'y_max': y_max
                            })

            if predictions_found:
                result['predictions'].append(max(predictions_found, key=lambda x: x['confidence']))

        result['processMs'] = round((time.time() - interference) * 1000, 2)
        result['inferenceMs'] = result['processMs']
        return jsonify(result)

    @app.route('/')
    def index():
        return render_template('index.html')

    return app


def convert_to_cpai_compatible(result):
    result = json.loads(result)

    response = {
        'success': "true",
        'processMs': result['duration'],
        'inferenceMs': result['duration'],
        'predictions': [],
        'message': '',
        'moduleId': 'ALPR',
        'moduleName': 'License Plate Reader',
        'code': 200,
        'command': 'alpr',
        'requestId': 'null',
        'inferenceDevice': 'none',
        'analysisRoundTripMs': 0,
        'processedBy': 'none',
        'timestamp': ''
    }

    if 'plates' in result:
        plates = result['plates']

        for plate in plates:
            warpedBox = plate['warpedBox']
            x_coords = warpedBox[0::2]
            y_coords = warpedBox[1::2]
            x_min = min(x_coords)
            x_max = max(x_coords)
            y_min = min(y_coords)
            y_max = max(y_coords)

            response['predictions'].append({
                'confidence': plate['confidences'][0] / 100,
                'label': "Plate: " + plate['text'],
                'plate': plate['text'],
                'x_min': x_min,
                'x_max': x_max,
                'y_min': y_min,
                'y_max': y_max
            })

    return response


if __name__ == '__main__':
    engine = threading.Thread(target=load_engine, daemon=True)
    engine.start()

    app = create_rest_server_flask()
    app.run(host='0.0.0.0', port=5000)

    unload_engine()