使用 M5Stack、 New York City MTA 的 API 和 Gravitee Designer。

多年来，世界一直在关注物联网设备。这些设备的范围从显示当前天气的闹钟到列出附近杂货价格的冰箱。无论具体情况如何，这些设备都依赖API 与数据源进行通信。但是，我们究竟如何连接消息、数据和设备呢？
在这篇文章中，我们将向您展示如何为物联网设备设计和建模数据的示例。我们将使用M5Stack（一种带有显示屏的小型模块化物联网设备）并连接到纽约市大都会交通管理局(NYC MTA) 的 API，以呈现各个车站的最新地铁时间。

虽然我们将专注于 M5Stack，但我们将讨论的概念将适用于跨各种设备设计 IoT 应用程序。

所以让我们开始吧！

先决条件

在本教程中，我们将关注有关如何从 API 请求数据的更大概念。一些编程知识会很有帮助。虽然您不需要 M5Stack，但如果您确实有一个，那么您可以跟随并将完成的项目上传到您自己的设备上。
考虑到这一点，您可以下载VS Code IDE和M5Stack 插件。如果您以前从未启动过 M5Stack，请按照他们的指南设置 wifi 和必要的固件。对于这个项目，我们将使用Python/ target=_blank class=infotextkey>Python 3，它是 M5Stack 使用的主要编程语言。

您需要注册一个 NYC MTA 开发者帐户以获得免费的开发者 API 密钥，以访问他们的实时地铁数据。

最后，您应该注册一个免费的 Gravitee 帐户以使用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

由于我们需要的唯一静态信息是站点 ID，我们可以简单地随机抽取一个站点 ID 并将其用于实时提要。在这种情况下，我选择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

从这些行中，我们只需要父站 ID (A42) 来识别通过车站的火车，包括北行 (A42N) 和南行 (A42S)。

实时提要以google 的 GTFS 格式表示，该格式基于协议缓冲区（也称为 protobuf）。虽然 NYC MTA 没有记录其特定提要的示例，但GTFS 有。从 GTFS 文档中，我们可以确定如何以 protobuf 格式获取特定车站的最新列车的到达时间。

下面是来自 GTFS 端点的响应示例，已转换为 JSON 以便于可视化：

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"
     }
  ]
}

由于 NYC MTA API 向您抛出的信息量很大，因此使用 Gravitee API Designer 对 API 返回的内容进行建模、映射和可视化数据会非常有帮助。这是我们的 API Designer 思维导图的快照：

API Designer 可帮助您识别 API 的所有资源（端点），以及与资源关联的数据属性。这些属性将包括端点需要的输入和它提供的输出。

在我们的地图中，我们有一个带有路径的资源/gtfs/。我们可以根据需要附加尽可能多的属性，并且可以使用数据类型注释每个属性。通过查看我们的地图，我们可以绘制从端点到右下角确定的到达和离开时间的直接路径。

因此，为了表示我们需要的数据，我们需要：

• 识别我们想要从中获取火车信息的车站的 ID
• 针对我们感兴趣的火车线路向 NYC MTA 的 GTFS 提要发出 HTTP 请求
• 遍历结果，将响应数组中的 stop_id 与我们的站 ID 进行比较
• 然后，我们可以根据特定车站和火车的时间信息采取行动

这代表了一些活动部件，但它不应该是我们无法处理的任何事情！

编码它

在我们的 M5Stack 上运行任何东西之前，让我们首先确保我们的代码在本地工作。我们将安装一些 Python 包以使我们的项目更易于构建。

壳

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

前两个包将协议缓冲区转换为 Python 字典（或哈希），这使得数据模型更易于使用。最后一个包使从 Python 发出 HTTP 请求变得更加容易。

我们将通过导入 Python 包来启动我们的程序：

Python

from google.transit import gtfs_realtime_pb2
import requests
import time

接下来，我们将向 NYC MTA GTFS 提要发出 HTTP 请求：

Python

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 列车的端点，我们可以通过-aceURL 上的后缀来识别它。（除了这个演示，我们不关心 E 火车——对不起，E 火车！）

让我们将 GTFS 协议缓冲区响应转换为字典：

Python

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)，这样我们就可以看到实际的数据结构是什么样的。如果这是一个真实的项目，这样的分析可能会帮助您确定字典中的哪些键和值需要迭代——但由于这是一个教程，我们已经介绍了这一点。

Python

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) 时停止北行和南行列车。
• 最后，我们将即将到来的火车到达时间列表保存在两个单独的全局变量中。

接下来，让我们展示这些信息：

Python

# 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 代码可以与 NYC MTA API 通信，是时候让这个代码在我们的 M5Stack 上运行了。对 M5Stack 进行编程的最简单方法是通过免费的 UI Flow IDE，它只是一个通过 WiFi 与您的设备通信的网页。您可以通过他们的文档了解有关如何配置设备以进行 WiFi 访问的更多信息。

虽然 M5Stack 可以通过 WYSIWYG UI 元素进行编程，但它也可以接受（和运行）Python 代码。然而，WYSIWYG 元素的主要优点是它使在屏幕上绘制的文本更容易可视化：

在这个 GIF 中，我在示例 M5Stack 屏幕上创建了一个带有默认字符串“Text”的标签。当我切换到 Python 时，我们看到标签是一个名为 M5TextBox 的对象的实例化。当标签被拖动时，它的 X 和 Y 坐标（构造函数中的前两个参数）在 Python 中会发生变化。这样可以很容易地看到您的程序将如何显示。您还可以通过单击标签本身来更改 Python 代码中使用的变量（以及其他属性）：

大多数情况下，我们编写的 Python 脚本只需稍作修改即可在 M5Stack 上使用。我们可以从本地机器复制 Python 代码并将其粘贴到 UI Flow IDE 的 Python 选项卡中。

在我们的代码中，我们找到### UI FOR M5STACK SHOULD GO HERE ###注释并将其下面的所有内容替换为以下代码：

Python

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_label
• northbound_label
• southbound_label

其次，我们将所有内容放在一个while循环中，它将获取当前时间并设置标签文本。循环将休眠五秒钟，然后重新启动该过程。

就是这样！当我们点击Run按钮时，我们应该看到我们的火车字符串每五秒更新一次，并使用最新的路线数据。

结论

就是这样！物联网设备经常被业余爱好者使用，但如果你继续从事这个项目，有几个现实世界的考虑因素。一个考虑因素是速率限制，确保您以有效的方式从 MTA API 请求数据。另一个考虑因素是连接性。如果您的设备暂时无法访问 WiFi，它将如何重新建立连接以获取所需的信息？

一旦您开始考虑这些生产级问题，或者如果您想在多个设备上扩展您的项目，您还需要考虑 API 管理。我在本文前面提到了 Gravitee Designer，这在设计阶段非常有用。Gravitee 还有其他用于 API 管理的工具，例如 API 网关、监控和实时分析、部署。

对于习惯于为传统服务器和 Web 浏览器编写代码的开发人员来说，物联网应用程序开发似乎令人生畏。然而，物联网设备的飞跃实际上很小。今天的设备内置了对流行语言和框架的支持，使物联网成为一种有趣且创新的方式来构建或集成 API 和应用程序。