使用 python 在没有 protobuf 的情况下实时解析 GTFS
Parse GTFS realtime without protobuf with python
我正在尝试解析 GTFS
实时 trip_update
纯文本文件格式的数据,而不是 pb (protobuf) 格式。
(这里是提要url)
https://extranet.trainose.gr/epivatikos/transit/trip_updates
但是,我找到的唯一示例是处理 pb 文件的。
from google.transit import gtfs_realtime_pb2
....
response = requests.get(url, allow_redirects=True)
feed.ParseFromString(response.content)
for entity in feed.entity:
那么我该如何解析不是 pb 的提要呢?
谢谢。
Human-readable text 不是发送和接收 protobuf 消息的标准格式。 (如果你只想要 plain-text,你应该使用像 JSON 这样的标准文本格式。)原则上它仅用于调试目的。因此 Python Protobuf 库中没有用于解析 plain-text 消息的方法。这里正确的解决方案是找到一个实际的 protobuf 端点,也许通过与域所有者联系。 (编辑:显然 Python 库中实际上有一个 text-formatted 消息的 Parse
方法——参见源代码 here .)
也就是说,C++ Protobuf 库似乎包含直接解析文本格式的方法,因此如果您无法访问真正的 protobuf,这可能是一个备用选项:https://developers.google.com/protocol-buffers/docs/reference/cpp/google.protobuf.text_format。
据我所知,对于跨版本的文本格式的一致性没有严格的保证,但它在库中公开的事实表明它在实践中可能相当稳定。这个讨论给人同样的印象(因为有 Google-internal 解析文本格式的工具):https://github.com/protocolbuffers/protobuf/issues/1297#issuecomment-390825524.
事实证明,有一种方法可以像这样处理痛苦的文本提要:
response = requests.get(url, allow_redirects=True)
...
try:
from google.protobuf import text_format
text_format.Parse(response.content.decode('UTF-8'), feed, allow_unknown_extension=True)
print("Parse with text format successfully.")
printResults(feed)
except text_format.ParseError as e:
raise IOError("Cannot parse text %s." % (str(e)))
实际上这是我的整个脚本
from google.transit import gtfs_realtime_pb2
import os
import requests
def main():
feed = gtfs_realtime_pb2.FeedMessage()
url = ('https://feed.utl.com/feed')
get_feed(feed, url)
def printResults(feed):
from datetime import datetime
ts = int(str(feed.header.timestamp))
print("Last update: " + datetime.fromtimestamp(ts).strftime('%d-%m-%Y %H:%M:%S'))
for entity in feed.entity:
print (str(entity.trip_update.trip.trip_id)+';')
with open('output.txt', mode='w') as f:
for entity in feed.entity:
if entity.HasField('trip_update'):
f.write(str(entity.trip_update.trip.trip_id)+';')
def get_feed(feed, url):
proxies = {'http': '127.0.0.1:5555','https': '127.0.0.1:5555'}
response = requests.get(url, allow_redirects=True,proxies=proxies)
try:
feed.ParseFromString(response.content)
printResults(feed)
except :
print("Oops! That was no valid data. Try again...\n\n" + response.content)
try:
from google.protobuf import text_format
text_format.Parse(response.content.decode('UTF-8'), feed, allow_unknown_extension=True)
print("Parse with text format successfully.")
printResults(feed)
except text_format.ParseError as e:
raise IOError("Cannot parse file %s." % (str(e)))
if __name__ == "__main__":
main()
我正在尝试解析 GTFS
实时 trip_update
纯文本文件格式的数据,而不是 pb (protobuf) 格式。
(这里是提要url)
https://extranet.trainose.gr/epivatikos/transit/trip_updates
但是,我找到的唯一示例是处理 pb 文件的。
from google.transit import gtfs_realtime_pb2
....
response = requests.get(url, allow_redirects=True)
feed.ParseFromString(response.content)
for entity in feed.entity:
那么我该如何解析不是 pb 的提要呢? 谢谢。
Human-readable text 不是发送和接收 protobuf 消息的标准格式。 (如果你只想要 plain-text,你应该使用像 JSON 这样的标准文本格式。)原则上它仅用于调试目的。因此 Python Protobuf 库中没有用于解析 plain-text 消息的方法。这里正确的解决方案是找到一个实际的 protobuf 端点,也许通过与域所有者联系。 (编辑:显然 Python 库中实际上有一个 text-formatted 消息的 Parse
方法——参见源代码 here .)
也就是说,C++ Protobuf 库似乎包含直接解析文本格式的方法,因此如果您无法访问真正的 protobuf,这可能是一个备用选项:https://developers.google.com/protocol-buffers/docs/reference/cpp/google.protobuf.text_format。
据我所知,对于跨版本的文本格式的一致性没有严格的保证,但它在库中公开的事实表明它在实践中可能相当稳定。这个讨论给人同样的印象(因为有 Google-internal 解析文本格式的工具):https://github.com/protocolbuffers/protobuf/issues/1297#issuecomment-390825524.
事实证明,有一种方法可以像这样处理痛苦的文本提要:
response = requests.get(url, allow_redirects=True)
...
try:
from google.protobuf import text_format
text_format.Parse(response.content.decode('UTF-8'), feed, allow_unknown_extension=True)
print("Parse with text format successfully.")
printResults(feed)
except text_format.ParseError as e:
raise IOError("Cannot parse text %s." % (str(e)))
实际上这是我的整个脚本
from google.transit import gtfs_realtime_pb2
import os
import requests
def main():
feed = gtfs_realtime_pb2.FeedMessage()
url = ('https://feed.utl.com/feed')
get_feed(feed, url)
def printResults(feed):
from datetime import datetime
ts = int(str(feed.header.timestamp))
print("Last update: " + datetime.fromtimestamp(ts).strftime('%d-%m-%Y %H:%M:%S'))
for entity in feed.entity:
print (str(entity.trip_update.trip.trip_id)+';')
with open('output.txt', mode='w') as f:
for entity in feed.entity:
if entity.HasField('trip_update'):
f.write(str(entity.trip_update.trip.trip_id)+';')
def get_feed(feed, url):
proxies = {'http': '127.0.0.1:5555','https': '127.0.0.1:5555'}
response = requests.get(url, allow_redirects=True,proxies=proxies)
try:
feed.ParseFromString(response.content)
printResults(feed)
except :
print("Oops! That was no valid data. Try again...\n\n" + response.content)
try:
from google.protobuf import text_format
text_format.Parse(response.content.decode('UTF-8'), feed, allow_unknown_extension=True)
print("Parse with text format successfully.")
printResults(feed)
except text_format.ParseError as e:
raise IOError("Cannot parse file %s." % (str(e)))
if __name__ == "__main__":
main()