My Train Orb (also Scraping Munich Train Schedules)

In an effort to optimize my commute, I wanted a glowing orb whose color indicated when I should start walking to the train stop. My train comes fairly frequently (every 10 minutes). The glowing orb lets me tell at a glance if I should be getting my stuff together, walking out the door, or running.

The glowing orb part is easy.
  • Arduino
  • Tri-color LED
  • Ping Pong ball
  • Mix with Python appropriately.
I'll post the code for the orb some other time. The hard part was actually getting the train schedule information. MVV (the mass transit authority in Munich) only provides PDFs for train schedules.



To parse those, I used pdftohtml. pdftohtml can generate XML which provides the absolute position of all the text elements on the schedule. Since they're in a grid, it wasn't too hard to parse that into something useful.
from BeautifulSoup import BeautifulSoup
import datetime
import re

_DIGITS_RE = re.compile(r'^\d+$')

# Schedule paramaters (page, hour column bounds, minute column bounds).
_MON_THUR_SCHEDULE = (1, (32, 38), (37, 311))
_FRI_SCHEDULE = (1, (32, 38), (311, 556))
_SAT_SCHEDULE = (1, (32, 38), (556, 802))
_SUN_SCHEDULE = (2, (41, 49), (48, 830))
_WEEK_SCHEDULE = {
0: _MON_THUR_SCHEDULE,
1: _MON_THUR_SCHEDULE,
2: _MON_THUR_SCHEDULE,
3: _MON_THUR_SCHEDULE,
4: _FRI_SCHEDULE,
5: _SAT_SCHEDULE,
6: _SUN_SCHEDULE,
}


class MvvSchedule(object):

def __init__(self):
schedule_xml = open('schedule.xml').read()
self.soup = BeautifulSoup(schedule_xml)

def _GetSchedule(self, page, hour_col, minutes_col):
soup = self.soup.find('page', number=str(page))

InRange = lambda bounds: lambda x: bounds[0] < int(x) < bounds[1]
hours = soup.findAll('text', left=InRange(hour_col))
# Get the row height for each hour to use as bounds.
hours.sort(key=lambda x: int(x.get('top')))
hour_bounds = [(h[0]['top'], h[1]['top']) for h in zip(hours, hours[1:])]
# We assume the bottom is a single row high.
hour_bounds.append(
(hours[-1]['top'], 2 * int(hours[-1]['top']) - int(hours[-2]['top'])))
hour_bounds = [(int(u) - 3, int(l) - 3) for u, l in hour_bounds]

schedule = {}
for hour, bounds in zip(hours, hour_bounds):
minutes = soup.findAll(
'text', top=InRange(bounds), left=InRange(minutes_col))
minutes = [int(m.string) for m in minutes
if m.string and _DIGITS_RE.match(m.string)]
schedule[int(hour.string)] = minutes

return schedule

def GetTimeToNextTrain(self):
now = datetime.datetime.now()
schedule_params = _WEEK_SCHEDULE[now.weekday()]
schedule = self._GetSchedule(*schedule_params)
times = [datetime.datetime(now.year, now.month, now.day, hour, minute)
for hour, minutes in schedule.iteritems() for minute in minutes]
times.sort()
stop_time = None
while times[-1] > now:
stop_time = times.pop()
if stop_time is not None:
return stop_time - now

if __name__ == '__main__':
mvv = MvvSchedule()
time_left = mvv.GetTimeToNextTrain()
if time_left is None:
print 'No more trains today.'
else:
print 'The next train arrives in %s.' % time_left

1 comment: