Фото: kwan fung на сайте Unsplash

Уже несколько лет не снижается ажиотаж вокруг IoT-устройств. Эти устройства могут быть почти чем угодно: от будильника, показывающего погоду, до холодильника, сообщающего о ценах в ближайших продуктовых магазинах. Какой бы ни была реализация, для общения с источниками данных эти устройства используют API. Но как конкретно подключаются сообщения, данные и устройства?

В этом посте мы покажем пример проектирования и моделирования данных для IoT-устройства. Для этого будет использовано M5Stack — небольшое модульное IoT-устройство с экраном, и подключение к API Metropolitan Transportation Authority Нью-Йорка (MTA) для получения актуального графика движения поездов на разных станциях.

Хотя мы будем работать с M5Stack, рассмотренные в статье концепции применимы к проектированию IoT-приложения для широкого спектра устройств.

▍ Требования

В этом туториале мы сосредоточимся на общих концептуальных идеях запросов данных от API. Очень полезным будет знание программирования. Хотя наличие M5Stack необязательно, если у вас есть это устройство, вы сможете загрузить готовый проект на собственное устройство.

Для начала скачайте IDE VS Code и плагин M5Stack. Если вы никогда не запускали M5Stack, то следуйте указаниям производителя по настройке WiFi и необходимого встроенного ПО. Для этого проекта мы будем применять Python 3 — основной язык программирования, используемый M5Stack.

Вам понадобится зарегистрировать аккаунт MTA, чтобы получить бесплатный ключ API разработчика для доступа к данным метро, обновляемым в реальном времени.

Кроме того, необходимо зарегистрировать бесплатный аккаунт Gravitee, чтобы использовать API designer, который упрощает визуализацию и понимание потоков данных в вызовах API.

Источником вдохновения для этого проекта послужил данный опенсорсный проект, поэтому если этот репозиторий был вам полезен, наградите его звездой.

Проектирование взаимодействия с API

Прежде чем приступать к написанию кода, давайте подумаем, какая информация нам нужна для выполнения этого проекта:

Информация о станциях метро.
Какие поезда проходят через эти станции.
Последние данные в реальном времени об этих поездах.

Согласно документации, API разделён на фиды статических данных и фиды данных реального времени.

Фиды статических данных содержат информацию о станциях. С помощью этой информации мы можем получать актуальные данные о поездах от API фидов данных реального времени. MTA предоставляет данные в следующем формате CSV:

stop_id,stop_code,stop_name,stop_desc,stop_lat,stop_lon,zone_id,stop_url,location_type,parent_station

Так как единственная необходимая нам статическая информация — это ID станции, мы можем просто подставить ID случайной станции и использовать его для фидов реального времени. В данном случае я выбрал станцию «Хойт-стрит — Скермерхорн-стрит» из-за её относительной сложности: через неё проходит два поезда (A и C). Также станции идентифицируются по направлению: на север (N) или на юг (S).

A42,,Hoyt-Schermerhorn Sts,,40.688484,-73.985001,,,1,
A42N,,Hoyt-Schermerhorn Sts,,40.688484,-73.985001,,,0,A42
A42S,,Hoyt-Schermerhorn Sts,,40.688484,-73.985001,,,0,A42

Из этих строк нам нужна лишь ID родительской станции (A42) для идентификации поездов, проходящих через станцию как на север (A42N), так и на юг (A42S).

Фиды реального времени передаются в формате Google GTFS, основанном на буферах протоколов (также называемых protobuf). Хотя у MTA нет задокументированных примеров её конкретных фидов, они есть у GTFS. Из документации GTFS мы можем понять, как получать время прибытия ближайших поездов на конкретную станцию в формате protobuf.

Вот пример ответа конечной точки GTFS, для простоты визуализации конвертированный в JSON:

{
  "trip":{
     "trip_id":"120700_A..N",
     "start_time":"20:07:00",
     "start_date":"20220531",
     "route_id":"A"
  },
  "stop_time_update":[
     {
        "arrival":{
           "time":1654042672
        },
        "departure":{
           "time":1654042672
        },
        "stop_id":"H06N"
     },

     //…другие остановки…

     {
        "arrival":{
           "time":1654044957
        },
        "departure":{
           "time":1654044957
        },
        "stop_id":"A42N"
     }
  ]
}

Так как API MTA возвращает большой объём информации, будет очень полезно применить Gravitee API Designer для моделирования того, что возвращает API, сопоставления данных и их визуализации. Вот скриншот нашей диаграммы связей API Designer:

API Designer позволяет идентифицировать все ресурсы (конечные точки) API, а также атрибуты данных, связанные с ресурсами. Среди таких атрибутов могут быть необходимые конечной точке входящие данные и предоставляемые ею исходящие данные.

На нашей диаграмме есть ресурс с путём /gtfs/. Мы можем подключить любое количество атрибутов и аннотировать каждый из этих атрибутов типами данных. Посмотрев на диаграмму, мы можем провести прямой путь от конечной точки до времени прибытия и отправления, указанного в правом нижнем углу.

Чтобы описать нужные данные, нам необходимо:

Определить ID станции, от которой мы хотим получать информацию о поездах.
Отправить HTTP-запрос к GTFS-фиду MTA для тех линий, которые нас интересуют.
Итеративно обойти результаты, сравнивая stop_id в массиве ответа с ID нашей станции.
Затем мы можем обработать информацию о времени для конкретной станции и поезда.

Задача состоит из нескольких элементов, но здесь нет ничего такого, с чем бы мы не справились!

Пишем код

Прежде чем запускать что-либо на M5Stack, давайте убедимся, что наш код работает локально. Мы установим несколько пакетов Python, чтобы упростить сборку проекта.

pip3 install --upgrade gtfs-realtime-bindings
pip3 install protobuf3_to_dict
pip3 install requests

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

Мы начнём программу с импорта пакетов Python:

from google.transit import gtfs_realtime_pb2
import requests
import time

Затем отправим HTTP-запрос к GTFS-фиду MTA:

api_key = "YOUR_API_KEY"

# Запрашиваем фид данных состояния метро у API MTA
headers = {'x-api-key': api_key}
feed = gtfs_realtime_pb2.FeedMessage()
response = requests.get(
    'https://api-endpoint.mta.info/Dataservice/mtagtfsfeeds/nyct%2Fgtfs-ace',
    headers=headers)
feed.ParseFromString(response.content)

Используемая здесь конечная точка GTFS отвечает за поезда A/C/E, что можно понять по суффиксу -ace в URL. (Хотя для этого демо нам не нужен поезд E.)

Давайте преобразуем этот ответ буфера протоколов GTFS в словарь:

from protobuf_to_dict import protobuf_to_dict
subway_feed = protobuf_to_dict(feed)  # преобразуем фид данных MTA в словарь
realtime_data = subway_feed['entity']

На этом этапе я крайне рекомендую использовать print(realtime_data), чтобы мы видели, как выглядит реальная структура данных. Если бы это был реальный проект, такой анализ помог бы нам понять, какие ключи и значения словаря нужно обрабатывать, но мы уже это разобрали.

def station_time_lookup(train_data, station):
   for trains in train_data:
       if trains.__contains__('trip_update'):
           unique_train_schedule = trains['trip_update']
           if unique_train_schedule.__contains__('stop_time_update'):
             unique_arrival_times = unique_train_schedule['stop_time_update']
             for scheduled_arrivals in unique_arrival_times:
                 stop_id = scheduled_arrivals.get('stop_id', False)
                 if stop_id == f'{station}N':
                     time_data = scheduled_arrivals['arrival']
                     unique_time = time_data['time']
                     if unique_time != None:
                         northbound_times.append(unique_time)
                 elif stop_id == f'{station}S':
                     time_data = scheduled_arrivals['arrival']
                     unique_time = time_data['time']
                     if unique_time != None:
                         southbound_times.append(unique_time)

# Сохраняем глобальный список для хранения времени разных поездов
northbound_times = []
southbound_times = []

# Выполняем написанную выше функцию для ID станции "Хойт-стрит — Скермерхорн-стрит"
station_time_lookup(realtime_data, 'A42')

Ого, внезапно у нас стало много кода! Но не волнуйтесь — он не особо сложен:

Мы итеративно обходим массив информации о поездах на линиях A/C.
Для каждого элемента массива проверяем, что существуют значения для всех нужных нам ключей. Это безопасное программирование, поскольку мы не уверены полностью, что данный сторонний сервис будет иметь нужные нам данные в нужный момент!
После этого итеративно обходим всю информацию о станциях и останавливаемся, когда находим нужный нам родительский ID (A42) для поездов, идущих на север и на юг.
В конце мы сохраняем списки времени прибытия поездов в две глобальные переменные.

Далее представим эту информацию в удобном виде:

# Сортируем полученное время в хронологическом порядке
northbound_times.sort()
southbound_times.sort()

# Извлекаем из списка первое и второе ближайшее время прибытия
nearest_northbound_arrival_time = northbound_times[0]
second_northbound_arrival_time = northbound_times[1]

nearest_southbound_arrival_time = southbound_times[0]
second_southbound_arrival_time = southbound_times[1]

### ЗДЕСЬ ДОЛЖЕН НАХОДИТЬСЯ UI ДЛЯ M5STACK ###

def print_train_arrivals(
        direction,
        time_until_train,
        nearest_arrival_time,
        second_arrival_time):
    if time_until_train <= 0:
        next_arrival_time = second_arrival_time
    else nearest_arrival_time:
        next_arrival_time_s = time.strftime(
            "%I:%M %p",
            time.localtime(next_arrival_time))
    print(f"The next {direction} train will arrive at {next_arrival_time_s}")

# Получаем текущее время, чтобы вычислить количество минут до прибытия
current_time = int(time.time())
time_until_northbound_train = int(
    ((nearest_northbound_arrival_time - current_time) / 60))
time_until_southbound_train = int(
    ((nearest_southbound_arrival_time - current_time) / 60))
current_time_s = time.strftime("%I:%M %p")
print(f"It's currently {current_time_s}")

print_train_arrivals(
    "northbound",
    time_until_northbound_train,
    nearest_northbound_arrival_time,
    second_northbound_arrival_time)
print_train_arrivals(
    "southbound",
    time_until_southbound_train,
    nearest_southbound_arrival_time,
    time_until_southbound_train)

В основном этот код выполняет форматирование данных. Он состоит из следующих ключевых этапов:

Мы сортируем время прибытия на станцию поездов на север и на юг.
Берём первые два времени (прибытие ближайших поездов).
Сравниваем это время с текущим временем, чтобы получить время в минутах до прибытия поезда. Передаём это время прибытия поезда в print_train_arrivals.
Если следующий поезд прибывает меньше, чем через минуту, то отображаем второе время прибытия — боюсь, на этот поезд вы не успеете! В противном случае показываем ближайшее время прибытия.

Если запустить скрипт в терминале, то вы увидите подобное сообщение:

It's currently 05:59 PM
The next northbound train will arrive at 06:00 PM
The next southbound train will arrive at 06:02 PM

Развёртываем код на M5Stack

Протестировав локально обмен данными кода на Python с API MTA, можно запустить этот код на M5Stack. Проще всего запрограммировать M5Stack при помощи бесплатного IDE UI Flow, который является простой веб-страницей, общающейся с устройством через WiFi. Подробнее о настройке устройства для доступа к WiFi можно прочитать в документации.

Хотя M5Stack можно программировать через UI-элементы WYSIWYG, устройство также может принимать (и исполнять) код на Python. Однако главное преимущество элементов WYSIWYG заключается в том, что они сильно упрощают визуализацию отображаемого на экране текста:

В этом GIF я создал метку со стандартной строкой «Text» на примере экрана M5Stack. При переключении на Python мы видим, что эта метка является экземпляром объекта M5TextBox. При перетаскивании метки её координаты X и Y (первые два аргумента конструктора) меняются в Python. Благодаря этому, можно легко увидеть, как будет отображаться программа. Также, нажав на саму метку, можно изменить используемую в коде на Python переменную (вместе с другими свойствами):

В целом написанный нами скрипт на Python с небольшими изменениями можно использовать в M5Stack. Мы можем скопировать код на Python с локальной машины и вставить её во вкладку Python в IDE UI Flow.

В нашем коде найдём комментарий ### ЗДЕСЬ ДОЛЖЕН НАХОДИТЬСЯ UI ДЛЯ M5STACK ### и заменим всё, что ниже него, следующим кодом:

time_label = M5TextBox(146, 27, "", lcd.FONT_Default, 0xFFFFFF, rotate=0)
northbound_label = M5TextBox(146, 95, "", lcd.FONT_Default, 0xFFFFFF, rotate=0)
southbound_label = M5TextBox(146, 163, "", lcd.FONT_Default, 0xFFFFFF, rotate=0)

def print_train_arrivals(
        direction,
        label,
        time_until_train,
        nearest_arrival_time,
        second_arrival_time):
    if time_until_train <= 0:
        next_arrival_time = second_arrival_time
    else nearest_arrival_time:
        next_arrival_time_s = time.strftime(
            "%I:%M %p",
            time.localtime(next_arrival_time))
    label.setText(f"The next {direction} train will arrive at {next_arrival_time_s}")

while True:
    # Получаем текущее время, чтобы можно было узнать количество минут до прибытия
    current_time = int(time.time())
    time_until_northbound_train = int(
        ((nearest_northbound_arrival_time - current_time) / 60))
    time_until_southbound_train = int(
        ((nearest_southbound_arrival_time - current_time) / 60))
    current_time_s = time.strftime("%I:%M %p")
    time_label.setText(f"It's currently {current_time_s}")

    print_train_arrivals(
        "northbound",
        northbound_label,
        time_until_northbound_train,
        nearest_northbound_arrival_time,
        second_northbound_arrival_time)
    print_train_arrivals(
        "southbound",
        southbound_label,
        time_until_southbound_train,
        nearest_southbound_arrival_time,
        time_until_southbound_train)
 
    sleep 5

Основная часть этого кода выглядит знакомо! Чтобы он мог работать на M5Stack, мы внесли два важных изменения.

Во-первых, мы создали метки, которые будут шаблонами для данных о времени и поездах:

time_label
northbound_label
southbound_label

Во-вторых, мы поместили всё внутрь цикла while, который будет получать текущее время и устанавливать текст метки. Цикл будет ждать в течение пяти секунд, а затем перезапускать процесс.

Вот и всё! Нажав на кнопку Run, мы увидим, как строки обновляются каждые пять секунд и отображают новые данные о маршруте.

Заключение

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

Когда вы начнёте думать об этих вопросах уровня продакшена или захотите масштабировать проект на несколько устройств, вам также придётся продумать и работу с API. В этой статье я упоминал Gravitee Designer, который отлично подходит на этапе проектирования. У Gravitee есть другие инструменты для работы с API, например, для шлюза API, мониторинга и аналитики в реальном времени, а также развёртывания.

Процесс разработки IoT-приложений может показаться пугающим разработчикам, привыкшим писать код для традиционных серверов и веб-браузеров. Однако на самом деле для перехода к работе с IoT-устройствами нужно довольно мало. Современные устройства со встроенной поддержкой популярных языков и фреймворков, делают IoT интересным и инновационным способом сборки или интеграции с API и приложениями.

Конкурс статей от RUVDS.COM. Три денежные номинации. Главный приз — 100 000 рублей.

Создание IoT-приложения с использованием HTTP API