RideLogicAVLS/experimental/ridelogic_avls.py

#!/usr/bin/python3
#

# library to convert from JSON data to realtime GTFS
# protobuf objects
#

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 "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

###
###
###

def write_protobuf(ofn, feed_message_data):
  feedmsg = gtfs_feed_message(feed_message_data)
  fp = open(ofn, "wb")
  fp.write(feedmsg.SerializeToString())
  fp.close()


def example_usage():
  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()