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

多年來(lái)，世界一直在關(guān)注物聯(lián)網(wǎng)設(shè)備。這些設(shè)備的范圍從顯示當(dāng)前天氣的鬧鐘到列出附近雜貨價(jià)格的冰箱。無(wú)論具體情況如何，這些設(shè)備都依賴API 與數(shù)據(jù)源進(jìn)行通信。但是，我們究竟如何連接消息、數(shù)據(jù)和設(shè)備呢？
在這篇文章中，我們將向您展示如何為物聯(lián)網(wǎng)設(shè)備設(shè)計(jì)和建模數(shù)據(jù)的示例。我們將使用M5Stack（一種帶有顯示屏的小型模塊化物聯(lián)網(wǎng)設(shè)備）并連接到紐約市大都會(huì)交通管理局(NYC MTA) 的 API，以呈現(xiàn)各個(gè)車站的最新地鐵時(shí)間。

雖然我們將專注于 M5Stack，但我們將討論的概念將適用于跨各種設(shè)備設(shè)計(jì) IoT 應(yīng)用程序。

所以讓我們開(kāi)始吧！

先決條件

在本教程中，我們將關(guān)注有關(guān)如何從 API 請(qǐng)求數(shù)據(jù)的更大概念。一些編程知識(shí)會(huì)很有幫助。雖然您不需要 M5Stack，但如果您確實(shí)有一個(gè)，那么您可以跟隨并將完成的項(xiàng)目上傳到您自己的設(shè)備上。
考慮到這一點(diǎn)，您可以下載VS Code IDE和M5Stack 插件。如果您以前從未啟動(dòng)過(guò) M5Stack，請(qǐng)按照他們的指南設(shè)置 wifi 和必要的固件。對(duì)于這個(gè)項(xiàng)目，我們將使用Python/ target=_blank class=infotextkey>Python 3，它是 M5Stack 使用的主要編程語(yǔ)言。

您需要注冊(cè)一個(gè) NYC MTA 開(kāi)發(fā)者帳戶以獲得免費(fèi)的開(kāi)發(fā)者 API 密鑰，以訪問(wèn)他們的實(shí)時(shí)地鐵數(shù)據(jù)。

最后，您應(yīng)該注冊(cè)一個(gè)免費(fèi)的 Gravitee 帳戶以使用API 設(shè)計(jì)器，這將使您更輕松地可視化和理解 API 調(diào)用中的數(shù)據(jù)流！

這個(gè)項(xiàng)目的源材料受到這個(gè)開(kāi)源項(xiàng)目的啟發(fā)，所以如果有幫助，請(qǐng)繼續(xù)為這個(gè)存儲(chǔ)庫(kù)加注星標(biāo)。

設(shè)計(jì) API 交互

在編寫一行代碼之前，讓我們退后一步，考慮一下我們需要什么樣的信息來(lái)完成這個(gè)項(xiàng)目：

• 相關(guān)地鐵站信息
• 哪些列車經(jīng)過(guò)這些車站
• 有關(guān)這些列車的最新實(shí)時(shí)數(shù)據(jù)

根據(jù)文檔，API 分為靜態(tài)數(shù)據(jù)饋送和實(shí)時(shí)數(shù)據(jù)饋送。

靜態(tài)數(shù)據(jù)饋送包含有關(guān)電臺(tái)的信息。有了這些信息，我們就可以從實(shí)時(shí)數(shù)據(jù)饋送 API 中獲取實(shí)際的實(shí)時(shí)列車數(shù)據(jù)。MTA 提供的數(shù)據(jù)采用以下 CSV 格式：

stop_id,stop_code,stop_name,stop_desc,stop_lat,stop_lon,zone_id,stop_url,location_type,parent_station

由于我們需要的唯一靜態(tài)信息是站點(diǎn) ID，我們可以簡(jiǎn)單地隨機(jī)抽取一個(gè)站點(diǎn) ID 并將其用于實(shí)時(shí)提要。在這種情況下，我選擇Hoyt–Schermerhorn 站是因?yàn)樗鄬?duì)復(fù)雜：兩列單獨(dú)的火車通過(guò)它（A 和 C）。車站也通過(guò)它們是北行 (N) 還是南行 (S) 來(lái)識(shí)別。

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) 來(lái)識(shí)別通過(guò)車站的火車，包括北行 (A42N) 和南行 (A42S)。

實(shí)時(shí)提要以google 的 GTFS 格式表示，該格式基于協(xié)議緩沖區(qū)（也稱為 protobuf）。雖然 NYC MTA 沒(méi)有記錄其特定提要的示例，但GTFS 有。從 GTFS 文檔中，我們可以確定如何以 protobuf 格式獲取特定車站的最新列車的到達(dá)時(shí)間。

下面是來(lái)自 GTFS 端點(diǎn)的響應(yīng)示例，已轉(zhuǎn)換為 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 對(duì) API 返回的內(nèi)容進(jìn)行建模、映射和可視化數(shù)據(jù)會(huì)非常有幫助。這是我們的 API Designer 思維導(dǎo)圖的快照：

API Designer 可幫助您識(shí)別 API 的所有資源（端點(diǎn)），以及與資源關(guān)聯(lián)的數(shù)據(jù)屬性。這些屬性將包括端點(diǎn)需要的輸入和它提供的輸出。

在我們的地圖中，我們有一個(gè)帶有路徑的資源/gtfs/。我們可以根據(jù)需要附加盡可能多的屬性，并且可以使用數(shù)據(jù)類型注釋每個(gè)屬性。通過(guò)查看我們的地圖，我們可以繪制從端點(diǎn)到右下角確定的到達(dá)和離開(kāi)時(shí)間的直接路徑。

因此，為了表示我們需要的數(shù)據(jù)，我們需要：

• 識(shí)別我們想要從中獲取火車信息的車站的 ID
• 針對(duì)我們感興趣的火車線路向 NYC MTA 的 GTFS 提要發(fā)出 HTTP 請(qǐng)求
• 遍歷結(jié)果，將響應(yīng)數(shù)組中的 stop_id 與我們的站 ID 進(jìn)行比較
• 然后，我們可以根據(jù)特定車站和火車的時(shí)間信息采取行動(dòng)

這代表了一些活動(dòng)部件，但它不應(yīng)該是我們無(wú)法處理的任何事情！

編碼它

在我們的 M5Stack 上運(yùn)行任何東西之前，讓我們首先確保我們的代碼在本地工作。我們將安裝一些 Python 包以使我們的項(xiàng)目更易于構(gòu)建。

殼

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

前兩個(gè)包將協(xié)議緩沖區(qū)轉(zhuǎn)換為 Python 字典（或哈希），這使得數(shù)據(jù)模型更易于使用。最后一個(gè)包使從 Python 發(fā)出 HTTP 請(qǐng)求變得更加容易。

我們將通過(guò)導(dǎo)入 Python 包來(lái)啟動(dòng)我們的程序：

Python

from google.transit import gtfs_realtime_pb2
import requests
import time

接下來(lái)，我們將向 NYC MTA GTFS 提要發(fā)出 HTTP 請(qǐng)求：

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)

到目前為止，一切都很好。我們?cè)谶@里使用的 GTFS 端點(diǎn)是用于 A/C/E 列車的端點(diǎn)，我們可以通過(guò)-aceURL 上的后綴來(lái)識(shí)別它。（除了這個(gè)演示，我們不關(guān)心 E 火車——對(duì)不起，E 火車！）

讓我們將 GTFS 協(xié)議緩沖區(qū)響應(yīng)轉(zhuǎn)換為字典：

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']

在這一點(diǎn)上，我強(qiáng)烈建議發(fā)布一個(gè)print(realtime_data)，這樣我們就可以看到實(shí)際的數(shù)據(jù)結(jié)構(gòu)是什么樣的。如果這是一個(gè)真實(shí)的項(xiàng)目，這樣的分析可能會(huì)幫助您確定字典中的哪些鍵和值需要迭代——但由于這是一個(gè)教程，我們已經(jīng)介紹了這一點(diǎn)。

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')

突然我們有很多代碼！但別擔(dān)心——我們正在做的事情并沒(méi)有那么復(fù)雜：

• 我們遍歷 A/C 線路的火車信息數(shù)組。
• 對(duì)于每個(gè)數(shù)組條目，我們驗(yàn)證我們是否擁有我們需要的所有鍵的值。這是防御性編碼，因?yàn)槲覀儾荒?100% 確定這個(gè)第三方服務(wù)在我們需要的時(shí)候有我們需要的東西！
• 之后，我們遍歷所有車站信息，并在我們到達(dá)我們需要的父 ID ( A42) 時(shí)停止北行和南行列車。
• 最后，我們將即將到來(lái)的火車到達(dá)時(shí)間列表保存在兩個(gè)單獨(dú)的全局變量中。

接下來(lái)，讓我們展示這些信息：

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)

我們上面所做的大部分工作都是數(shù)據(jù)格式化。關(guān)鍵步驟如下：

• 我們對(duì)車站北行和南行列車的到達(dá)時(shí)間進(jìn)行排序。
• 我們乘坐前兩次（“最快”的火車到達(dá)）。
• 我們將這些時(shí)間與當(dāng)前時(shí)間進(jìn)行比較，以獲得火車到達(dá)的距離（以分鐘為單位）。我們將這些火車到達(dá)時(shí)間傳遞給 print_train_arrivals。
• 如果下一班火車不到一分鐘就到了，我們將顯示第二班到站時(shí)間——恐怕你趕不上那班火車了！否則，我們將顯示最近的到達(dá)時(shí)間。

如果您在終端上運(yùn)行此腳本，您應(yīng)該會(huì)看到類似于以下內(nèi)容的消息：

殼

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

現(xiàn)在我們已經(jīng)在本地測(cè)試了我們的 Python 代碼可以與 NYC MTA API 通信，是時(shí)候讓這個(gè)代碼在我們的 M5Stack 上運(yùn)行了。對(duì) M5Stack 進(jìn)行編程的最簡(jiǎn)單方法是通過(guò)免費(fèi)的 UI Flow IDE，它只是一個(gè)通過(guò) WiFi 與您的設(shè)備通信的網(wǎng)頁(yè)。您可以通過(guò)他們的文檔了解有關(guān)如何配置設(shè)備以進(jìn)行 WiFi 訪問(wèn)的更多信息。

雖然 M5Stack 可以通過(guò) WYSIWYG UI 元素進(jìn)行編程，但它也可以接受（和運(yùn)行）Python 代碼。然而，WYSIWYG 元素的主要優(yōu)點(diǎn)是它使在屏幕上繪制的文本更容易可視化：

在這個(gè) GIF 中，我在示例 M5Stack 屏幕上創(chuàng)建了一個(gè)帶有默認(rèn)字符串“Text”的標(biāo)簽。當(dāng)我切換到 Python 時(shí)，我們看到標(biāo)簽是一個(gè)名為 M5TextBox 的對(duì)象的實(shí)例化。當(dāng)標(biāo)簽被拖動(dòng)時(shí)，它的 X 和 Y 坐標(biāo)（構(gòu)造函數(shù)中的前兩個(gè)參數(shù)）在 Python 中會(huì)發(fā)生變化。這樣可以很容易地看到您的程序?qū)⑷绾物@示。您還可以通過(guò)單擊標(biāo)簽本身來(lái)更改 Python 代碼中使用的變量（以及其他屬性）：

大多數(shù)情況下，我們編寫的 Python 腳本只需稍作修改即可在 M5Stack 上使用。我們可以從本地機(jī)器復(fù)制 Python 代碼并將其粘貼到 UI Flow IDE 的 Python 選項(xiàng)卡中。

在我們的代碼中，我們找到### UI FOR M5STACK SHOULD GO HERE ###注釋并將其下面的所有內(nèi)容替換為以下代碼：

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

其中大部分應(yīng)該看起來(lái)很熟悉！有兩個(gè)主要修改可以讓這個(gè)代碼在 M5Stack 上運(yùn)行。

首先，我們創(chuàng)建了作為時(shí)間和訓(xùn)練數(shù)據(jù)占位符的標(biāo)簽：

• time_label
• northbound_label
• southbound_label

其次，我們將所有內(nèi)容放在一個(gè)while循環(huán)中，它將獲取當(dāng)前時(shí)間并設(shè)置標(biāo)簽文本。循環(huán)將休眠五秒鐘，然后重新啟動(dòng)該過(guò)程。

就是這樣！當(dāng)我們點(diǎn)擊Run按鈕時(shí)，我們應(yīng)該看到我們的火車字符串每五秒更新一次，并使用最新的路線數(shù)據(jù)。

結(jié)論

就是這樣！物聯(lián)網(wǎng)設(shè)備經(jīng)常被業(yè)余愛(ài)好者使用，但如果你繼續(xù)從事這個(gè)項(xiàng)目，有幾個(gè)現(xiàn)實(shí)世界的考慮因素。一個(gè)考慮因素是速率限制，確保您以有效的方式從 MTA API 請(qǐng)求數(shù)據(jù)。另一個(gè)考慮因素是連接性。如果您的設(shè)備暫時(shí)無(wú)法訪問(wèn) WiFi，它將如何重新建立連接以獲取所需的信息？

一旦您開(kāi)始考慮這些生產(chǎn)級(jí)問(wèn)題，或者如果您想在多個(gè)設(shè)備上擴(kuò)展您的項(xiàng)目，您還需要考慮 API 管理。我在本文前面提到了 Gravitee Designer，這在設(shè)計(jì)階段非常有用。Gravitee 還有其他用于 API 管理的工具，例如 API 網(wǎng)關(guān)、監(jiān)控和實(shí)時(shí)分析、部署。

對(duì)于習(xí)慣于為傳統(tǒng)服務(wù)器和 Web 瀏覽器編寫代碼的開(kāi)發(fā)人員來(lái)說(shuō)，物聯(lián)網(wǎng)應(yīng)用程序開(kāi)發(fā)似乎令人生畏。然而，物聯(lián)網(wǎng)設(shè)備的飛躍實(shí)際上很小。今天的設(shè)備內(nèi)置了對(duì)流行語(yǔ)言和框架的支持，使物聯(lián)網(wǎng)成為一種有趣且創(chuàng)新的方式來(lái)構(gòu)建或集成 API 和應(yīng)用程序。