时间戳

首页 > 解决方案 > 无法使用 Python gtfs_realtime_pb2 模块获取服务警报 Protobuff 以包含 header_text 或 description_text

service - 无法使用 Python gtfs_realtime_pb2 模块获取服务警报 Protobuff 以包含 header_text 或 description_text

问题描述

我们在将 header_text 和 description_text 添加到服务警报 protobuff 文件时遇到了困难。我们正在尝试匹配此页面上显示的示例。

https://developers.google.com/transit/gtfs-realtime/examples/alerts

我们的数据从以下字典开始：

alerts_dict = {
"header": {
    "gtfs_realtime_version": "1",
    "timestamp": "1543318671",
    "incrementality": "FULL_DATASET"
},
"entity": [{
    "497": {
        "active_period": [{
            "start": 1525320000,
            "end": 1546315200
        }],
        "url": "http://www.capmetro.org/planner",
        "effect": 4,
        "header_text": "South 183: Airport",
        "informed_entity": [{
            "route_type": "3",
            "route_id": "17",
            "trip": "",
            "stop_id": "3304"
        }, {
            "route_type": "3",
            "route_id": "350",
            "trip": "",
            "stop_id": "3304"
        }],
        "description_text": "Stop closed temporarily",
        "cause": 2
    },
    "460": {
        "active_period": [{
            "start": 1519876800,
            "end": 1546315200
        }],
        "url": "http://www.capmetro.org/planner",
        "effect": 4,
        "header_text": "Ave F / Duval Detour",
        "informed_entity": [{
            "route_type": "3",
            "route_id": "7",
            "trip": "",
            "stop_id": "1167"
        }, {
            "route_type": "3",
            "route_id": "7",
            "trip": "",
            "stop_id": "1268"
        }],
        "description_text": "Stop closed temporarily",
        "cause": 2
    }
}]

}

我们的Python代码如下：

newfeed = gtfs_realtime_pb2.FeedMessage()
newfeedheader = newfeed.header
newfeedheader.gtfs_realtime_version = '2.0'

for alert_id, alert_dict in alerts_dict["entity"][0].iteritems():
    print(alert_id)
    print(alert_dict)


    newentity = newfeed.entity.add()
    newalert = newentity.alert
    newentity.id = str(alert_id)

    newtimerange = newalert.active_period.add()
    newtimerange.end = alert_dict['active_period'][0]['end']
    newtimerange.start = alert_dict['active_period'][0]['start']

    for informed in alert_dict['informed_entity']:
        newentityselector = newalert.informed_entity.add()
        newentityselector.route_id = informed['route_id']
        newentityselector.route_type = int(informed['route_type'])
        newentityselector.stop_id = informed['stop_id']

    print(alert_dict['description_text'])
    newdescription = newalert.header_text

    newdescription = alert_dict['description_text']

    newalert.cause = alert_dict['cause']
    newalert.effect = alert_dict['effect']



pb_feed = newfeed.SerializeToString()
with open("servicealerts.pb", 'wb') as fout:
  fout.write(pb_feed)

令人沮丧的部分是我们没有收到任何类型的错误消息。一切似乎都运行正常，但生成的 pb 文件不包含新的 header_text 或 description_text 项。

我们可以使用以下代码读取 pb 文件：

feed = gtfs_realtime_pb2.FeedMessage()
response = open("servicealerts.pb")
feed.ParseFromString(response.read())
print(feed)

我们非常感谢任何人可以提供的任何帮助，为我们指明正确的方向来解决这个问题。

标签： servicereal-timealerts

解决方案

我能够找到答案。这个Python Notebook表明，通过正确格式化字典，可以用几行代码生成 PB。

from google.transit import gtfs_realtime_pb2
from google.protobuf.json_format import MessageToDict

newfeed = gtfs_realtime_pb2.FeedMessage()
ParseDict(alerts_dict, newfeed)

pb_feed = newfeed.SerializeToString()
with open("servicealerts.pb", 'wb') as fout:
    fout.write(pb_feed)

我所要做的就是正确地按字典格式化。

    if ALERT_GROUP_ID not in entity_dict.keys():

        entity_dict[ALERT_GROUP_ID] = {"id": ALERT_GROUP_ID,
            "alert":{
                "active_period": [{
                    "start": int(START_TIME),
                    "end": int(END_TIME)
                }],
                "cause": cause_dict.get(CAUSE, ""),
                "effect": effect_dict.get(EFFECT),
                "url": {
                    "translation": [{
                        "text": URL,
                        "language": "en"
                    }]
                },
                "header_text": {
                    "translation": [{
                        "text": HEADER_TEXT,
                        "language": "en"
                    }]
                },
                "informed_entity": [{
                    'route_id': ROUTE_ID,
                    'route_type': ROUTE_TYPE,
                    'trip': TRIP,
                    'stop_id': STOP_ID
                }],
                "description_text": {
                    "translation": [{
                    "text": "Stop closed temporarily",
                    "language": "en"
                    }]
                },
            },
        }
        # print(entity_dict[ALERT_GROUP_ID]["alert"]['informed_entity'])
    else:

        entity_dict[ALERT_GROUP_ID]["alert"]['informed_entity'].append({
            'route_id': ROUTE_ID,
            'route_type': ROUTE_TYPE,
            'trip': TRIP,
            'stop_id': STOP_ID
            })

推荐阅读