![](https://habrastorage.org/webt/pv/yn/pi/pvynpiort-9vsb881emjwtq5bg4.jpeg)
Уже несколько лет не снижается ажиотаж вокруг IoT-устройств. Эти устройства могут быть почти чем угодно: от будильника, показывающего погоду, до холодильника, сообщающего о ценах в ближайших продуктовых магазинах. Какой бы ни была реализация, для общения с источниками данных эти устройства используют API. Но как конкретно подключаются сообщения, данные и устройства?
В этом посте мы покажем пример проектирования и моделирования данных для IoT-устройства. Для этого будет использовано M5Stack — небольшое модульное IoT-устройство с экраном, и подключение к API Metropolitan Transportation Authority Нью-Йорка (MTA) для получения актуального графика движения поездов на разных станциях.
![uiflow](https://habrastorage.org/webt/af/e1/2o/afe12ooi7ui3vqcxwdymozqxixk.jpeg)
Хотя мы будем работать с 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:
![](https://habrastorage.org/webt/qi/8w/yu/qi8wyuloyitg9wdkyskfor6_maw.png)
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 заключается в том, что они сильно упрощают визуализацию отображаемого на экране текста:
![](https://habrastorage.org/webt/9q/z_/pg/9qz_pgwlidpjnjcrmjyqjwzjzy8.gif)
В этом GIF я создал метку со стандартной строкой «Text» на примере экрана M5Stack. При переключении на Python мы видим, что эта метка является экземпляром объекта M5TextBox. При перетаскивании метки её координаты X и Y (первые два аргумента конструктора) меняются в Python. Благодаря этому, можно легко увидеть, как будет отображаться программа. Также, нажав на саму метку, можно изменить используемую в коде на Python переменную (вместе с другими свойствами):
![](https://habrastorage.org/webt/na/c_/ia/nac_iaouvfdvkzrbzcikvnl5z9s.gif)
В целом написанный нами скрипт на 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 рублей.
![](https://habrastorage.org/webt/sz/7j/pf/sz7jpfj8i1pa6ocj-eia09dev4q.png)