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

Posted 2023-03-13

【中文标题】无法使用 Python gtfs_realtime_pb2 模块获取服务警报 Protobuff 以包含 header_text 或 description_text

【英文标题】：Can't get Service Alerts Protobuff to include header_text or description_text using Python gtfs_realtime_pb2 module

【发布时间】：2019-01-25 15:43:16

【问题描述】：

我们在将 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)

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

【参考方案1】：

我找到了答案。这个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
            )