vcs.maemo.org Git - pywienerlinien/blob - gotovienna/routing.py

   1 #!/usr/bin/env python
   2 # -*- coding: UTF-8 -*-
   3
   4 from BeautifulSoup import BeautifulSoup, NavigableString
   5 from urllib2 import urlopen
   6 from urllib import urlencode
   7 from datetime import datetime, time
   8 from textwrap import wrap
   9 import argparse
  10 import sys
  11 import os.path
  12
  13 from gotovienna import defaults
  14
  15 POSITION_TYPES = ('stop', 'address', 'poi')
  16 TIMEFORMAT = '%H:%M'
  17 DEBUGLOG = os.path.expanduser('~/gotoVienna.debug')
  18
  19 class ParserError(Exception):
  20
  21     def __init__(self, msg='Parser error'):
  22         self.message = msg
  23
  24 class PageType:
  25     UNKNOWN, CORRECTION, RESULT = range(3)
  26
  27
  28 def guess_location_type(location):
  29     """Guess type (stop, address, poi) of a location
  30
  31     >>> guess_location_type('pilgramgasse')
  32     'stop'
  33
  34     >>> guess_location_type('karlsplatz 14')
  35     'address'
  36
  37     >>> guess_location_type('reumannplatz 12/34')
  38     'address'
  39     """
  40     parts = location.split()
  41     first_part = parts[0]
  42     last_part = parts[-1]
  43
  44     # Assume all single-word locations are stops
  45     if len(parts) == 1:
  46         return 'stop'
  47
  48     # If the last part is numeric, assume address
  49     if last_part.isdigit() and len(parts) > 1:
  50         return 'address'
  51
  52     # Addresses with door number (e.g. "12/34")
  53     if all(x.isdigit() or x == '/' for x in last_part):
  54         return 'address'
  55
  56     # Sane default - assume it's a stop/station name
  57     return 'stop'
  58
  59
  60 def search(origin_tuple, destination_tuple, dtime=None):
  61     """ build route request
  62     returns html result (as urllib response)
  63     """
  64     if not dtime:
  65         dtime = datetime.now()
  66
  67     origin, origin_type = origin_tuple
  68     destination, destination_type = destination_tuple
  69
  70     if origin_type is None:
  71         origin_type = guess_location_type(origin)
  72         print 'Guessed origin type:', origin_type
  73
  74     if destination_type is None:
  75         destination_type = guess_location_type(destination)
  76         print 'Guessed destination type:', destination_type
  77
  78     if (origin_type not in POSITION_TYPES or
  79             destination_type not in POSITION_TYPES):
  80         raise ParserError('Invalid position type')
  81
  82     post = defaults.search_post
  83     post['name_origin'] = origin
  84     post['type_origin'] = origin_type
  85     post['name_destination'] = destination
  86     post['type_destination'] = destination_type
  87     post['itdDateDayMonthYear'] = dtime.strftime('%d.%m.%Y')
  88     post['itdTime'] = dtime.strftime('%H:%M')
  89     params = urlencode(post)
  90     url = '%s?%s' % (defaults.action, params)
  91
  92     try:
  93         f = open(DEBUGLOG, 'a')
  94         f.write(url + '\n')
  95         f.close()
  96     except:
  97         print 'Unable to write to DEBUGLOG: %s' % DEBUGLOG
  98
  99     return urlopen(url)
 100
 101
 102 class sParser:
 103     """ Parser for search response
 104     """
 105
 106     def __init__(self, html):
 107         self.soup = BeautifulSoup(html)
 108
 109     def check_page(self):
 110         if self.soup.find('form', {'id': 'form_efaresults'}):
 111             return PageType.RESULT
 112
 113         if self.soup.find('div', {'class':'form_error'}):
 114             return PageType.CORRECTION
 115
 116         return PageType.UNKNOWN
 117
 118     state = property(check_page)
 119
 120     def get_correction(self):
 121         nlo = self.soup.find('select', {'id': 'nameList_origin'})
 122         nld = self.soup.find('select', {'id': 'nameList_destination'})
 123
 124         if not nlo and not nld:
 125             raise ParserError('Unable to parse html')
 126
 127         if nlo:
 128             origin = map(lambda x: x.text, nlo.findAll('option'))
 129         else:
 130             origin = []
 131         if nld:
 132             destination = map(lambda x: x.text, nld.findAll('option'))
 133         else:
 134             destination = []
 135
 136         return (origin, destination)
 137
 138     def get_result(self):
 139         return rParser(str(self.soup))
 140
 141
 142
 143 class rParser:
 144     """ Parser for routing results
 145     """
 146
 147     def __init__(self, html):
 148         self.soup = BeautifulSoup(html)
 149         self._overview = None
 150         self._details = None
 151
 152     @classmethod
 153     def get_tdtext(cls, x, cl):
 154             return x.find('td', {'class': cl}).text
 155
 156     @classmethod
 157     def get_change(cls, x):
 158         y = rParser.get_tdtext(x, 'col_change')
 159         if y:
 160             return int(y)
 161         else:
 162             return 0
 163
 164     @classmethod
 165     def get_price(cls, x):
 166         y = rParser.get_tdtext(x, 'col_price')
 167         if y == '*':
 168             return 0.0
 169         if y.find(','):
 170             return float(y.replace(',', '.'))
 171         else:
 172             return 0.0
 173
 174     @classmethod
 175     def get_date(cls, x):
 176         y = rParser.get_tdtext(x, 'col_date')
 177         if y:
 178             return datetime.strptime(y, '%d.%m.%Y').date()
 179         else:
 180             return None
 181
 182     @classmethod
 183     def get_time(cls, x):
 184         y = rParser.get_tdtext(x, 'col_time')
 185         if y:
 186             if (y.find("-") > 0):
 187                 return map(lambda z: time(*map(int, z.split(':'))), y.split('-'))
 188             else:
 189                 return map(lambda z: time(*map(int, z.split(':'))), wrap(y, 5))
 190         else:
 191             return []
 192
 193     @classmethod
 194     def get_duration(cls, x):
 195         y = rParser.get_tdtext(x, 'col_duration')
 196         if y:
 197             return time(*map(int, y.split(":")))
 198         else:
 199             return None
 200
 201     def __iter__(self):
 202         for detail in self.details():
 203             yield detail
 204
 205     def _parse_details(self):
 206         tours = self.soup.findAll('div', {'class': 'data_table tourdetail'})
 207
 208         trips = map(lambda x: map(lambda y: {
 209                         'time': rParser.get_time(y),
 210                         'station': map(lambda z: z[2:].strip(),
 211                                        filter(lambda x: type(x) == NavigableString, y.find('td', {'class': 'col_station'}).contents)), # filter non NaviStrings
 212                         'info': map(lambda x: x.strip(),
 213                                     filter(lambda z: type(z) == NavigableString, y.find('td', {'class': 'col_info'}).contents)),
 214                     }, x.find('tbody').findAll('tr')),
 215                     tours) # all routes
 216         return trips
 217
 218     @property
 219     def details(self):
 220         """returns list of trip details
 221         [ [ { 'time': [datetime.time, datetime.time] if time else [],
 222               'station': [u'start', u'end'] if station else [],
 223               'info': [u'start station' if station else u'details for walking', u'end station' if station else u'walking duration']
 224             }, ... # next trip step
 225           ], ... # next trip possibility
 226         ]
 227         """
 228         if not self._details:
 229             self._details = self._parse_details()
 230
 231         return self._details
 232
 233     def _parse_overview(self):
 234
 235         # get overview table
 236         table = self.soup.find('table', {'id': 'tbl_fahrten'})
 237
 238         # check if there is an overview table
 239         if table and table.findAll('tr'):
 240             # get rows
 241             rows = table.findAll('tr')[1:] # cut off headline
 242
 243             overview = map(lambda x: {
 244                                'date': rParser.get_date(x),
 245                                'time': rParser.get_time(x),
 246                                'duration': rParser.get_duration(x), # grab duration
 247                                'change': rParser.get_change(x),
 248                                'price': rParser.get_price(x),
 249                            },
 250                            rows)
 251         else:
 252             raise ParserError('Unable to parse overview')
 253
 254         return overview
 255
 256     @property
 257     def overview(self):
 258         """dict containing
 259         date: datetime
 260         time: [time, time]
 261         duration: time
 262         change: int
 263         price: float
 264         """
 265         if not self._overview:
 266             try:
 267                 self._overview = self._parse_overview()
 268             except AttributeError:
 269                 f = open(DEBUGLOG, 'w')
 270                 f.write(str(self.soup))
 271                 f.close()
 272
 273         return self._overview
 274