Хабрахабр

Распознаём тексты на 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. Сделать это несложно, достаточно неукоснительно следовать инструкциям от производителя. Сначала нужно собрать наш девайс, а потом запустить.

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

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

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

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

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

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

Привяжем свою карту, чтобы злой Amazon в случае нашей опрометчивости списал с нас пару шекелей. Будем считать, что у вас уже есть бесплатный аккаунт AWS. 04 LTS (HVM) с SSD Volume Type. Воспользуемся AWS EC2 и создадим виртуальную машину на Ubuntu Server 18. Создадим ssh-ключ (или используем уже готовый) и попробуем подключиться к нашей машине. Для данной ОС и при использовании бесплатного аккаунта доступна только одна конфигурация, выбираем её (не волнуйтесь, одного гигабайта оперативной памяти нам хватит с головой). Обратите внимание, что Elastic IP, не привязанный ни к какой виртуальной машине, будет стоить вам денег во время разработки. Так же создадим Elastic IP (что-то вроде статического 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 channels даст нам возможность поработать с веб-сокетами (а именно, сделать костыльную трансляцию через передачу картинки без обновления страницы). Я решил использовать django, так как это позволит быстро и просто создать небольшой веб-сервис.

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

$ 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

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

# 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.

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

Нужно прописать эту логику в файлах 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", ) 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

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

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

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

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

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

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, – ну не круто ли?

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

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

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

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

Репозитории на github:
https://github.com/CookiesDeathCookies/AndroidThingsTextRecognition-Backend
https://github.com/CookiesDeathCookies/AndroidThingsTextRecognition-Android

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

Теги
Показать больше

Похожие статьи

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *

Кнопка «Наверх»
Закрыть