В течение многих лет мир был наполнен IoT-устройствами. Эти устройства варьируются от будильников, показывающих текущую погоду, до холодильников, указывающих цены на ближайшие продукты. Независимо от специфики, эти устройства используют API для связи с источниками данных. Но как именно мы связываем сообщения, данные и устройства?
В этом посте мы покажем вам пример того, как разработать и смоделировать данные для IoT-устройства. Мы будем использовать M5Stack — небольшое модульное IoT-устройство с экраном — и подключимся к API Управления городского транспорта Нью-Йорка (NYC MTA) для отображения последнего времени в метро для различных станций.
Хотя мы сосредоточимся на M5Stack, концепции, которые мы обсудим, применимы для разработки IoT-приложений для широкого спектра устройств.
Итак, давайте приступим!
Предварительные условия
В этом руководстве мы сосредоточимся на более крупных концептуальных идеях, связанных с тем, как запрашивать данные из API. Некоторые знания программирования будут очень полезны. Хотя вам не нужен M5Stack, если он у вас есть, вы можете следовать за нами и загрузить готовый проект на свое собственное устройство.
Для этого вы можете скачать IDE VS Code и плагин M5Stack. Если вы никогда раньше не загружали M5Stack, следуйте их руководству по настройке WiFi и необходимой прошивки. Для этого проекта мы будем использовать Python 3, который является основным языком программирования, используемым M5Stack.
Вам нужно будет зарегистрироваться в аккаунте разработчика NYC MTA для получения бесплатного API-ключа разработчика, чтобы получить доступ к данным метро в режиме реального времени.
Наконец, вам следует зарегистрировать бесплатный аккаунт Gravitee для использования конструктора API, который облегчит визуализацию и понимание потока данных в ваших вызовах API!
Исходный материал для этого проекта был вдохновлен этим проектом с открытым исходным кодом, так что зайдите в этот репозиторий, если он был вам полезен.
Проектирование взаимодействия с 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
Поскольку единственная статическая информация, которая нам нужна, — это идентификатор станции, мы можем просто выбрать произвольный идентификатор станции и использовать его для получения данных в реальном времени. В данном случае я выбрал станцию Hoyt-Schermerhorn из-за ее относительной сложности: через нее проходят два отдельных поезда (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
Из этих строк нам нужен только идентификатор родительской остановки (A42), чтобы определить поезда, проходящие через станцию, как северного направления (A42N), так и южного (A42S).
Каналы реального времени представлены в формате Google GTFS, который основан на протокольных буферах (также называемых protobuf). В то время как NYC 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"
},
//…more stops…
{
"arrival":{
"time":1654044957
},
"departure":{
"time":1654044957
},
"stop_id":"A42N"
}
]
}
Поскольку API NYC MTA выдает огромное количество информации, может быть очень полезно использовать Gravitee API Designer для моделирования того, что возвращает API, отображения и визуализации данных. Вот снимок нашей ментальной карты API Designer:
API Designer поможет вам определить все ресурсы (конечные точки) вашего API, а также атрибуты данных, связанные с каждым ресурсом. Эти атрибуты включают входные данные, необходимые конечной точке, и выходные данные, которые она предоставляет.
В нашей карте у нас есть ресурс с путем /gtfs/. Мы можем подключить столько атрибутов, сколько необходимо, и аннотировать каждый из этих атрибутов типами данных. Взглянув на нашу карту, мы можем нарисовать прямой путь от конечной точки до времени прибытия и отправления, указанного в правом нижнем углу.
Итак, чтобы представить нужные нам данные, нам необходимо:
- Определить идентификатор станции, с которой мы хотим получить информацию о поездах.
- Выполнить HTTP-запрос к каналу GTFS NYC 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-запрос к каналу NYC MTA GTFS:
api_key = "YOUR_API_KEY"
# Requests subway status data feed from the NYC MTA API
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 — простите, поезд E!)
Давайте преобразуем буферный ответ протокола GTFS в словарь:
from protobuf_to_dict import protobuf_to_dict
subway_feed = protobuf_to_dict(feed) # converts MTA data feed to a dictionary
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)
# Keep a global list to collect various train times
northbound_times = []
southbound_times = []
# Run the above function for the station ID for Hoyt-Schermerhorn
station_time_lookup(realtime_data, 'A42')
Внезапно у нас появилось много кода! Но не волнуйтесь — то, что мы делаем, не так уж сложно:
- Мы выполняем итерации по массиву информации о поездах для линий A/C.
- Для каждой записи массива мы проверяем, что у нас есть значения для всех нужных нам ключей. Это защитное кодирование, потому что мы не можем быть на 100% уверены, что у стороннего сервиса есть то, что нам нужно, когда нам это нужно!
- После этого мы перебираем всю информацию о станциях и останавливаемся, когда останавливаемся на нужном нам ID родителя (
A42) для поездов северного и южного направления. - Наконец, мы сохраняем списки времени прибытия предстоящих поездов в двух отдельных глобальных переменных.
Далее, давайте представим эту информацию:
# Sort collected times in chronological order
northbound_times.sort()
southbound_times.sort()
# Pop off the earliest and second earliest arrival times from the list
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 FOR M5STACK SHOULD GO HERE ###
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}")
# Grab the current time so that you can find out the minutes to arrival
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 NYC MTA, пришло время запустить этот код на нашем M5Stack. Самый простой способ программирования M5Stack — через бесплатную среду разработки UI Flow IDE, которая представляет собой веб-страницу, взаимодействующую с вашим устройством через WiFi. Вы можете узнать больше о том, как настроить ваше устройство для доступа по WiFi в их документации.
Хотя M5Stack можно программировать с помощью элементов пользовательского интерфейса WYSIWYG, он также может принимать (и выполнять) код Python. Однако главное преимущество элементов WYSIWYG в том, что они значительно упрощают визуализацию текста, нарисованного на экране:
В этом GIF на примере экрана M5Stack я создал метку со строкой по умолчанию «Text». Когда я переключаюсь на Python, мы видим, что ярлык является инстанцией объекта под названием M5TextBox. При перетаскивании ярлыка его координаты X и Y (первые два аргумента в конструкторе) изменяются в Python. Это позволяет легко увидеть, как будет отображаться ваша программа. Вы также можете изменить переменную, используемую в коде Python (а также другие свойства), щелкнув по самой метке:
По большей части, написанный нами сценарий Python можно использовать на M5Stack с небольшими изменениями. Мы можем скопировать код Python с нашей локальной машины и вставить его во вкладку Python в UI Flow IDE.
В нашем коде мы найдем комментарий ## UI FOR M5STACK SHOULD GOERE ### и заменим все, что находится под ним, следующим кодом:
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:
# Grab the current time so that you can find out the minutes to arrival
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_labelnorthbound_labelsouthbound_label
Во-вторых, мы поместили все внутри цикла while, который будет получать текущее время и устанавливать текст меток. Цикл будет спать в течение пяти секунд, а затем перезапустит процесс.
Вот и все! Когда мы нажмем кнопку Run, мы увидим, что наши строки поездов обновляются каждые пять секунд с последними данными о маршруте.
Заключение
Вот и все! IoT-устройства часто используются любителями, но если вы продолжите работу над этим проектом, есть несколько реальных соображений. Одно из соображений — ограничение скорости, чтобы убедиться, что вы запрашиваете данные у API MTA эффективным способом. Другое соображение — подключение. Если ваше устройство временно потеряет доступ к WiFi, как оно восстановит соединение, чтобы получить необходимую информацию?
Как только вы начнете думать об этих проблемах производственного уровня, или если вы хотите масштабировать свой проект на множество устройств, вам также нужно будет подумать об управлении API. Ранее в этой статье я упоминал Gravitee Designer, который отлично подходит для этапа проектирования. В Gravitee есть и другие инструменты для управления API, такие как шлюз API, мониторинг и аналитика в реальном времени, развертывание.
Разработка приложений IoT может показаться сложной задачей для разработчиков, которые привыкли писать код для традиционных серверов и веб-браузеров. Однако на самом деле скачок к IoT-устройствам совсем небольшой. Современные устройства со встроенной поддержкой популярных языков и фреймворков делают IoT увлекательным и инновационным способом создания или интеграции с API и приложениями.