WaterOffice.get_from_wateroffice() - Code Metrics - abkfenris/gage-web - Measure and Improve Code Quality continuously with Scrutinizer

WaterOffice.get_from_wateroffice() A
last analyzed 2017-06-17 08:20 UTC

↳ Parent: WaterOffice

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
c	4
b	0
f	0
dl	0
loc	20
rs	9.4285
cc	3

"""
Get flows from the Canadian Water Office

Sample.remote_id should be in the form of 2 Letter Province code, underscore,
then site number. e.g.:
The Cheticamp River (http://wateroffice.ec.gc.ca/report/report_e.html?type=realTime&stn=01FC002)
would be NS_01FC002
If discharge is prefered set the remote_parameter to discharge
"""
import csv

import arrow
import requests

from .base import RemoteGage, add_new_sample


class WaterOffice(RemoteGage):
    """
    BC_07EA004
    """
    @staticmethod
    def get_from_wateroffice(remote_id):
        province = remote_id.split('_')[0]
        url = 'http://dd.weather.gc.ca/hydrometric/csv/{}/hourly/{}_hourly_hydrometric.csv'.format(province, remote_id)
        response = requests.get(url)
        riter = response.iter_lines(decode_unicode=True)
        next(riter)
        reader = csv.reader(riter)
        lines = []
        for row in reader:
            lines.append(row)
        last = lines[-1]
        level = float(last[2])
        try:
            discharge = float(last[6])
        except ValueError:
            discharge = None
        dt = arrow.get(last[1]).datetime

        return dt, level, discharge

    def get_sample(self, sensor_id):
        sensor = self.sensor(sensor_id)
        dt, level, discharge = self.get_from_wateroffice(sensor.remote_id)
        if sensor.remote_parameter == 'discharge':
            add_new_sample(sensor.id, dt, discharge)
        else:
            add_new_sample(sensor.id, dt, level)


1			"""
2			Get flows from the Canadian Water Office
3
4			Sample.remote_id should be in the form of 2 Letter Province code, underscore,
5			then site number. e.g.:
6			The Cheticamp River (http://wateroffice.ec.gc.ca/report/report_e.html?type=realTime&stn=01FC002)
7			would be NS_01FC002
8			If discharge is prefered set the remote_parameter to discharge
9			"""
10			import csv
11
12			import arrow
13			import requests
14
15			from .base import RemoteGage, add_new_sample
16
17
18			class WaterOffice(RemoteGage):
19			"""
20			BC_07EA004
21			"""
22			@staticmethod
23			def get_from_wateroffice(remote_id):
24			province = remote_id.split('_')[0]
25			url = 'http://dd.weather.gc.ca/hydrometric/csv/{}/hourly/{}_hourly_hydrometric.csv'.format(province, remote_id)
26			response = requests.get(url)
27			riter = response.iter_lines(decode_unicode=True)
28			next(riter)
29			reader = csv.reader(riter)
30			lines = []
31			for row in reader:
32			lines.append(row)
33			last = lines[-1]
34			level = float(last[2])
35			try:
36			discharge = float(last[6])
37			except ValueError:
38			discharge = None
39			dt = arrow.get(last[1]).datetime
40
41			return dt, level, discharge
42
43			def get_sample(self, sensor_id):
44			sensor = self.sensor(sensor_id)
45			dt, level, discharge = self.get_from_wateroffice(sensor.remote_id)
46			if sensor.remote_parameter == 'discharge':
47			add_new_sample(sensor.id, dt, discharge)
48			else:
49			add_new_sample(sensor.id, dt, level)
50

abkfenris / gage-web

WaterOffice.get_from_wateroffice() A last analyzed 2017-06-17 08:20 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

WaterOffice.get_from_wateroffice() A
last analyzed 2017-06-17 08:20 UTC