#!/usr/bin/python import calendar, re, sys import transitfeed from optparse import OptionParser DAY_LETTER = ['M', 'T', 'W', 'R', 'F', 'Sa', 'Su'] def GetRouteStops(route): route_stops = [] route_stoptimes = [] for trip in route.trips: stoptimes = trip.GetStopTimes() if trip.direction_id == '1': stoptimes.reverse() route_stoptimes.append(stoptimes) while len(route_stoptimes) > 0: for sts in route_stoptimes: st = sts.pop() # Exclude stops without passenger service. if (st.drop_off_type != '1' or st.pickup_type != '1') and \ not st.stop in route_stops: route_stops.append(st.stop) route_stoptimes = [ sts for sts in route_stoptimes if len(sts) > 0 ] route_stops.reverse() return route_stops def GetPeriodName(period): # Simplest cases: no days set is "Never" and one day set is that day name. day_count = len([ t for t in period.day_of_week if t ]) if day_count == 0: return 'Never' elif day_count == 1: for i, day in enumerate(period.day_of_week): if day: return calendar.day_name[i] # Check for contiguous day ranges. Use "Weekend" or "Weekday" if possible, # otherwise just the day endpoints. first_day = '' last_day = '' for i, day in enumerate(period.day_of_week): if day and first_day == '': first_day = calendar.day_abbr[i] elif not day and first_day != '' and last_day == '': last_day = calendar.day_abbr[i - 1] elif day and last_day != '': first_day = '' break if last_day == '': last_day = calendar.day_abbr[6] if first_day != '': if first_day == calendar.day_abbr[0] and last_day == calendar.day_abbr[4]: return 'Weekday' elif first_day == calendar.day_abbr[5] and last_day == calendar.day_abbr[6]: return 'Weekend' else: return first_day + '-' + last_day # Two-day lists get three-character abbreviations. if day_count == 2: result = '' for i, day in enumerate(period.day_of_week): if day and result == '': result = calendar.day_abbr[i] elif day: return result + ',' + calendar.day_abbr[i] # Three-day (or more) lists get letters. result = '' for i, day in enumerate(period.day_of_week): if day: result = result + DAY_LETTER[i] return result def ShortenStopName(name, max=20): name = re.sub(re.compile('\\b([nsew])/b\\b', re.I), '\\1b', name) name = re.sub(re.compile('\\b(rd|ct|ave|dr|cir|st|way|ln|pl|blvd|pkwy|hwy|ter|loop|trl|grn)\\.?\\s+', re.I), '', name) name = re.sub('\\s+', ' ', name) name = re.sub(' $', '', name) dirmatch = re.search('\\b([nsew]b)\\b', name, re.I) name = re.sub(re.compile('\\b([nsew]b)\\s*', re.I), '', name) if dirmatch: direction = dirmatch.group(1) return name[0:max-(len(direction)+1)] + '/' + direction else: return name[0:max] def PrintStopNames(stops, stream): stream.write('\t'.join([ ShortenStopName(s.stop_name) for s in stops ]) + '\n') def PartitionTrips(route): lines = {} for trip in route.trips: try: lines[trip.service_id] except KeyError: lines[trip.service_id] = {} try: lines[trip.service_id][trip.direction_id].append(trip) except KeyError: lines[trip.service_id][trip.direction_id] = [trip] return lines def PrintDays(periods, stream): stream.write('\t'.join([ GetPeriodName(p) for p in periods ]) + '\n') # Keys are separate to enforce ordering. def PrintLineNumbers(lines, keys, stream): nums = [] cur_line = 7 for k in keys: for d in ['0', '1']: nums.append(cur_line) cur_line += len(lines[k][d]) stream.write('\t'.join([ str(n) for n in nums]) + '\n') def HashTripByStopID(trip): times = {} for st in trip.GetStopTimes(): times[st.stop.stop_id] = (st.arrival_secs, st.departure_secs) return times def SecsToTime(secs): if not secs: return '-----' return '%d:%02d' % (secs / 3600, secs % 3600 / 60) def PrintTimes(lines, keys, stops, stream): for k in keys: for d in ['0', '1']: for t in lines[k][d]: times = HashTripByStopID(t) line_start = True first_record = True for s in stops: if not line_start: stream.write('\t') try: if first_record: stream.write(SecsToTime(times[s.stop_id][1])) first_record = False else: stream.write(SecsToTime(times[s.stop_id][0])) except KeyError: stream.write('-----') line_start = False stream.write('\n') def PrintCSV(schedule, route, stop_order, stream): if stop_order: stops = [ schedule.GetStop(s) for s in stop_order ] else: stops = GetRouteStops(route) lines = PartitionTrips(route) keys = lines.keys() keys.sort() PrintStopNames(stops, stream) PrintDays([ schedule.GetServicePeriod(k) for k in keys ], stream) PrintLineNumbers(lines, keys, stream) # Abbreviations, abbreviation definitions, timetable notes stream.write('\n\n\n') PrintTimes(lines, keys, stops, stream) def FindRoute(schedule, route_short_name): for route in schedule.GetRouteList(): if route.route_short_name == route_short_name: return route raise Exception('Route %s not found.' % (route_short_name)) def main(): parser = OptionParser(usage="Usage: %prog [options] GTFS-file route CSV-file") parser.add_option('--list-routes', dest='list_routes', action="store_true", default=False, help="List the routes in the GFTS file. The route and CSV file parameters are not needed.") parser.add_option('--list-stops', dest='list_stops', action="store_true", default=False, help="List the stops and their IDs for a given route. The CSV file parameter is not needed. Stops are printed in the same order they would appear in the CSV file.") parser.add_option('--stop-order', dest='stop_ids', help="Specify the order in which the stops will be listed in the CSV file as a comma-delimited list of stop IDs. Use --list-stops to get the IDs. WARNING: No verification is done on supplied stop IDs; if IDs are wrong, missing, or out of order, results are undefined.") (options, args) = parser.parse_args() loader = transitfeed.Loader(args[0]) schedule = loader.Load() if options.list_routes: for r in schedule.GetRouteList(): print r.route_short_name + ':', r.route_long_name, '-', r.route_desc exit(0) if options.list_stops: for s in GetRouteStops(FindRoute(schedule, args[1])): print s.stop_id + ':', s.stop_name, '-', s.stop_desc exit(0) stop_order = None if options.stop_ids: stop_order = options.stop_ids.split(',') PrintCSV(schedule, FindRoute(schedule, args[1]), stop_order, open(args[2], 'w')) if __name__ == '__main__': main()