# The MIT License # # Copyright (c) 2008 StopFinder Company # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. """ This module contains: - Model classes that represent fundamental StopFinder datatypes - A client class that abstracts the process of sending queries to the StopFinder API. Author: Michael DiBernardo (debo@stopfinder.com) """ import datetime import urllib import xml.dom.minidom #---------------------------------------------------------------------- # Client data types. #---------------------------------------------------------------------- class SFLocation(object): """ A location in the StopFinder directory. This class is intended to be used solely for extension, and should not be instantiated directly. """ def __init__(self, name, address, latitude, longitude): """ Constructor for SFLocation. """ self._name = name self._address = address self._latitude = latitude self._longitude = longitude def name(self): """ Get the name of this location. """ return self._name def address(self): """ Get the address of this location. """ return self._address def latitude(self): """ Get the latitude of this location. """ return self._latitude def longitude(self): """ Get the longitude of this location. """ return self._longitude class SFParkingLot(SFLocation): """ A parking lot in the StopFinder directory. """ def __init__(self, name, address, latitude, longitude, owner, capacity, is_garage, source, is_cheapest, is_closest): """ Instantiate a new SFParkingLot object. """ SFLocation.__init__(self, name, address, latitude, longitude) self._owner = owner self._capacity = capacity self._is_garage = is_garage self._source = source self._is_cheapest = is_cheapest self._is_closest = is_closest self._rates = list() class Rate(object): """ A rate that is applied to someone parking in a particular lot, on a particular range of days, within a particular range of times. Note that we consider a lot to be closed or inaccessible during any range of time for which we have no rate information. """ def __init__(self, day_type, starts, ends, price, price_scheme, daily_max, time_limit): self._day_type = day_type self._starts = starts self._ends = ends self._price = price self._price_scheme = price_scheme self._daily_max = daily_max self._time_limit = time_limit def day_type(self): """ Get the range of days during which this rate is applicable. Can be one of the following values: Weekday (Mon - Fri) Weekend (Sat and Sun) Everyday (Mon - Sun) Holiday (currently unused) Event (currently unused) Monday Tuesday ... and so on, for the rest of the days of the week. """ return self._day_type def starts(self): """ The time at which this rate takes effect. """ return self._starts def ends(self): """ The time at which this rate ceases to hold. """ return self._ends def price(self): """ The price paid per unit of time for this lot, in Canadian dollars (e.g. 6.50 = $6.50 CAD). """ return self._price def price_scheme(self): """ Whether self.price() is the amount paid per hour ("Hourly") or a flat rate ("Flat"). """ return self._price_scheme def daily_max(self): """ Returns the maximum one can expect to pay within this rate period. May be None. """ return self._daily_max def time_limit(self): """ Returns the maximum time IN MINUTES for which one is allowed to park during this rate period. May be None. """ return self._time_limit def owner(self): """ Get the owner of this lot. May be None. """ return self._owner def capacity(self): """ Get the capacity of this lot. May be None. """ return self._capacity def source(self): """ Get the source of this lot (e.g. the link to the website from which we inferred the rates.) This is not guaranteed to be an URL, so beware. """ return self._source def is_garage(self): """ Is this location an indoor garage? """ return self._is_garage def is_cheapest(self): """ Is this location the cheapest in the query? """ return self._is_cheapest def is_closest(self): """ Is this location the closest in the query? """ return self._is_closest def rates(self): """ Get the list of applicable rates for this lot. This may be empty, but never None. """ return self._rates def add_rate(self, rate): """ Add a rate to the list of rates for this lot. """ assert(type(rate) is SFParkingLot.Rate) self._rates.append(rate) class SFSubway(SFLocation): """ A subway station in the StopFinder directory. """ def __init__(self, name, address, latitude, longitude, line): SFLocation.__init__(self, name, address, latitude, longitude) self._line = line def line(self): """ Get the line on which this subway station lies. """ return self._line class SFBusStop(SFLocation): """ A TTC bus stop in the StopFinder directory. """ def __init__(self, name, address, latitude, longitude): SFLocation.__init__(self, name, address, latitude, longitude) self._route_stops = list() class RouteStop(object): """ A record of a bus route that passes by this stop. You can think of this as the little chart of times for a particular route that you would see posted at a specific bus stop. """ def __init__(self, name, number, link_to_schedule): self._name = name self._number = number self._link_to_schedule = link_to_schedule def name(self): """ The name of this route on the TTC website. """ return self._name def number(self): """ The route number. This is still a string, to accomodate for route numbers that may have other characters in them (e.g. 25A). """ return self._number def link_to_schedule(self): """ An url to the schedule of times when buses on this route are supposed to stop at this bus stop. """ return self._link_to_schedule def route_stops(self): """ Get the list of routes that pass by this lot. """ return self._route_stops def add_route_stop(self, route_stop): """ Add to the list of routes that pass by this lot. """ assert(type(route_stop) is SFBusStop.RouteStop) self._route_stops.append(route_stop) #---------------------------------------------------------------------- # Helper functions for DOM parsing. #---------------------------------------------------------------------- def get_element_text(element, converter=str): """ Given the result of a getElementsByTagName call that is expected to return 0 or 1 elements, this function will: - Return None if there were 0 elements - Retrieve the text of the single tag if there was a single element - Clean up the text. - Apply the optional conversion function to the text to pretty it up before returning it. """ if not element: return None assert len(element) == 1 # Value was empty i.e. if not element[0].firstChild: return None tag_text = element[0].firstChild.nodeValue tag_text = tag_text.strip() return converter(tag_text) def textbool(text): """ Converts string "True" to True and "False" to False. """ if text == "True": return True elif text == "False": return False else: assert False #---------------------------------------------------------------------- # The API client. #---------------------------------------------------------------------- class StopFinderAPIClient(object): """ Client that talks to the StopFinder API. The main method that you should be interested in here is find_stops, which issues queries. The rest of the public methods allow you to directly parse StopFinder XML, if that's something that you find yourself needing to do. """ def __init__(self, sfapi_url="http://stopfinder.com/sfapi/"): """ Create an instance of the StopFinder API client that will talk to sfapi_url to make requests. """ self._sfapi_url = sfapi_url def find_stops(self, latitude, longitude, include_lots=True, include_subways=True, include_busstops=True, arrival_day="", arrival_time="", staying=""): """ Query the StopFinder API with the given parameters. See http://stopfinder.com/page/api/ for a description of acceptable values for these parameters, and for the expected behaviour if you omit some of them. This method will not behave well if you feed it parameters that have unacceptable values, so it's best to wrap calls to this method in a try/except block. This method returns a single list that contains _all_ of the results. If you want to figure out which elements are parking lots, which are subways, and which are bus stops, you have to distinguish between them by using the type() builtin function. For example: client = StopFinderAPIClient() stops = client.find_stops(...) lot_filter = lambda stop: type(stop) is SFParkingLot lots = filter(lot_filter, stops) """ query_params = { 'latitude' : latitude, 'longitude' : longitude, 'include_lots' : include_lots, 'include_subways' : include_subways, 'include_busstops' : include_busstops, 'arrival_day' : arrival_day, 'arrival_time' : arrival_time, 'staying' : staying, } # URLify the query parameters. post_data = urllib.urlencode(query_params) # Grab a handle to the HTTP response. url_handle = urllib.urlopen(self._sfapi_url, post_data) # Parse away. return self.parse_sf_api_xml_from_handle(url_handle) def parse_sf_api_xml_from_handle(self, handle): """ Parses the XML result of a StopFinder API query. handle is expected to be a file-like object that streams the XML result of a StopFinder API query (such as the handles returned by urllib.urlopen or urllib2.urlopen). This method returns a single list that contains _all_ of the results. For advice on how to distinguish between the different types of stops in the list, see the find_stops documentation. """ doc = xml.dom.minidom.parse(handle) return self._parse_sf_api_xml_from_dom_doc(doc) def parse_sf_api_xml_from_string(self, xml_string): """ Parses the XML result of a StopFinder API query. xml_string is expected to be a string that contains the complete XML result of a StopFinder API query. This method returns a single list that contains _all_ of the results. For advice on how to distinguish between the different types of stops in the list, see the find_stops documentation. """ doc = xml.dom.minidom.parseString(xml_string) return self._parse_sf_api_xml_from_dom_doc(doc) # Lots of boring XML parsing code follows. This could definitely be cleaned # up, but the current setup makes what is going on pretty easy to # understand, so we're just going to leave it as-is for now. def _parse_sf_api_xml_from_dom_doc(self, doc): """ Parses the DOM-ified XML doc. """ results = list() results.extend(self._parse_parking_lots(doc)) results.extend(self._parse_subways(doc)) results.extend(self._parse_bus_stops(doc)) return results def _parse_parking_lots(self, doc): """ Parses parking lots out of XML response and returns them as a list. """ lot_elements = doc.getElementsByTagName("parkinglot") parsed_lots = [self._parse_parking_lot(lot_element) for lot_element in lot_elements] return parsed_lots def _parse_parking_lot(self, lot_element): """ Parses a single lot. """ # Parse location component. (name, address, latitude, longitude) = \ self._parse_location(lot_element) # Parse singleton elements for this lot. owner = get_element_text(lot_element.getElementsByTagName("owner")) capacity = get_element_text( lot_element.getElementsByTagName("capacity"), int) is_garage = get_element_text( lot_element.getElementsByTagName("isgarage"), textbool) source = get_element_text( lot_element.getElementsByTagName("source")) is_cheapest = get_element_text( lot_element.getElementsByTagName("ischeapest"), textbool) is_closest = get_element_text( lot_element.getElementsByTagName("isclosest"), textbool) # Instantiate the lot. lot = SFParkingLot(name, address, latitude, longitude, owner, capacity, is_garage, source, is_cheapest, is_closest) # Parse out the rates and add them to the lot. rate_elements = lot_element.getElementsByTagName("rate") for rate_element in rate_elements: rate = self._parse_rate(rate_element) lot.add_rate(rate) return lot def _parse_rate(self, rate_element): """ Parses a single rate. """ day_type = get_element_text( rate_element.getElementsByTagName("daytype")) price = get_element_text( rate_element.getElementsByTagName("price"), float) price_scheme = get_element_text( rate_element.getElementsByTagName("pricescheme")) daily_max = get_element_text( rate_element.getElementsByTagName("dailymax"), float) time_limit = get_element_text( rate_element.getElementsByTagName("timelimit"), float) starts = self._parse_time(rate_element.getElementsByTagName("starts")) ends = self._parse_time(rate_element.getElementsByTagName("ends")) rate = SFParkingLot.Rate(day_type, starts, ends, price, price_scheme, daily_max, time_limit) return rate def _parse_time(self, time_element): """ Parses an hour/minute time tag. """ assert len(time_element) == 1 time_element = time_element[0] hour = get_element_text(time_element.getElementsByTagName("hour"), int) minute = get_element_text( time_element.getElementsByTagName("minute"), int) return datetime.time(hour, minute) def _parse_subways(self, doc): """ Parses subways out of XML response and returns them as a list. """ subway_elements = doc.getElementsByTagName("subway") parsed_subways = [self._parse_subway(subway_element) for subway_element in subway_elements] return parsed_subways def _parse_subway(self, subway_element): """ Parses a single subway. """ # Parse location component. (name, address, latitude, longitude) = \ self._parse_location(subway_element) # Parse singleton elements for the subway. line = get_element_text(subway_element.getElementsByTagName("line")) subway = SFSubway(name, address, latitude, longitude, line) return subway def _parse_bus_stops(self, doc): """ Parses bus stops out of XML response and returns them as a list. """ bus_stop_elements = doc.getElementsByTagName("busstop") parsed_bus_stops = [self._parse_bus_stop(bus_stop_element) for bus_stop_element in bus_stop_elements] return parsed_bus_stops def _parse_bus_stop(self, bus_stop_element): """ Parses a single bus stop. """ # Parse location component. (name, address, latitude, longitude) = \ self._parse_location(bus_stop_element) # Parse the route stops. bus_stop = SFBusStop(name, address, latitude, longitude) route_stop_elements = bus_stop_element.getElementsByTagName("routestop") for route_stop_element in route_stop_elements: route_stop = self._parse_route_stop(route_stop_element) bus_stop.add_route_stop(route_stop) return bus_stop def _parse_route_stop(self, route_stop_element): """ Parses a single route stop. """ name = get_element_text( route_stop_element.getElementsByTagName("routename")) number = get_element_text( route_stop_element.getElementsByTagName("number")) schedule_link = get_element_text( route_stop_element.getElementsByTagName("schedule")) route_stop = SFBusStop.RouteStop(name, number, schedule_link) return route_stop def _parse_location(self, loc_element): """ Parses any tag that represents a location (has name, address, latitude, longitude, with address optional really.) """ name = get_element_text(loc_element.getElementsByTagName("name")) address = get_element_text(loc_element.getElementsByTagName("address")) latitude = get_element_text( loc_element.getElementsByTagName("latitude"), float) longitude = get_element_text( loc_element.getElementsByTagName("longitude"), float) return (name, address, latitude, longitude)