velib: use the JCDecaux REST API to support additional cities
This commit is contained in:
Vincent A
parent
0443855602
commit
6ac2dec74b
3 changed files with 110 additions and 144 deletions
|
|
@ -19,14 +19,20 @@
|
|||
|
||||
|
||||
from weboob.tools.backend import BaseBackend
|
||||
from weboob.capabilities.gauge import ICapGauge, GaugeSensor, Gauge, SensorNotFound
|
||||
from weboob.capabilities.gauge import ICapGauge, GaugeSensor, Gauge, GaugeMeasure, SensorNotFound
|
||||
|
||||
from .browser import VelibBrowser
|
||||
|
||||
import re
|
||||
|
||||
__all__ = ['VelibBackend']
|
||||
|
||||
SENSOR_TYPES = {u'available_bike_stands': u'Free stands', u'available_bikes': u'Available bikes', u'bike_stands': u'Total stands'}
|
||||
|
||||
|
||||
class BikeMeasure(GaugeMeasure):
|
||||
def __repr__(self):
|
||||
return '<GaugeMeasure level=%d>' % self.level
|
||||
|
||||
|
||||
class VelibBackend(BaseBackend, ICapGauge):
|
||||
NAME = 'velib'
|
||||
|
|
@ -39,13 +45,55 @@ class VelibBackend(BaseBackend, ICapGauge):
|
|||
BROWSER = VelibBrowser
|
||||
STORAGE = {'boards': {}}
|
||||
|
||||
def __init__(self, *a, **kw):
|
||||
super(VelibBackend, self).__init__(*a, **kw)
|
||||
self.cities = None
|
||||
|
||||
def _make_gauge(self, info):
|
||||
gauge = Gauge(info['id'])
|
||||
gauge.name = unicode(info['name'])
|
||||
gauge.city = unicode(info['city'])
|
||||
gauge.object = u'bikes'
|
||||
return gauge
|
||||
|
||||
def _make_sensor(self, sensor_type, info, gauge):
|
||||
id = '%s.%s' % (sensor_type, gauge.id)
|
||||
sensor = GaugeSensor(id)
|
||||
sensor.gaugeid = gauge.id
|
||||
sensor.name = SENSOR_TYPES[sensor_type]
|
||||
sensor.address = unicode(info['address'])
|
||||
sensor.history = []
|
||||
return sensor
|
||||
|
||||
def _make_measure(self, sensor_type, info):
|
||||
measure = BikeMeasure()
|
||||
measure.date = info['last_update']
|
||||
measure.level = float(info[sensor_type])
|
||||
return measure
|
||||
|
||||
def _parse_gauge(self, info):
|
||||
gauge = self._make_gauge(info)
|
||||
gauge.sensors = []
|
||||
|
||||
for type in SENSOR_TYPES:
|
||||
sensor = self._make_sensor(type, info, gauge)
|
||||
measure = self._make_measure(type, info)
|
||||
sensor.lastvalue = measure
|
||||
gauge.sensors.append(sensor)
|
||||
|
||||
return gauge
|
||||
|
||||
def iter_gauges(self, pattern=None):
|
||||
if pattern is None:
|
||||
for gauge in self.browser.get_station_list():
|
||||
yield gauge
|
||||
for jgauge in self.browser.get_station_list():
|
||||
yield self._parse_gauge(jgauge)
|
||||
else:
|
||||
self._fetch_cities()
|
||||
lowpattern = pattern.lower()
|
||||
for gauge in self.browser.get_station_list():
|
||||
|
||||
contract = self.cities.get(lowpattern)
|
||||
for jgauge in self.browser.get_station_list(contract=contract):
|
||||
gauge = self._parse_gauge(jgauge)
|
||||
if lowpattern in gauge.name.lower() or lowpattern in gauge.city.lower():
|
||||
yield gauge
|
||||
|
||||
|
|
@ -55,7 +103,6 @@ class VelibBackend(BaseBackend, ICapGauge):
|
|||
if gauge is None:
|
||||
raise SensorNotFound()
|
||||
|
||||
gauge.sensors = self.browser.get_station_infos(gauge)
|
||||
if pattern is None:
|
||||
for sensor in gauge.sensors:
|
||||
yield sensor
|
||||
|
|
@ -72,21 +119,28 @@ class VelibBackend(BaseBackend, ICapGauge):
|
|||
raise SensorNotFound()
|
||||
return sensor.lastvalue
|
||||
|
||||
def _get_gauge_by_id(self, id):
|
||||
for gauge in self.browser.get_station_list():
|
||||
if id == gauge.id:
|
||||
return gauge
|
||||
return None
|
||||
def _fetch_cities(self):
|
||||
if self.cities:
|
||||
return
|
||||
|
||||
def _get_sensor_by_id(self, id):
|
||||
reSensorId = re.search('(\d+)-((bikes|attach|status))', id, re.IGNORECASE)
|
||||
if reSensorId:
|
||||
gauge = reSensorId.group(1)
|
||||
pattern = reSensorId.group(2)
|
||||
sensor_generator = self.iter_sensors(gauge, pattern)
|
||||
if sensor_generator:
|
||||
return next(sensor_generator)
|
||||
else:
|
||||
return None
|
||||
self.cities = {}
|
||||
jcontract = self.browser.get_contracts_list()
|
||||
for jcontract in jcontract:
|
||||
for city in jcontract['cities']:
|
||||
self.cities[city.lower()] = jcontract['name']
|
||||
|
||||
def _get_gauge_by_id(self, id):
|
||||
jgauge = self.browser.get_station_infos(id)
|
||||
if jgauge:
|
||||
return self._parse_gauge(jgauge)
|
||||
else:
|
||||
return None
|
||||
|
||||
def _get_sensor_by_id(self, id):
|
||||
_, gauge_id = id.split('.', 1)
|
||||
gauge = self._get_gauge_by_id(gauge_id)
|
||||
if not gauge:
|
||||
raise SensorNotFound()
|
||||
for sensor in gauge.sensors:
|
||||
if sensor.id.lower() == id.lower():
|
||||
return sensor
|
||||
|
|
|
|||
|
|
@ -18,29 +18,49 @@
|
|||
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
import datetime
|
||||
from weboob.tools.browser import BaseBrowser
|
||||
|
||||
from .pages import ListStationsPage, InfoStationPage
|
||||
|
||||
|
||||
__all__ = ['VelibBrowser']
|
||||
|
||||
|
||||
class VelibBrowser(BaseBrowser):
|
||||
PROTOCOL = 'http'
|
||||
DOMAIN = 'www.velib.paris.fr/service'
|
||||
ENCODING = None
|
||||
ENCODING = 'utf-8'
|
||||
|
||||
PAGES = {
|
||||
'%s://%s/stationdetails/paris/.*' % (PROTOCOL, DOMAIN): InfoStationPage,
|
||||
'%s://%s/carto' % (PROTOCOL, DOMAIN): ListStationsPage,
|
||||
}
|
||||
API_KEY = '2282a34b49cf45d8129cdf93d88762914cece88b'
|
||||
BASE_URL = 'https://api.jcdecaux.com/vls/v1/'
|
||||
|
||||
def get_station_list(self):
|
||||
if not self.is_on_page(ListStationsPage):
|
||||
self.location(u'%s://%s/carto' % (self.PROTOCOL, self.DOMAIN))
|
||||
return self.page.get_station_list()
|
||||
def __init__(self, *a, **kw):
|
||||
kw['parser'] = 'json'
|
||||
BaseBrowser.__init__(self, *a, **kw)
|
||||
|
||||
def do_get(self, path, **query):
|
||||
qs = '&'.join('%s=%s' % kv for kv in query.items())
|
||||
if qs:
|
||||
qs = '&' + qs
|
||||
url = '%s%s?apiKey=%s%s' % (self.BASE_URL, path, self.API_KEY, qs)
|
||||
return self.get_document(self.openurl(url))
|
||||
|
||||
def get_contracts_list(self):
|
||||
return self.do_get('contracts')
|
||||
|
||||
def get_station_list(self, contract=None):
|
||||
if contract:
|
||||
doc = self.do_get('stations', contract=contract)
|
||||
else:
|
||||
doc = self.do_get('stations')
|
||||
for jgauge in doc:
|
||||
self._transform(jgauge)
|
||||
return doc
|
||||
|
||||
def get_station_infos(self, gauge):
|
||||
self.location('%s://%s/stationdetails/paris/%s' % (self.PROTOCOL, self.DOMAIN, gauge.id))
|
||||
return self.page.get_station_infos(gauge.id)
|
||||
station_id, contract = gauge.split('.', 1)
|
||||
doc = self.do_get('stations/%s' % station_id, contract=contract)
|
||||
return self._transform(doc)
|
||||
|
||||
def _transform(self, jgauge):
|
||||
jgauge['id'] = '%s.%s' % (jgauge['number'], jgauge['contract_name'])
|
||||
jgauge['city'] = jgauge['contract_name']
|
||||
jgauge['last_update'] = datetime.datetime.fromtimestamp(jgauge['last_update'] / 1000)
|
||||
return jgauge
|
||||
|
|
|
|||
|
|
@ -1,108 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 dud
|
||||
#
|
||||
# This file is part of weboob.
|
||||
#
|
||||
# weboob is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# weboob is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
from weboob.tools.browser import BasePage
|
||||
from weboob.capabilities.gauge import Gauge, GaugeMeasure, GaugeSensor
|
||||
from weboob.capabilities.base import NotLoaded
|
||||
import datetime
|
||||
import re
|
||||
|
||||
__all__ = ['InfoStationPage', 'ListStationsPage']
|
||||
|
||||
AdresseStation = {}
|
||||
|
||||
|
||||
class InfoStationPage(BasePage):
|
||||
def _create_bikes_sensor(self, value, gauge_id, last_update, adresse):
|
||||
levelbikes = GaugeSensor(gauge_id + '-bikes')
|
||||
levelbikes.name = u'Bikes'
|
||||
levelbikes.address = u'%s' % adresse
|
||||
lastvalue = GaugeMeasure()
|
||||
lastvalue.level = float(value)
|
||||
lastvalue.date = last_update
|
||||
if lastvalue.level < 1:
|
||||
lastvalue.alarm = u'Empty station'
|
||||
levelbikes.lastvalue = lastvalue
|
||||
levelbikes.history = NotLoaded
|
||||
levelbikes.gaugeid = gauge_id
|
||||
return levelbikes
|
||||
|
||||
def _create_attach_sensor(self, value, gauge_id, last_update, adresse):
|
||||
levelattach = GaugeSensor(gauge_id + '-attach')
|
||||
levelattach.name = u'Attach'
|
||||
levelattach.address = u'%s' % adresse
|
||||
lastvalue = GaugeMeasure()
|
||||
if lastvalue.level < 1:
|
||||
lastvalue.alarm = u'Full station'
|
||||
lastvalue.level = float(value)
|
||||
lastvalue.date = last_update
|
||||
levelattach.lastvalue = lastvalue
|
||||
levelattach.history = NotLoaded
|
||||
levelattach.gaugeid = gauge_id
|
||||
return levelattach
|
||||
|
||||
def _create_status_sensor(self, value, gauge_id, last_update, adresse):
|
||||
levelstatus = GaugeSensor(gauge_id + '-status')
|
||||
levelstatus.name = u'Status'
|
||||
levelstatus.address = u'%s' % adresse
|
||||
lastvalue = GaugeMeasure()
|
||||
status = float(value)
|
||||
if status == 0:
|
||||
status = 1
|
||||
else:
|
||||
status = -1
|
||||
if lastvalue.level < 1:
|
||||
lastvalue.alarm = u'Not available station'
|
||||
lastvalue.level = float(status)
|
||||
lastvalue.date = last_update
|
||||
levelstatus.lastvalue = lastvalue
|
||||
levelstatus.history = NotLoaded
|
||||
levelstatus.gaugeid = gauge_id
|
||||
return levelstatus
|
||||
|
||||
def _get_last_update(self, last_update):
|
||||
return datetime.datetime.now() - datetime.timedelta(seconds=int(re.match(r'\d+', last_update).group(0)))
|
||||
|
||||
def get_station_infos(self, gauge_id):
|
||||
sensors = []
|
||||
|
||||
last_update = datetime.datetime.fromtimestamp(float(self.parser.select(self.document.getroot(), 'updated', 1).text))
|
||||
adresse = AdresseStation[gauge_id]
|
||||
|
||||
sensors.append(self._create_bikes_sensor(self.parser.select(self.document.getroot(), 'available', 1).text, gauge_id, last_update, adresse))
|
||||
sensors.append(self._create_attach_sensor(self.parser.select(self.document.getroot(), 'free', 1).text, gauge_id, last_update, adresse))
|
||||
sensors.append(self._create_status_sensor(self.parser.select(self.document.getroot(), 'open', 1).text, gauge_id, last_update, adresse))
|
||||
|
||||
return sensors
|
||||
|
||||
|
||||
class ListStationsPage(BasePage):
|
||||
def get_station_list(self):
|
||||
gauges = []
|
||||
for marker in self.parser.select(self.document.getroot(), 'marker'):
|
||||
gauge = Gauge(int(marker.get('number')))
|
||||
gauge.name = unicode(marker.get('address')).rsplit('-', 1)[0]
|
||||
full_address = re.search(r'\d\d\d\d\d.*', unicode(marker.get('fulladdress')))
|
||||
if full_address:
|
||||
gauge.city = full_address.group()
|
||||
gauge.object = u'velib'
|
||||
gauges.append(gauge)
|
||||
AdresseStation[marker.get('number')] = marker.get('fulladdress')
|
||||
return gauges
|
||||
Loading…
Add table
Add a link
Reference in a new issue