nojhan/weboob-devel
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Convert sachsen module to browser2

Browse source
This commit is contained in:
Florent 2014-03-19 11:22:14 +01:00
commit 89584ee168
4 changed files with 139 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

2
modules/sachsen/__init__.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Florent Fourcot # Copyright(C) 2010-2014 Florent Fourcot
# #
# This file is part of weboob. # This file is part of weboob.
# #

4
modules/sachsen/backend.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Florent Fourcot # Copyright(C) 2010-2014 Florent Fourcot
# #
# This file is part of weboob. # This file is part of weboob.
# #
@ -65,6 +65,8 @@ class SachsenLevelBackend(BaseBackend, ICapGauge):
def iter_sensors(self, gauge, pattern=None): def iter_sensors(self, gauge, pattern=None):
if not isinstance(gauge, Gauge): if not isinstance(gauge, Gauge):
gauge = self._get_gauge_by_id(gauge) gauge = self._get_gauge_by_id(gauge)
if gauge is None:
raise SensorNotFound()
if pattern is None: if pattern is None:
for sensor in gauge.sensors: for sensor in gauge.sensors:
yield sensor yield sensor

35
modules/sachsen/browser.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Florent Fourcot # Copyright(C) 2010-2014 Florent Fourcot
# #
# This file is part of weboob. # This file is part of weboob.
# #
@ -18,36 +18,23 @@
# along with weboob. If not, see <http://www.gnu.org/licenses/>. # along with weboob. If not, see <http://www.gnu.org/licenses/>.
from weboob.tools.browser import BaseBrowser from weboob.tools.browser2 import PagesBrowser, URL
from .pages import ListPage, HistoryPage from .pages import ListPage, HistoryPage
__all__ = ['SachsenBrowser'] __all__ = ['SachsenBrowser']
class SachsenBrowser(BaseBrowser): class SachsenBrowser(PagesBrowser):
DOMAIN = u'www.umwelt.sachsen.de' BASEURL = 'http://www.umwelt.sachsen.de'
ENCODING = None
PAGES = {'.*inhalt_re.html.*': ListPage,
'.*hwz/MP/.*': HistoryPage
}
homepage = '/de/wu/umwelt/lfug/lfug-internet/hwz/inhalt_re.html' homepage = URL('/de/wu/umwelt/lfug/lfug-internet/hwz/inhalt_re.html.*', ListPage)
history = URL('/de/wu/umwelt/lfug/lfug-internet/hwz/MP/(?P<params>.*)/index.html', HistoryPage)
def __init__(self, *args, **kwargs):
BaseBrowser.__init__(self, *args, **kwargs)
def home(self):
self.location(self.homepage)
def get_rivers_list(self): def get_rivers_list(self):
if not self.is_on_page(ListPage): return self.homepage.stay_or_go().get_rivers_list()
self.location(self.homepage)
return self.page.get_rivers_list()
def iter_history(self, sensor): def iter_history(self, sensor, **kwargs):
self.location('/de/wu/umwelt/lfug/lfug-internet/hwz/MP/%d/index.html' self.history.go(params=int(sensor.gaugeid))
% int(sensor.gaugeid)) kwargs['sensor'] = sensor
return self.page.iter_history(sensor) return self.page.iter_history(**kwargs)

224
modules/sachsen/pages.py
View file

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright(C) 2010-2012 Romain Bignon, Florent Fourcot # Copyright(C) 2010-2014 Florent Fourcot
# #
# This file is part of weboob. # This file is part of weboob.
# #
@ -18,121 +18,137 @@
# along with weboob. If not, see <http://www.gnu.org/licenses/>. # along with weboob. If not, see <http://www.gnu.org/licenses/>.
from datetime import datetime, date, time from datetime import datetime, date, time
from weboob.tools.browser import BasePage from weboob.tools.browser2.page import HTMLPage, method, ListElement, ItemElement
from weboob.tools.browser2.filters import Env
from weboob.capabilities.gauge import Gauge, GaugeMeasure, GaugeSensor from weboob.capabilities.gauge import Gauge, GaugeMeasure, GaugeSensor
from weboob.capabilities.base import NotAvailable, NotLoaded from weboob.capabilities.base import NotAvailable, NotLoaded
import re
__all__ = ['ListPage', 'HistoryPage'] __all__ = ['ListPage', 'HistoryPage']
class ListPage(BasePage): class ListPage(HTMLPage):
alarmlevel = {"as1.gif": u"Alarmstufe 1", "as2.gif": u"Alarmstufe 2",
"as3.gif": u"Alarmstufe 3", "as4.gig": u"Alarmstufe 4",
"qua_grau.gif": u"No alarm function", "p_gruen.gif": u"",
"qua_weiss.gif": u"no data", "as0.gif": u"",
"MNW.gif": u""}
def get_rivers_list(self): @method
for pegel in self.document.getroot().xpath(".//a[@onmouseout='pegelaus()']"): class get_rivers_list(ListElement):
div = pegel.getparent() item_xpath = ".//a[@onmouseout='pegelaus()']"
img = div.find('.//img').attrib['src'].split('/')[1]
data = pegel.attrib['onmouseover'] \
.strip('pegelein(').strip(')').replace(",'", ",").split("',")
gauge = Gauge(int(data[7]))
gauge.name = unicode(data[0].strip("'"))
gauge.city = gauge.name.split(' ')[0]
gauge.object = unicode(data[1])
sensors = [] class item(ItemElement):
try: klass = Gauge
lastdate = date(*reversed([int(x)
for x in data[2].split(' ')[0].split(".")]))
lasttime = time(*[int(x)
for x in data[2].split(' ')[1].split(":")])
lastdate = datetime.combine(lastdate, lasttime)
except:
lastdate = NotAvailable
bildforecast = data[5] alarmlevel = {"as1.gif": u"Alarmstufe 1", "as2.gif": u"Alarmstufe 2",
if bildforecast == "pf_gerade.png": "as3.gif": u"Alarmstufe 3", "as4.gig": u"Alarmstufe 4",
forecast = u"stable" "qua_grau.gif": u"No alarm function", "p_gruen.gif": u"",
elif bildforecast == "pf_unten.png": "qua_weiss.gif": u"no data", "as0.gif": u"",
forecast = u"Go down" "MNW.gif": u""}
elif bildforecast == "pf_oben.png":
forecast = u"Go up"
else:
forecast = NotAvailable
try: obj_id = Env('id')
level = float(data[3]) obj_name = Env('name')
except: obj_city = Env('city')
level = NotAvailable obj_object = Env('object')
levelsensor = GaugeSensor(gauge.id + "-level") obj_sensors = Env('sensors')
levelsensor.name = u"Level"
levelsensor.unit = u"cm"
levelsensor.forecast = forecast
lastvalue = GaugeMeasure()
lastvalue.level = level
lastvalue.date = lastdate
try:
lastvalue.alarm = self.alarmlevel[img]
except KeyError:
lastvalue.alarm = u""
levelsensor.lastvalue = lastvalue
levelsensor.history = NotLoaded
levelsensor.gaugeid = gauge.id
sensors.append(levelsensor)
try: def init_sensor(self, _id, name, unit, value, forecast, alarm, date):
flow = float(data[4]) sensor = GaugeSensor("%s-%s" % (_id, name.lower()))
except: sensor.name = name
flow = NotAvailable sensor.unit = unit
flowsensor = GaugeSensor(gauge.id + "-flow") sensor.forecast = forecast
flowsensor.name = u"Flow" lastvalue = GaugeMeasure()
flowsensor.unit = u"m3/s" lastvalue.alarm = alarm
flowsensor.forecast = forecast
lastvalue = GaugeMeasure()
lastvalue.level = flow
lastvalue.date = lastdate
try:
lastvalue.alarm = self.alarmlevel[img]
except KeyError:
lastvalue.alarm = u""
flowsensor.lastvalue = lastvalue
flowsensor.history = NotLoaded
flowsensor.gaugeid = gauge.id
sensors.append(flowsensor)
gauge.sensors = sensors
yield gauge
class HistoryPage(BasePage):
def iter_history(self, sensor):
table = self.document.getroot().cssselect('table[width="215"]')
lines = table[0].cssselect("tr")
lines.pop(0) # remove header
lines.pop(0) # remove first value (already in lastvalue)
for line in lines:
history = GaugeMeasure()
leveldate = date(*reversed([int(x)
for x in line[0].text_content().split(' ')[0].split(".")]))
leveltime = time(*[int(x)
for x in line[0].text_content().split(' ')[1].split(":")])
history.date = datetime.combine(leveldate, leveltime)
if sensor.name == u"Level":
try: try:
history.level = float(line[1].text_content()) lastvalue.level = float(value)
except: except:
history.level = NotAvailable lastvalue.level = NotAvailable
elif sensor.name == u"Flow": lastvalue.date = date
try: sensor.lastvalue = lastvalue
history.level = float(line[2].text_content()) sensor.history = NotLoaded
except: sensor.gaugeid = unicode(_id)
history.level = NotAvailable
# TODO: history.alarm return sensor
yield history
def parse(self, el):
div = el.getparent()
img = div.find('.//img').attrib['src'].split('/')[1]
data = el.attrib['onmouseover'] \
.strip('pegelein(').strip(')').replace(",'", ",").split("',")
self.env['id'] = data[7].strip()
self.env['name'] = unicode(data[0].strip("'"))
self.env['city'] = self.env['name'].split(' ')[0]
self.env['object'] = unicode(data[1])
sensors = []
try:
datenumbers = data[2].split(' ')[0].split(".")
timenumbers = data[2].split(' ')[1].split(":")
lastdate = date(*reversed([int(x) for x in datenumbers]))
lasttime = time(*[int(x) for x in timenumbers])
lastdate = datetime.combine(lastdate, lasttime)
except:
lastdate = NotAvailable
bildforecast = data[5]
if bildforecast == "pf_gerade.png":
forecast = u"stable"
elif bildforecast == "pf_unten.png":
forecast = u"Go down"
elif bildforecast == "pf_oben.png":
forecast = u"Go up"
else:
forecast = NotAvailable
try:
alarm = self.alarmlevel[img]
except KeyError:
alarm = u""
levelsensor = self.init_sensor(self.env['id'], u"Level",
u"cm", data[3], forecast,
alarm, lastdate)
sensors.append(levelsensor)
flowsensor = self.init_sensor(self.env['id'], u"Flow",
u"m3/s", data[4], forecast,
alarm, lastdate)
sensors.append(flowsensor)
self.env['sensors'] = sensors
class HistoryPage(HTMLPage):
@method
class iter_history(ListElement):
item_xpath = '//table[@width="215"]/tr'
class item(ItemElement):
klass = GaugeMeasure
verif = re.compile("\d\d.\d\d.\d+ \d\d:\d\d")
obj_date = Env('date')
obj_level = Env('level')
obj_id = None
def condition(self):
if self.verif.match(self.el[0].text_content()):
return True
return False
def parse(self, line):
leveldate = date(*reversed([int(x)
for x in line[0].text_content().split(' ')[0].split(".")]))
leveltime = time(*[int(x)
for x in line[0].text_content().split(' ')[1].split(":")])
self.env['date'] = datetime.combine(leveldate, leveltime)
if self.env['sensor'].name == u"Level":
try:
self.env['level'] = float(line[1].text_content())
except:
self.env['level'] = NotAvailable
elif self.env['sensor'].name == u"Flow":
try:
self.env['level'] = float(line[2].text_content())
except:
self.env['level'] = NotAvailable
# TODO: history.alarm