使用 python 在没有 protobuf 的情况下实时解析 GTFS-技术脚本

Parse GTFS realtime without protobuf with python

我正在尝试解析 GTFS 实时 trip_update 数据，该数据是纯文本文件格式，而不是 pb (protobuf) 格式。

(这里是提要网址)

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: 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...\

\

" + 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()

那么我该如何解析不是 pb 的提要呢？

谢谢。

事实证明，有一种方法可以用这样的方式处理疼痛文本提要：

from google.transit import gtfs_realtime_pb2

....

response = requests.get(url, allow_redirects=True)

feed.ParseFromString(response.content)

for entity in feed.entity: 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...\

\

" + 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()

其实这是我的整个脚本

from google.transit import gtfs_realtime_pb2

....

response = requests.get(url, allow_redirects=True)

feed.ParseFromString(response.content)

for entity in feed.entity: 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...\

\

" + 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()

人类可读的文本不是发送和接收 protobuf 消息的标准格式。 (如果你只想要纯文本，你应该使用像 JSON 这样的标准文本格式。)原则上它仅用于调试目的。因此，Python Protobuf 库中没有用于解析纯文本消息的方法。正确的解决方案是找到一个实际的 protobuf 端点，或许可以通过与域所有者取得联系。 (编辑：显然 Python 库中实际上有一个用于文本格式消息的 Parse 方法——请参见此处的源代码。)

也就是说，C Protobuf 库似乎包含直接解析文本格式的方法，所以如果您无法访问真正的 protobuf，这可能是一个备用选项：https://developers.google。 com/protocol-buffers/docs/reference/cpp/google.protobuf.text_format.

据我所知，对于跨版本的文本格式的一致性没有严格的保证，但它在库中公开的事实表明它在实践中可能相当稳定。这个讨论给出了同样的印象(因为有解析文本格式的谷歌内部工具)：https://github.com/protocolbuffers/protobuf/issues/1297#issuecomment-390825524。

使用 python 在没有 protobuf 的情况下实时解析 GTFS

Parse GTFS realtime without protobuf with python

相关推荐

Spring部署设置openshift

检查Java中正则表达式中模式的第n次出现

如何让 JTable 停留在已编辑的单元格上

Weblogic 12c 部署

Resteasy Content-Type 默认值

代码不会停止运行，在 Java 中

Out of memory java heap space

Log4j 记录到共享日志文件