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

[mareeinfo] add new module mareeinfo

Browse source
This commit is contained in:
Bezleputh 2014-09-15 00:58:10 +02:00 • committed by Romain Bignon
commit 11ce95fd6a
5 changed files with 369 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

24
modules/mareeinfo/__init__.py Normal file
View file

@ -0,0 +1,24 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Bezleputh
#
# 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 .backend import MareeinfoBackend
__all__ = ['MareeinfoBackend']

68
modules/mareeinfo/backend.py Normal file
View file

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Bezleputh
#
# 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.backend import BaseBackend
from weboob.capabilities.base import find_object
from weboob.capabilities.gauge import CapGauge, Gauge, SensorNotFound
from .browser import MareeinfoBrowser
__all__ = ['MareeinfoBackend']
class MareeinfoBackend(BaseBackend, CapGauge):
NAME = 'mareeinfo'
DESCRIPTION = u'Un module qui permet d\' aller a la pêche aux moules totalement informé'
MAINTAINER = u'Bezleputh'
EMAIL = 'carton_ben@yahoo.fr'
LICENSE = 'AGPLv3+'
VERSION = '1.0'
BROWSER = MareeinfoBrowser
def get_last_measure(self, sensor_id):
gauge_id = sensor_id.split('-')[0]
return find_object(self.iter_sensors(gauge_id), id=sensor_id, error=SensorNotFound).lastvalue
def iter_gauge_history(self, sensor_id):
gauge_id = sensor_id.split('-')[0]
return find_object(self.iter_sensors(gauge_id), id=sensor_id, error=SensorNotFound).history
def iter_gauges(self, pattern=None):
for _gauge in self.browser.get_harbor_list(pattern):
if pattern is not None:
gauge = self.browser.get_harbor_infos(_gauge)
yield gauge
else:
yield _gauge
def iter_sensors(self, gauge, pattern=None):
if not isinstance(gauge, Gauge):
gauge = find_object(self.iter_gauges(), id=gauge, error=SensorNotFound)
gauge = self.browser.get_harbor_infos(gauge)
if pattern is None:
for sensor in gauge.sensors:
yield sensor
else:
lowpattern = pattern.lower()
for sensor in gauge.sensors:
if lowpattern in sensor.name.lower():
yield sensor

35
modules/mareeinfo/browser.py Normal file
View file

@ -0,0 +1,35 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Bezleputh
#
# 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.browser2 import PagesBrowser, URL
from .pages import IndexPage
class MareeinfoBrowser(PagesBrowser):
BASEURL = 'http://maree.info'
harbor_page = URL('', '(?P<_id>.*)', IndexPage)
def get_harbor_list(self, pattern):
return self.harbor_page.go().get_harbor_list(pattern=pattern)
def get_harbor_infos(self, gauge):
return self.harbor_page.go(_id=gauge.id).get_harbor_infos(obj=gauge)

205
modules/mareeinfo/pages.py Normal file
View file

@ -0,0 +1,205 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Bezleputh
#
# 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.browser2.page import HTMLPage, method
from weboob.tools.browser2.elements import ListElement, ItemElement
from weboob.tools.browser2.filters import CleanText, Link, DateTime, CleanDecimal, Regexp, XPath
from weboob.capabilities.gauge import Gauge, GaugeMeasure, GaugeSensor
from datetime import timedelta
import re
class IndexPage(HTMLPage):
@method
class get_harbor_list(ListElement):
item_xpath = "//a[@class='Port PP'] | //a[@class='Port PS']"
class item(ItemElement):
klass = Gauge
obj_id = CleanText(Link('.'), replace=[('/', '')])
obj_name = CleanText('.')
obj_city = CleanText('.')
obj_object = u'Port'
def validate(self, obj):
if self.env['pattern']:
return self.env['pattern'].lower() in obj.name.lower()
return True
@method
class get_harbor_infos(ItemElement):
klass = Gauge
def _create_coef_sensor(self, gauge_id, AM=True):
name = CleanText('//tr[@class="MJE"]/th[4]')(self)
_name = 'matin' if AM else 'aprem'
value = self._get_coef_value(AM=AM)
if value:
coef = GaugeSensor(u'%s-%s-%s' % (gauge_id, name, _name))
coef.name = '%s %s' % (name, _name)
coef.lastvalue = value
coef.gaugeid = gauge_id
coef.history = []
for jour in range(0, 7):
measure = self._get_coef_value(AM=AM, jour=jour)
if measure:
coef.history.append(measure)
return coef
def _get_coef_value(self, AM=True, jour=0):
if AM:
time = DateTime(CleanText('//tr[@id="MareeJours_%s"]/td[1]/b[1]' % jour))(self)
value = CleanText('//tr[@id="MareeJours_%s"]/td[3]/b[1]' % jour)(self)
else:
time, value = None, None
if len(XPath('//tr[@id="MareeJours_%s"]/td[1]/b' % jour)(self)) > 1:
time = DateTime(CleanText('//tr[@id="MareeJours_%s"]/td[1]/b[2]' % jour))(self)
value = CleanText('//tr[@id="MareeJours_%s"]/td[3]/b[2]' % jour)(self)
if time and value:
measure = GaugeMeasure()
measure.level = float(value)
measure.date = time + timedelta(days=jour)
return measure
def _create_high_tide(self, gauge_id, AM=True):
name = CleanText('//tr[@class="MJE"]/th[3]')(self)
_name = 'matin' if AM else 'aprem'
value = self._get_high_tide_value(AM=AM)
if value:
tide = GaugeSensor(u'%s-%s-%s' % (gauge_id, name, _name))
tide.name = u'Pleine Mer %s' % (_name)
tide.unit = u'm'
tide.lastvalue = value
tide.gaugeid = gauge_id
tide.history = []
for jour in range(0, 7):
measure = self._get_high_tide_value(AM=AM, jour=jour)
if measure:
tide.history.append(measure)
return tide
def _get_high_tide_value(self, AM=True, jour=0):
if AM:
time = DateTime(CleanText('//tr[@id="MareeJours_%s"]/td[1]/b[1]' % jour))(self)
value = CleanDecimal('//tr[@id="MareeJours_0"]/td[2]/b[1]', replace_dots=True)(self)
else:
time, value = None, None
if len(XPath('//tr[@id="MareeJours_%s"]/td[1]/b' % jour)(self)) > 1:
time = DateTime(CleanText('//tr[@id="MareeJours_%s"]/td[1]/b[2]' % jour),
default=None)(self)
value = CleanDecimal('//tr[@id="MareeJours_0"]/td[2]/b[2]', replace_dots=True,
default=None)(self)
if time and value:
measure = GaugeMeasure()
measure.level = float(value)
measure.date = time + timedelta(days=jour)
return measure
def _create_low_tide(self, gauge_id, AM=True):
name = CleanText('//tr[@class="MJE"]/th[3]')(self)
_name = 'matin' if AM else 'aprem'
value = self._get_low_tide_value(AM=AM)
if value:
tide = GaugeSensor(u'%s-%s-%s' % (gauge_id, name, _name))
tide.name = u'Basse Mer %s' % (_name)
tide.unit = u'm'
tide.lastvalue = value
tide.gaugeid = gauge_id
tide.history = []
for jour in range(0, 7):
measure = self._get_low_tide_value(AM=AM, jour=jour)
if measure:
tide.history.append(measure)
return tide
def _is_low_tide_first(self, jour):
return XPath('//tr[@id="MareeJours_%s"]/td[1]' % jour)(self)[0].getchildren()[0].tag != 'b'
def _get_low_tide_value(self, AM=True, jour=0):
slow_tide_pos = 1 if self._is_low_tide_first(jour) else 2
m = re.findall('(\d{2}h\d{2})', CleanText('//tr[@id="MareeJours_%s"]/td[1]' % jour)(self))
re_time = '(\d{2}h\d{2}).*(\d{2}h\d{2}).*(\d{2}h\d{2})'
re_value = '(.*)m(.*)m(.*)m'
if len(m) > 3:
re_time = '(\d{2}h\d{2}).*(\d{2}h\d{2}).*(\d{2}h\d{2}).*(\d{2}h\d{2})'
re_value = '(.*)m(.*)m(.*)m(.*)m'
if AM:
time = DateTime(Regexp(CleanText('//tr[@id="MareeJours_%s"]/td[1]' % jour),
re_time,
'\\%s' % slow_tide_pos))(self)
value = CleanDecimal(Regexp(CleanText('//tr[@id="MareeJours_%s"]/td[2]' % jour),
re_value,
'\\%s' % slow_tide_pos),
replace_dots=True, default=None)(self)
else:
slow_tide_pos += 2
time, value = None, None
if len(m) > slow_tide_pos - 1:
time = DateTime(Regexp(CleanText('//tr[@id="MareeJours_%s"]/td[1]' % jour),
re_time,
'\\%s' % slow_tide_pos))(self)
value = CleanDecimal(Regexp(CleanText('//tr[@id="MareeJours_%s"]/td[2]' % jour),
re_value,
'\\%s' % slow_tide_pos),
replace_dots=True, default=None)(self)
if time and value:
measure = GaugeMeasure()
measure.level = float(value)
measure.date = time + timedelta(days=jour)
return measure
def obj_sensors(self):
sensors = []
high_tide_PM = self._create_high_tide(self.obj.id)
if high_tide_PM:
sensors.append(high_tide_PM)
high_tide_AM = self._create_high_tide(self.obj.id, AM=False)
if high_tide_AM:
sensors.append(high_tide_AM)
low_tide_AM = self._create_low_tide(self.obj.id)
if low_tide_AM:
sensors.append(low_tide_AM)
low_tide_PM = self._create_low_tide(self.obj.id, AM=False)
if low_tide_PM:
sensors.append(low_tide_PM)
coef_AM = self._create_coef_sensor(self.obj.id)
if coef_AM:
sensors.append(coef_AM)
coef_PM = self._create_coef_sensor(self.obj.id, AM=False)
if coef_PM:
sensors.append(coef_PM)
return sensors

37
modules/mareeinfo/test.py Normal file
View file

@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Bezleputh
#
# 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.test import BackendTest
class MareeinfoTest(BackendTest):
BACKEND = 'mareeinfo'
def test_mareeinfo(self):
l = list(self.backend.iter_gauges())
self.assertTrue(len(l) > 0)
gauge = l[0]
s = list(self.backend.iter_sensors(gauge))
self.assertTrue(len(s) > 0)
sensor = s[0]
self.assertTrue(self.backend.get_last_measure(sensor.id) is not None)
self.assertTrue(len(self.backend.iter_gauge_history(sensor.id)) > 0)