如何通过NYC MTA Subway API获取特定站点特定方向列车到站时间

项目背景 #

我正在开发基于树莓派LED矩阵的项目，目标是显示公寓附近站点两趟特定列车的计划到站时间。已完成MTA开发者账号注册并接入API，但不清楚如何利用API数据查询特定线路、特定方向列车的计划到站时间（类似Apple Maps等应用展示的站点对应方向列车到站信息）。

当前使用API端点获取1-7号线数据，通过GTFS实时proto文件解析为对象，但遇到数据匹配问题：数据中无direction_id字段，计算出的到站时间与实际应用显示不符。

数据示例 #

vehicle {
  trip {
    trip_id: "096900_1..S03R"
    start_date: "20220815"
    route_id: "1"
  }
  current_stop_sequence: 37
  current_status: STOPPED_AT
  timestamp: 1660597518
  stop_id: "139S"
}
, id: "000005"
trip_update {
  trip {
    trip_id: "097200_1..N03R"
    start_date: "20220815"
    route_id: "1"
  }
  stop_time_update {
    arrival {
      time: 1660597411
    }
    departure {
      time: 1660597471
    }
    stop_id: "103N"
  }
  stop_time_update {
    arrival {
      time: 1660597531
    }
    stop_id: "101N"
  }
}

核心疑问 #

1. 如何判断列车的行驶方向？
2. 如何获取列车到特定站点的准确到站时间？
3. 当前方案是否可行？有没有更优解？

现有参考代码 #

subwaydata = gtfs_realtime_pb2.FeedMessage()
subwaydata.ParseFromString(response.content)

currentTime = subwaydata.header.timestamp

for entity in subwaydata.entity:
    if (entity.trip_update != None and entity.trip_update.trip.route_id == "ROUTE"):
        for stop_time_update in entity.trip_update.stop_time_update:
            if (stop_time_update.stop_id == "STATION"):
                print("Trip ID: " + entity.trip_update.trip.trip_id)
                print("Train: " + str(round((stop_time_update.arrival.time - currentTime) / 60)))
    # elif (entity.vehicle != None and entity.vehicle.trip.route_id == "ROUTE" and entity.vehicle.stop_id == "STATION"):
    #     print(entity.vehicle)
    elif (entity.trip_update != None and entity.trip_update.trip.route_id == "ROUTE"):
        for stop_time_update in entity.trip_update.stop_time_update:
            if (stop_time_update.stop_id == "STATION"):
                print("Trip ID: " + entity.trip_update.trip.trip_id)
                print("Train: " + str(round((stop_time_update.arrival.time - currentTime) / 60)))
    # elif (entity.vehicle != None and entity.vehicle.trip.route_id == "ROUTE" and entity.vehicle.stop_id == "STATION"):
    #     print(entity.vehicle)

print(subwaydata.header)

解决方案 #

一、判断列车行驶方向 #

MTA的GTFS实时数据通过trip_id和stop_id隐含方向信息，无需依赖direction_id：

• 从trip_id提取：示例中096900_1..S03R末尾的S代表南向（Southbound），N代表北向（Northbound）；多数线路用N/S/U/D（上城/下城）区分方向。
• 从stop_id验证：示例中139S、103N的末尾字母对应站台方向，可直接匹配目标方向。
• 结合静态GTFS数据：下载MTA静态GTFS数据，通过trips.txt建立trip_id与direction_id的映射，stops.txt可补充站点方向信息，进一步提升准确性。

二、获取准确到站时间 #

1. 优先使用trip_update数据：stop_time_update中的arrival.time是Unix时间戳，计算时注意：
   • 用API返回的subwaydata.header.timestamp（服务器时间）作为当前时间，避免本地时间同步误差。
   • 若arrival.time为空，可用departure.time替代；过滤掉arrival.time小于当前服务器时间的过期记录。
2. vehicle数据补全：当trip_update无目标站点记录时，可通过vehicle.current_stop_sequence结合静态GTFS的站点序列估算到站时间，但精度低于trip_update。

三、现有方案优化 #

原代码存在重复逻辑，可简化并增加方向过滤、数据有效性校验：

import gtfs_realtime_pb2

subwaydata = gtfs_realtime_pb2.FeedMessage()
subwaydata.ParseFromString(response.content)

current_time = subwaydata.header.timestamp
TARGET_ROUTE = "1"          # 目标线路
TARGET_STOP = "101N"        # 带方向的目标站点ID
TARGET_DIRECTION = "N"      # 目标方向（N/S/U/D）

arrival_list = []

for entity in subwaydata.entity:
    # 只处理trip_update类型的实体
    if not entity.HasField("trip_update"):
        continue
    trip_update = entity.trip_update
    # 过滤目标线路
    if trip_update.trip.route_id != TARGET_ROUTE:
        continue
    # 从trip_id提取方向
    trip_dir = trip_update.trip.trip_id.split(".")[-1][0]
    if trip_dir != TARGET_DIRECTION:
        continue
    # 筛选目标站点的有效到站记录
    for stop_time in trip_update.stop_time_update:
        if stop_time.stop_id != TARGET_STOP:
            continue
        if stop_time.arrival.time > current_time:
            mins_to_arrive = round((stop_time.arrival.time - current_time) / 60)
            arrival_list.append({
                "trip_id": trip_update.trip.trip_id,
                "minutes": mins_to_arrive
            })

# 按到站时间排序，取前2趟列车
arrival_list.sort(key=lambda x: x["minutes"])
for idx, train in enumerate(arrival_list[:2], 1):
    print(f"第{idx}趟列车({train['trip_id']})：{train['minutes']}分钟后到达")

print(f"数据更新时间：{current_time}")

额外建议 #

• 下载MTA静态GTFS数据，建立本地站点-线路-方向映射表，提升匹配准确性。
• 处理异常情况：比如arrival.time为空、字段缺失等，避免程序崩溃。
• 缓存API数据（每30秒请求一次），避免频繁调用触发限流，同时降低树莓派负载。

内容的提问来源于stack exchange，提问作者evvanerb