使用 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 和應用程序。