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

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

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

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

Заключение