Как я управлял автомобилем через обычный сайт

Поговорим про возможность управлять различными устройствами через сайт. В прошлые выходные я решил провести необычный для себя эксперимент — выполнить пару простых упражнений на автодроме, управляя автомобилем через самый обычный сайт. Более того, стало интересно сделать своими руками минимальный функционал для выполнения упражнений в беспилотном режиме.

Прежде всего, для беспилотного управления требуется выполнить разметку данных. Человек, который размечает данные называется асессором. Важно понимать, что асессор — это вид деятельности, а не профессия, так как бывают специальные наборы данных (например, снимки магнитно-резонансного томографа), которые может размечать только опытный специалист соответствующего направления.

Интерфейс для асессора достаточно простой и состоит всего из двух блоков (фотография и кнопки с оценками). Необходимо посмотреть на предложенную фотографию и осмысленно нажать одну из кнопок. Информация об оценке запишется в базу данных, а в интерфейсе загрузится очередная фотография и процедура повторится. В качестве примера создам описанный интерфейс. Пока нет данных с видеорегистратора добавлю фотографию имаго того самого комара (culex pipiens), который укусил меня прошлой ночью:

В ряде случаев, во время разметки фотографий применяют окулографию. К рыбам selachii этот способ не имеет ровным счётом никакого отношения. Окулография (Eye tracking) — это отслеживание взгляда человека. Оно позволяет нанести на изображения метки в тех местах, которые человек более внимательно осматривал. Вот зона повышенного внимания моего взгляда, когда я увидел в окно пролетающий мимо самолёт:

Когда мы снимаем видео на мобильный телефон, то датчик звука (микрофон) и набор фотодиодов (матрица видеокамеры) регистрируют показания, а специальные программно-аппаратные компоненты преобразуют собранную информацию в поток данных или готовый файл нужного формата. Другое устройство регистрирует направление взгляда и записывает массив точек. После этого достаточно наложить любым понравившимся способом эти точки на картинку. Аналогично и с другими устройствами, например, на видео можно накладывать показания неинвазивного нейроинтерфейса — электроэнцефалографа. По сути, мы просто совмещаем показания разных устройств.

Но как же происходит получения информации с датчиков? Сейчас следует копнуть глубже. Согласно общеизвестному определению, датчик — это устройство, которое преобразует измеряемую величину в пригодный для регистрации сигнал. Упрощённый пример: с помощью мультиметра видно, что датчик света меняет своё сопротивление при облучении светом. Отсюда и название «фоторезистор» (от латинских слов: свет и сопротивление). В конструкции фоторезистора есть пластина из сульфида кадмия, именно она уменьшает свое сопротивление при увеличении интенсивности света.

Если я регистрирую показания датчика с помощью мультиметра:

Если регистрация через платформу, то данные уже оцифрованы:

Подобную регистрацию можно провести с помощью специальной платформы, например, широко известной Arduino, которая позволит увидеть на компьютере (Serial.print) прочитанные (analogRead) данные. Все собранные данные логичнее записать в любое удобное для нас хранилище (PostgreSQL, MySQL, Elasticsearch, Redis) или в очередь (RabbitMQ, Kafka). Однако, в данном конкретном случае мне будет нужно реализовать несколько иной вариант — прямой вызов через API по протоколу HTTP. Начать я решил с написания простого отладочного скрипта, который выполняет обучение модели и проверяет, что система работает:

import logging
import requests
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
from catboost import CatBoostClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, confusion_matrix

#
# Основные настройки (константы)
#
LOGFILE='ml.log'
HOST='http://127.0.0.1:8080'

#
# Конфигурация логирования
#
logging.basicConfig(level=logging.INFO, filename=LOGFILE)

#
# Запрос набора данных
#
raw = requests.get('{}/dataset'.format(HOST)).json()

#
# Загрузка в Pandas для комфортной работы
#
df = pd.DataFrame(raw['dataset'], columns=['label', 'a', 'b'])

#
# Описательная статистика
#
logging.info(df.describe())

#
# Сгруппируем по меткам и посмотрим среднее значение
#
logging.info(df.groupby(['label']).mean())

#
# Настройки стилей для отображения графики
#
sns.set_style('whitegrid')

#
# Визуально отобразим набор данных
#
sns_plot = sns.jointplot(data=df, x='a', y='b', hue='label')
sns_plot.savefig('jointplot.png')

#
# Разделение на train и test для последующей проверки
#
X_train, X_test, y_train, y_test = train_test_split(df[['a', 'b']], df['label'])

#
# Обучение модели
#
model = CatBoostClassifier(iterations=500, depth=2, verbose=False)
model.fit(X_train, y_train)

#
# Проверка модели (F1)
#
logging.info(f1_score(y_test, model.predict(X_test)))

#
# Проверка модели (интуитивно понятный вариант)
#
logging.info(confusion_matrix(y_test, model.predict(X_test)))

#
# Посмотрим на важность признаков
#
logging.info(model.get_feature_importance())

#
# Сохраним предварительную версию модели
#
model.save_model('_model_v0')

#
# Функция для простого получения данных
#
def fetch(action):
    return requests.get('{}/{}'.format(HOST, action)).json()

#
# Проверим, что API позволяет получать информацию от датчиков
#
logging.info(fetch('recive')['result'] == [192, 70])

#
# Проверим, что API позволяет запускать устройства
#
logging.info(fetch('launch')['result'] == 0)

#
# Проверим, что модель работает через API
#
logging.info(fetch('predict/48,33')['result'] == 1)

Наличия методов чтения показаний датчиков и запуска произвольного устройства уже достаточно для управления автомобилем через сайт. Добавим возможность принимать решения в полностью автоматическом режиме. Пусть сервер для первого прототипа будет таким:

from aiohttp import web
from robot import Device
from datasets import load_car
from catboost import CatBoostClassifier

#
# Произвольный классификатор (в данном случае CatBoost)
#
model = CatBoostClassifier()

#
# Просто обёртка над коммерческой библиотекой для платформы
#
device = Device()

#
# Отладка работы с машинным обучением
#
async def predict(request):
    if model.is_fitted() == False:
        model.load_model('_model_v0')
    data = request.match_info.get('data', [])
    result = {'result': int(model.predict([data.split(',')]) == [1])}
    return web.json_response(result)

#
# Получение набора данных для машинного обучения
#
async def dataset(request):
    result = {'dataset': load_car()}
    return web.json_response(result)

#
# Отладка получения информации с датчиков
#
async def recive(request):
    result = {'result': device.recive()}
    return web.json_response(result)

#
# Отладка запуска устройств
#
async def launch(request):
    result = {'result': device.launch()}
    return web.json_response(result)

#
# Сервер
#
app = web.Application()
app.add_routes([
    web.get('/predict/{data}', predict),
    web.get('/dataset', dataset),
    web.get('/recive', recive),
    web.get('/launch', launch)
])

if __name__ == '__main__':
    web.run_app(app)

Обратите внимание, что модель принимает на вход только два предиктора. Такое сокращение размерности нужно для удобства визуализации. В реальных беспилотных системах это не будет нормально работать, но в моём упрощённом эксперименте такого решения вполне достаточно. Важно пояснить, что все эксперименты проводились на специальном спортивном объекте, а не на дорогах общего пользования.

Всего несколько строк кода, но этого хватит, чтобы «разум» беспилотного автомобиля был способен действовать самостоятельно (в рамках одного упражнения). Будем рассматривать это как задачу бинарной классификации. Есть только два состояния: тормозить и ничего не предпринимать. Если интересно, то мера F1 в моём конкретном случае была равна 0.92.

Попробуем конструкцию в работе? Разгоняюсь. Скорость 30 км/ч. На дороге препятствие — воздушный шарик с гелием, который привязан к небольшой деревянной палочке. Приблизительно за 25 метров происходит срабатывание устройства торможения. Код для полностью беспилотного выполнения упражнения (экстренное торможение):

from robot import Robot
from classifiers import CarStopClassifier

class CarBreak:

    def __init__(self):
        self.robot = Robot()
        self.classifier = CarStopClassifier()

    def check(self):
        return self.classifier.predict([self.robot.recive()]) == [1]

    def stop(self):
        return self.robot.stop()

if __name__ == '__main__':
    car_break = CarBreak()
    while True:
        if car_break.check():
            print('STOP!', car_break.stop())
            break

Как вы понимаете, вместо автомашины может быть кондиционер или беспилотный подводный аппарат. А вместо видеокамеры и пары датчиков (в заметке я всё это называю «датчиками») может быть, допустим, один единственный термометр или сразу много устройств. Я намеренно постарался показать наиболее общий подход. Самое важное: всегда соблюдайте технику безопасности и выполняйте подобные «эксперименты выходного дня» только на специальных спортивных объектах.