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

   1 # -*- coding: utf-8 -*-
   2
   3 from gotovienna.BeautifulSoup import BeautifulSoup
   4 from urllib2 import urlopen
   5 from datetime import time
   6 import re
   7 import collections
   8 from errors import LineNotFoundError, StationNotFoundError
   9
  10 from gotovienna import defaults
  11
  12 class ITipParser:
  13     def __init__(self):
  14         self._stations = {}
  15         self._lines = {}
  16
  17     def get_stations(self, name):
  18         """ Get station by direction
  19         {'Directionname': [('Station name', 'url')]}
  20         """
  21         if not self._stations.has_key(name):
  22             st = {}
  23
  24             if not self.lines.has_key(name):
  25                 return None
  26
  27             bs = BeautifulSoup(urlopen(self.lines[name]))
  28             tables = bs.findAll('table', {'class': 'text_10pix'})
  29             for i in range(2):
  30                 dir = tables[i].div.contents[-1].strip('&nbsp;')
  31
  32                 sta = []
  33                 for tr in tables[i].findAll('tr', {'onmouseout': 'obj_unhighlight(this);'}):
  34                     if tr.a:
  35                         sta.append((tr.a.text, defaults.line_overview + tr.a['href']))
  36                     else:
  37                         sta.append((tr.text.strip('&nbsp;'), None))
  38
  39                 st[dir] = sta
  40             self._stations[name] = st
  41
  42         return self._stations[name]
  43
  44     @property
  45     def lines(self):
  46         """ Dictionary of Line names with url as value
  47         """
  48         if not self._lines:
  49             bs = BeautifulSoup(urlopen(defaults.line_overview))
  50             # get tables
  51             lines = bs.findAll('td', {'class': 'linie'})
  52
  53             for line in lines:
  54                 if line.a:
  55                     href = defaults.line_overview + line.a['href']
  56                     if line.text:
  57                         self._lines[line.text] = href
  58                     elif line.img:
  59                         self._lines[line.img['alt']] = href
  60
  61         return self._lines
  62
  63     def get_url_from_direction(self, line, direction, station):
  64         stations = self.get_stations(line)
  65
  66         for stationname, url in stations.get(direction, []):
  67             if stationname == station:
  68                 return url
  69
  70         return None
  71
  72     def get_departures(self, url):
  73         """ Get list of next departures
  74         integer if time until next departure
  75         time if time of next departure
  76         """
  77
  78         #TODO parse line name and direction for station site parsing
  79
  80         if not url:
  81             # FIXME prevent from calling this method with None
  82             return []
  83
  84         bs = BeautifulSoup(urlopen(url))
  85         result_lines = bs.findAll('table')[-1].findAll('tr')
  86
  87         dep = []
  88         for tr in result_lines[1:]:
  89             th = tr.findAll('th')
  90             if len(th) < 2:
  91                 #TODO replace with logger
  92                 print "[DEBUG] Unable to find th in:\n%s" % str(tr)
  93                 continue
  94
  95             # parse time
  96             time = th[-2].text.split(' ')
  97             if len(time) < 2:
  98                 #print 'Invalid time: %s' % time
  99                 # TODO: Issue a warning OR convert "HH:MM" format to countdown
 100                 continue
 101
 102             time = time[1]
 103
 104             if time.find('rze...') >= 0:
 105                     dep.append(0)
 106             elif time.isdigit():
 107                 # if time to next departure in cell convert to int
 108                 dep.append(int(time))
 109             else:
 110                 # check if time of next departue in cell
 111                 t = time.strip('&nbsp;').split(':')
 112                 if len(t) == 2 and all(map(lambda x: x.isdigit(), t)):
 113                     t = map(int, t)
 114                     dep.append(time(*t))
 115                 else:
 116                     # Unexpected content
 117                     #TODO replace with logger
 118                     print "[DEBUG] Invalid data:\n%s" % time
 119
 120         return dep
 121
 122
 123 UBAHN, TRAM, BUS, NIGHTLINE, OTHER = range(5)
 124 LINE_TYPE_NAMES = ['U-Bahn', 'Strassenbahn', 'Bus', 'Nightline', 'Andere']
 125
 126 def get_line_sort_key(name):
 127     """Return a sort key for a line name
 128
 129     >>> get_line_sort_key('U6')
 130     ('U', 6)
 131
 132     >>> get_line_sort_key('D')
 133     ('D', 0)
 134
 135     >>> get_line_sort_key('59A')
 136     ('A', 59)
 137     """
 138     txt = ''.join(x for x in name if not x.isdigit())
 139     num = ''.join(x for x in name if x.isdigit()) or '0'
 140
 141     return (txt, int(num))
 142
 143 def get_line_type(name):
 144     """Get the type of line for the given name
 145
 146     >>> get_line_type('U1')
 147     UBAHN
 148     >>> get_line_type('59A')
 149     BUS
 150     """
 151     if name.isdigit():
 152         return TRAM
 153     elif name.endswith('A') or name.endswith('B') and name[1].isdigit():
 154         return BUS
 155     elif name.startswith('U'):
 156         return UBAHN
 157     elif name.startswith('N'):
 158         return NIGHTLINE
 159     elif name in ('D', 'O', 'VRT', 'WLB'):
 160         return TRAM
 161
 162     return OTHER
 163
 164 def categorize_lines(lines):
 165     """Return a categorized version of a list of line names
 166
 167     >>> categorize_lines(['U4', 'U3', '59A'])
 168     [('U-Bahn', ['U3', 'U4']), ('Bus', ['59A'])]
 169     """
 170     categorized_lines = collections.defaultdict(list)
 171
 172     for line in sorted(lines):
 173         line_type = get_line_type(line)
 174         categorized_lines[line_type].append(line)
 175
 176     for lines in categorized_lines.values():
 177         lines.sort(key=get_line_sort_key)
 178
 179     return [(LINE_TYPE_NAMES[key], categorized_lines[key])
 180             for key in sorted(categorized_lines)]
 181
 182
 183 class Line:
 184     def __init__(self, name):
 185         self._stations = None
 186         self.parser = ITipParser()
 187         if name.strip() in self.parser.lines():
 188             self.name = name.strip()
 189         else:
 190             raise LineNotFoundError('There is no line "%s"' % name.strip())
 191
 192     @property
 193     def stations(self):
 194         if not self._stations:
 195             self._stations = parser.get_stations(self.name)
 196         return self._stations
 197
 198     def get_departures(self, stationname):
 199         stationname = stationname.strip().lower()
 200         stations = self.stations
 201
 202         found = false
 203
 204         for direction in stations.keys():
 205             # filter stations starting with stationname
 206             stations[direction] = filter(lambda station: station[0].lower().starts_with(stationname), stations)
 207             found = found or bool(stations[direction])
 208
 209         if found:
 210             # TODO return departures
 211             raise NotImplementedError()
 212         else:
 213             raise StationNotFoundError('There is no stationname called "%s" at route of line "%s"' % (stationname, self.name))