RideLogicAVLS/experimental/gtfs_example_write.py

#!/usr/bin/python3
#

from google.transit import gtfs_realtime_pb2 as gtfspb
import urllib.request
import sys

def gtfs_vehicle_position(data, vp = None):
  if vp is None:
    vp = gtfspb.VehiclePosition()

  if "trip" in data:
    trip = data["trip"]

    if "trip_id" in trip:  vp.trip.trip_id = trip["trip_id"]
    if "route_id" in trip: vp.trip.route_id = trip["route_id"]
    if "direction_id" in trip: vp.trip.direction_id = trip["direction_id"]
    if "start_time" in trip: vp.trip.start_time = trip["start_time"]
    if "start_date" in trip: vp.trip.start_date = trip["start_date"]
    if "schedule_relationship" in trip: vp.trip.schedule_relationship = trip["schedule_relationship"]
    #vp.trip.schedule_relationship = gtfspb.TripDescriptor.ScheduleRelationship.SCHEDULED


  if "stop_id" in data: vp.stop_id = data["stop_id"]
  if "current_stop_sequence" in data: vp.current_stop_sequence = data["current_stop_sequence"]
  if "current_state" in data: vp.current_status = data["current_state"]
  #vp.current_status = gtfspb.VehiclePosition.VehicleStopStatus.INCOMING_AT
  if "timestamp" in data: vp.timestamp = data["timestamp"]
  if "congestion_level" in data: vp.congestion_level = data["congestion_level"]
  if "occupancy_status" in data: vp.occupancy_status = data["occupancy_status"]
  #vp.congestion_level = gtfspb.VehiclePosition.CongestionLevel.UNKNOWN_CONGESTION_LEVEL
  #vp.occupancy_status = gtfspb.VehiclePosition.OccupancyStatus.NO_DATA_AVAILABLE
  if "occupancy_percentage" in data: vp.occupancy_percentage = data["occupancy_percentage"]

  if "vehicle_id" in data: vp.vehicle.id = data["vehicle_id"]
  if "label" in data: vp.vehicle.label = data["label"]
  if "license_plate" in data: vp.vehicle.license_plate = data["license_plate"]

  if "position" in data:
    p = data["position"]
    if "latitude" in p: vp.position.latitude = p["latitude"]
    if "longitude" in p: vp.position.longitude = p["longitude"]
    if "bearing" in p: vp.position.bearing = p["bearing"]
    if "odometer" in p: vp.position.odometer = p["odometer"]
    if "speed" in p: vp.position.speed = p["speed"]
  return vp

def gtfs_alert():
  a = gtfspb.Alert()
  return a

def gtfs_trip_update():
  u = gtfspb.TripUpdate()
  return u

def gtfs_feed_entity(data, f = None):
  if f is None:
    f = gtfspb.FeedEntity()

  if "id" in data: f.id = data["id"]

  #if not ("#type" in data): return f
  #if data["#type"] == "VehiclePosition":
  if "vehicle" in data:
    gtfs_vehicle_position(data["vehicle"], f.vehicle)
  return f

def gtfs_feed_message(data, fm = None):
  if fm is None:
    fm = gtfspb.FeedMessage()

  if "header" in data:
    h = data["header"]
    if "gtfs_realtime_version" in h: fm.header.gtfs_realtime_version = h["gtfs_realtime_version"]
    if "timestamp" in h: fm.header.timestamp = h["timestamp"]

  if "entity" in data:
    for ent in data["entity"]:
      gtfs_feed_entity(ent, fm.entity.add())
  return fm

###
###
###

vehicle_position_data = {
  "trip" : {
    "trip_id": "testid4",
    "route_id": "testroute3",
    "direction_id": 0,
    "start_time": "teststartime...",
    "start_date": "teststartdate...",
    "schedule_relationship": gtfspb.TripDescriptor.ScheduleRelationship.SCHEDULED
  },

  "stop_id" : "teststop",
  "curent_stop_sequence": 3,
  "current_state" : gtfspb.VehiclePosition.VehicleStopStatus.INCOMING_AT,
  "timestamp" : 123,
  "congestion_level" : gtfspb.VehiclePosition.CongestionLevel.UNKNOWN_CONGESTION_LEVEL,
  "occupancy_status" : gtfspb.VehiclePosition.OccupancyStatus.NO_DATA_AVAILABLE,
  "occupancy_percentage" : 1,
  "vehicle_id" : "testvehicleid0",
  "label" : "testlabel",
  "license_plate" : "testlicenseplate",
  "position" : {
    "latitude" : 42.9,
    "longitude" : -78.1,
    "bearing" : 90.0,
    "odometer" : 1.0,
    "speed" : 0.1
  }
}

feed_entity_data = {
  "id" : "testfeedid",
  "#type": "VehiclePosition",
  "vehicle": vehicle_position_data
}

feed_message_data = {
  "header" : {
    "gtfs_realtime_version" : "2.0",
    "timestamp" : 0
  },
  "entity" : [
    feed_entity_data,
    feed_entity_data
  ]
}


###
###
###



vp = gtfs_vehicle_position(vehicle_position_data)

vp.stop_id = "ok"
print(">>>\n", vp)

fe = gtfs_feed_entity(feed_entity_data)
print("feed entity>>>\n", fe)

feedmsg = gtfs_feed_message(feed_message_data)
print("feed message>>>\n", feedmsg)


fp = open("outpy.pb", "wb")
fp.write(feedmsg.SerializeToString())
fp.close()