Распознаём тексты на Android Things с ABBYY RTR SDK и django

Привет! Меня зовут Азат Калмыков, я студент второго курса ОП “Прикладная математика и информатика” Факультета компьютерных наук НИУ ВШЭ и стажёр в отделе мобильной разработки компании ABBYY. В этом материале я расскажу про свой небольшой проект, выполненный в рамках летней стажировки.

Представьте себе небольшой конвейер. По нему едут товары или какие-то детали, на которых важно распознавать текст (возможно, это некий уникальный идентификатор, а может, и что-то более интересное). Хорошим примером будут посылки. Работу конвейера дистанционно контролирует оператор, который отслеживает неполадки и в случае чего решает проблемы. Что может ему в этом помочь? Девайс на платформе Android Things может быть неплохим решением: он мобильный, легко настраивается и может работать через Wi-Fi. Мы решили попробовать использовать технологии ABBYY и узнать, насколько они подходят для таких ситуаций — распознавания текста в потоке на “нестандартных устройствах” из категории Internet of Things. Мы сознательно будем упрощать многие вещи, так как просто строим концепт. Если стало интересно, добро пожаловать под кат.

Android Things

К нам в офис ABBYY с конференции Google I/O приехала замечательная штука под названием Android Things Starter Kit. Не пропадать же добру, и мы захотели с ней поиграться в поиске различных сценариев использования наших библиотек распознавания. Сначала нужно собрать наш девайс, а потом запустить. Сделать это несложно, достаточно неукоснительно следовать инструкциям от производителя.

Прочитать подробнее про платформу можно тут и тут.

Что пришло в мои руки

А в конце поста я покажу, как выглядит собранный девайс

Что же мы делаем?

Мы напишем приложение под платформу Android Things, которое будет обрабатывать изображение с камеры, отправляя на наш сервер распознанный текст и (периодически) кадры, чтобы условный оператор мог понять, что происходит на конвейере. Сервер будет написан на django.

Спешу заметить, что для выполнения этого проекта от вас не потребуется никаких вложений, а также регистрации и смс (ладно, на AWS всё-таки надо будет зарегистрироваться и получить бесплатный аккаунт).

Запускаем ракету в космос сервер

Будем считать, что у вас уже есть бесплатный аккаунт AWS. Привяжем свою карту, чтобы злой Amazon в случае нашей опрометчивости списал с нас пару шекелей. Воспользуемся AWS EC2 и создадим виртуальную машину на Ubuntu Server 18.04 LTS (HVM) с SSD Volume Type. Для данной ОС и при использовании бесплатного аккаунта доступна только одна конфигурация, выбираем её (не волнуйтесь, одного гигабайта оперативной памяти нам хватит с головой). Создадим ssh-ключ (или используем уже готовый) и попробуем подключиться к нашей машине. Так же создадим Elastic IP (что-то вроде статического IP) и сразу же привяжем к нашей машине. Обратите внимание, что Elastic IP, не привязанный ни к какой виртуальной машине, будет стоить вам денег во время разработки.

Подключаемся к серверу. Устанавливаем на машине необходимый тулкит.

Питон третьей версии предустановлен. Дело осталось за малым.

$ sudo apt-get update
$ sudo apt-get install python3-pip
$ sudo pip3 install virtualenv 

Установим докер, он понадобится нам позже.

$ sudo apt-get install docker.io

Также нужно открыть порт 8000. К нему мы и будем обращаться при использовании веб-сервиса. Порт 22 для ssh открыт по умолчанию.

Ура! Теперь у нас есть удалённый компьютер для запуска наших приложений. Код будем писать прямо на сервере.

Django (+ channels)

Я решил использовать django, так как это позволит быстро и просто создать небольшой веб-сервис. Дополнительная библиотека django channels даст нам возможность поработать с веб-сокетами (а именно, сделать костыльную трансляцию через передачу картинки без обновления страницы).

Создаём директорию, в которой разместим проект. Устанавливаем django вместе с django channels, не отклоняясь от инструкции в документации.

$ mkdir Project
$ cd Project
$ virtualenv venv
$ source venv/bin/activate
$ pip install -U channels # в том числе подтягивает за собой django
$ pip install channels_redis # для взаимодействия с Redis
$ pip install djangorestframework
$ django-admin startproject mysite
$ cd mysite 

Создаём проект. У нас будет 3 поддиректории. Основная будет иметь то же название — mysite (создаётся автоматически), другие две — streaming и uploading. Первая будет отвечать за отображение информации на веб-страничке, а вторая — за её загрузку через REST API.

$ python3 manage.py startapp streaming
$ cd streaming
$ rm -r migrations admin.py apps.py models.py tests.py
$ cd ..
$ python3 manage.py startapp uploading
$ cd uploading
$ rm -r migrations admin.py apps.py models.py tests.py

Настраиваем django channels. Закомментируем строку с WSGI_APPLICATION и добавим новую с ASGI_APPLICATION. Теперь наше приложение будет работать асинхронно.

# mysite/settings.py 

# ...

# WSGI_APPLICATION = ...
ASGI_APPLICATION = 'mysite.routing.application'

# ...

Также обновляем значение списка INSTALLED_APPS.

# mysite/settings.py

# ... 

INSTALLED_APPS = [
    'channels',
    'streaming',
    'uploading',
    'rest_framework',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

# ...

Архитектура

Мы напишем код, основываясь на официальном туториале django channels. Структура нашего небольшого сервиса будет выглядеть следующим образом:

M.Y.I.P:8000/frame — веб-страница, которая будет показывать результат, условно, та страница, на которую смотрит оператор
M.Y.I.P:8000/upload/upload_text/ — адрес для POST-запроса, отправки распознанного текста
M.Y.I.P:8000/upload/upload_image/ — адрес для PUT-запроса, отправки отдельных изображений

Нужно прописать эту логику в файлах urls.py соответствующих директорий.

# mysite/mysite/urls.py

from django.contrib import admin
from django.conf.urls import include, url

urlpatterns = [
    url(r'^frame/', include('streaming.urls')),
    url(r'^upload/', include('uploading.urls')),
]

REST API

Переходим к описанию логики нашего API.

# mysite/uploading/urls.py

from django.conf.urls import url
from rest_framework.urlpatterns import format_suffix_patterns

from . import views

urlpatterns = [
        url(r'^upload_text/$', views.UploadTextView.as_view()),
        url(r'^upload_image/$', views.UploadImageView.as_view()),
]

urlpatterns = format_suffix_patterns(urlpatterns)

# mysite/uploading/views.py

from django.shortcuts import render
from rest_framework.views import APIView
from rest_framework.response import Response
from channels.layers import get_channel_layer
from rest_framework.parsers import FileUploadParser
from asgiref.sync import async_to_sync
import base64

# Create your views here.
class UploadTextView(APIView):
    def post(self, request, format=None):
        message = request.query_params['message']
        if not message:
            raise ParseError("Empty content")

        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)("chat", {
            "type": "chat.message",
            "message": message,
        })

        return Response({'status': 'ok'})

class UploadImageView(APIView):
    parser_class = (FileUploadParser,)

    def put(self, request, format=None):
        if 'file' not in request.data:
            raise ParseError("Empty content")

        f = request.data['file']

        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)("chat", {
            "type": "chat.message",
            "image64": base64.b64encode(f.read()).decode("ascii"),
        })

        return Response({'status': 'ok'})

Веб-страничка

Вся информация уместится на одной страничке, поэтому логика будет несложная.

# mysite/streaming/urls.py

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^', views.index, name='index'),
]

# mysite/streaming/views.py

from django.shortcuts import render
from django.utils.safestring import mark_safe
import json

# Create your views here.
def index(request):
    return render(request, 'index.html', {})

Нам нужно написать небольшой html-документ для отображения результатов. В нём будет встроенный скрипт для подключения к веб-сокету и наполнения контентом.

<!-- mysite/streaming/templates/index.html -->

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8"/>
    <title>Live from Android Things</title>
</head>
<body>
    <textarea id="chat-log" cols="100" rows="20"></textarea><br/>
    <img id="frame">
</body>
<script>
    var chatSocket = new WebSocket(
        'ws://' + window.location.host + '/ws/chat/');
    chatSocket.onmessage = function(e) {
        var data = JSON.parse(e.data);
        var message = data['message'];
                var image64 = data['image64'];

        if (image64) {
                document.querySelector('#frame').setAttribute(
                        'src', 'data:image/png;base64,' + image64  
                    );
        } else if (message) {
                document.querySelector('#chat-log').value += (message + '\n');
        }
    };
    chatSocket.onclose = function(e) {
        console.error('Chat socket closed unexpectedly');
    };
</script>
</html>

Настройка routing, сокетов

Как наиболее удачно перевести слово routing на русский? Попробуем выкинуть из головы этот вопрос и просто настроим его (или её).

# mysite/mysite/settings.py

# ...

ALLOWED_HOSTS = ['*']  # заменяем [] на ['*'], разрешаем все хосты

# ...

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

Теперь нужно прописать логику “пересылки” (файлы routing.py аналогичны файлам urls.py, только теперь для веб-сокетов).

# mysite/mysite/routing.py

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import streaming.routing

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': AuthMiddlewareStack(
        URLRouter(
            streaming.routing.websocket_urlpatterns
        )
    ),
})

# mysite/streaming/routing.py

from django.conf.urls import url

from . import consumers

websocket_urlpatterns = [
    url(r'^ws/chat/$', consumers.FrameConsumer),
]

А теперь реализуем сам FrameConsumer в consumers.py

# mysite/streaming/consumers.py

from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer, JsonWebsocketConsumer
import json

class FrameConsumer(WebsocketConsumer):
    def connect(self):
        self.room_group_name = 'chat'

        # Join room group
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,
            self.channel_name
        )

        self.accept()

    def disconnect(self, close_code):
        # Leave room group
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Send message to room group
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    # Receive message from room group
    def chat_message(self, event):
        if 'message' in event:
            # Send message to WebSocket
            self.send(text_data=json.dumps({
                'message': event['message']
            }))
        elif 'image64' in event:
            self.send(text_data=json.dumps({
                'image64': event['image64']
            }))

Ну и наконец, с потными ладошками запускаем.

$ docker run -p 6379:6379 -d redis:2.8 
$ python manage.py runserver 0.0.0.0:8000

А теперь собственно про Android

Мы будем использовать RTR SDK от ABBYY для распознавания текста. Ultimate pack RTR SDK для наших целей можно скачать вот тут. Мы реализуем довольно простой интерфейс для обработки кадров, наше приложение будет основано на сэмпле из скачанного по предыдущей ссылке архива (/sample-textcapture). Мы выкинем из приложения лишние части, немного отшлифуем для работы с конкретно Android Things и реализуем общение с сервером.

Библиотечный файл .aar лежит в директории libs скачанного архива, импортируем. Скопируем содержимое директории assets архива (там по сути файлы необходимые для самого процесса распознавания) в assets нашего проекта. Туда же скопируем файл лицензии из архива, без него приложение не запустится.

Для того чтобы реализовать нужный нам функционал ABBYY RTR SDK, нужно создать объект типа Engine, а с помощью уже него объект типа ITextCaptureService, который мы позже запустим.

try {
    mEngine = Engine.load(this, LICENSE_FILE_NAME);
    mTextCaptureService = mEngine.createTextCaptureService(textCaptureCallback);

    return true;
} // ...

В этом случае нужно передать объект типа ITextCaptureService.Callback, создадим его прямо в нашем классе MainActivity, он должен реализовывать 3 метода.

private ITextCaptureService.Callback textCaptureCallback = new ITextCaptureService.Callback() {

    @Override
    public void onRequestLatestFrame(byte[] buffer) {
        // Метод хочет, чтобы мы заполнили полученный буфер новым кадром.
        // Мы делегируем это камере.
        mCamera.addCallbackBuffer(buffer);
    }

    @Override
    public void onFrameProcessed(
            ITextCaptureService.TextLine[] lines,
            ITextCaptureService.ResultStabilityStatus resultStatus, ITextCaptureService.Warning warning) {
        // Здесь мы получаем результаты обработки изображения, то есть текст

        if (resultStatus.ordinal() >= 3) {
            // Результаты достаточно стабильны, чтобы показать их пользователю
            mSurfaceViewWithOverlay.setLines(lines, resultStatus);
        } else {
            // Нестабильный результат, лучше ничего не показывать
            mSurfaceViewWithOverlay.setLines(null, ITextCaptureService.ResultStabilityStatus.NotReady);
        }

        // Показываем warnings
        // ...
    }

    @Override
    public void onError(Exception e) {
        // Здесь обрабатываем ошибки
    }        
};

Мы делегировали получение кадра объекту камеры. Покажу, что происходит внутри.

private Camera.PreviewCallback cameraPreviewCallback = new Camera.PreviewCallback() {
    @Override
    public void onPreviewFrame(byte[] data, Camera camera) {
        // Если пришло время отправлять (если ещё ничего не отправляется)
        if (!mIsUploading) {
            mIsUploading = true;

            // Отправляем на сервер
            new UploadImageTask(mCameraPreviewSize.width, mCameraPreviewSize.height).execute(data);
        }

        // Заполняем полученный ранее буфер
        mTextCaptureService.submitRequestedFrame(data);
    }
};

Для отправки сообщений напишем пару классов, которые в свою очередь будут делегировать свою работу объекту класса Uploader.

public static class UploadTextTask extends AsyncTask<String, Void, Void> {

    @Override
    protected Void doInBackground(String... params) {
        mUploader.uploadText(params[0]);
        return null;
    }

}

public static class UploadImageTask extends AsyncTask<byte[], Void, Void> {

    private int mCameraPreviewWidth;
    private int mCameraPreviewHeight;

    public UploadImageTask(int width, int height) {
        mCameraPreviewWidth = width;
        mCameraPreviewHeight = height;
    }

    @Override
    protected Void doInBackground(final byte[]... params) {
        byte[] jpegBytes = convertToJpegBytes(params[0]);
        if (jpegBytes != null) {
            mUploader.uploadImage(jpegBytes);
        }
        return null;
    }

    private byte[] convertToJpegBytes(byte[] rawBytes) {
        YuvImage yuvImage = new YuvImage(
                rawBytes,
                ImageFormat.NV21,
                mCameraPreviewWidth,
                mCameraPreviewHeight,
                null
        );

        try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
            yuvImage.compressToJpeg(
                    new Rect(0, 0, mCameraPreviewWidth, mCameraPreviewHeight),
                    40,
                    os
            );
            return os.toByteArray();
        } catch (IOException e) {
            Log.d(TAG, "compress error");
            return null;
        }
    }

    // ...
}

Само общение с сетью в классе Uploader реализовано с помощью удобной библиотеки OkHttp3. Она позволяет сильно упростить взаимодействие с сервером.

Результат

Получаем работающее клиент-серверное приложение с распознавалкой от ABBYY, встроенное в Internet of Things, – ну не круто ли?

Собранный девайс и небольшая нативная реклама моего работодателя

Текст распознался

Селфи-панорама с обзором нескольких устройств

Видос, как это всё может выглядеть в реале

Репозитории на github:

→ AndroidThingsTextRecognition-Backend
→ AndroidThingsTextRecognition-Android

Забирайте и пользуйтесь!