-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
276 lines (220 loc) · 8.26 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
"""
Smart electricity meter consumption data scraper for e-st.lv
"""
import json
import argparse
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode
import requests
from pyquery import PyQuery
class STScraper:
"""
Smart electricity meter consumption data scraper class for e-st.lv
"""
BASE_HOST = 'https://mans.e-st.lv'
LOGIN_URL = BASE_HOST + '/lv/private/user-authentification/'
DATA_URL = BASE_HOST + '/lv/private/paterini-un-norekini/paterinu-grafiki/'
# Data url parameters
PERIOD_DAY = 'D'
PERIOD_MONTH = 'M'
PERIOD_YEAR = 'Y'
GRANULARITY_HOUR = 'H'
GRANULARITY_DAY = 'D'
def __init__(self, login, password, object_id, meter_id):
"""
Class initialisation
:param login: username
:param password: user password
:param meter_id: object EIC ID
:param meter_id: smart meter ID
"""
self.login = login
self.password = password
self.object_id = object_id
self.meter_id = meter_id
self.session = requests.Session()
def _get_current_date(self):
"""Get the current year, month, and day as a dictionary."""
now = datetime.now()
return {'year': now.strftime('%Y'), 'month': now.strftime('%m'), 'day': now.strftime('%d')}
def _get_current_year(self):
"""
Get the current year e.g. 2024
"""
return self._get_current_date()['year']
def _get_current_month(self):
"""
Get the current month e.g. 02
"""
return self._get_current_date()['month']
def _get_current_day(self):
"""
Get the current day e.g. 14
"""
return self._get_current_date()['day']
def _get_data_url(self, period=None, year=None, month=None, day=None, granularity=None):
"""
Prepare data url depending on request type and parameters
:param period: report time period
:type period: str | should be one of self.PERIOD_<*>
:param year: year
:type year: str | None
:param month: month
:type month: str | None
:param day: day
:type day: str | None
:param granularity: report data type
:type granularity: str | should be one of self.GRANULARITY_<*>
:return: generated data url
"""
params = {
'objectEic': self.object_id,
'counterNumber': self.meter_id,
'period': period
}
year = year or self._get_current_year()
if period == self.PERIOD_YEAR:
params['year'] = year
if period == self.PERIOD_MONTH:
params['year'] = year
params['month'] = month or self._get_current_month()
params['granularity'] = granularity
if period == self.PERIOD_DAY:
month = month or self._get_current_month()
day = day or self._get_current_day()
params['date'] = f'{day}.{month}.{year}'
params['granularity'] = granularity
print(self.DATA_URL + '?' + urlencode(params))
return self.DATA_URL + '?' + urlencode(params)
@staticmethod
def _format_timestamp(timestamp):
"""
Convert JS timestamp to human readable date and time
:param timestamp: timestamp
:return: datetime e.g. 2024-02-14 02:00:00
:rtype: str
"""
return datetime.fromtimestamp(
int(timestamp) / 1000.0, tz=timezone(timedelta(hours=0))
).strftime('%Y-%m-%d %H:%M:%S')
def _format_response(self, response_data, neto=True):
"""
Parse out the real data from graph json
:param response_data: graph json source
:return: parsed data
"""
response_cons_data = response_data['values']['A+']['total']['data']
if "A-" in response_data['values'].keys():
response_neto_data = response_data['values']['A-']['total']['data']
else:
neto = False
cons_data = [{
'data': self._format_timestamp(item['timestamp']),
'value': item['value']
} for item in response_cons_data]
if neto:
neto_data = [{
'data': self._format_timestamp(item['timestamp']),
'value': item['value']
} for item in response_neto_data]
data = {
'A+': cons_data,
'A-': neto_data
}
else:
data = cons_data
return data
def get_day_data(self, neto=True, year=None, month=None, day=None):
"""
Get the data of the specified day
:param year:
:param month:
:param day:
:return:
"""
response = self._fetch_remote_data(period=self.PERIOD_DAY, month=month,
year=year, day=day,
granularity=self.GRANULARITY_HOUR)
return self._format_response(response, neto)
def get_month_data(self, neto=True, year=None, month=None):
"""
Get the data of the specified month
:param year:
:param month:
:return:
"""
response = self._fetch_remote_data(period=self.PERIOD_MONTH, month=month, year=year,
granularity=self.GRANULARITY_DAY)
return self._format_response(response, neto)
def get_year_data(self, neto=True, year=None):
"""
Get the data of the specified year
:param year:
:return:
"""
response = self._fetch_remote_data(period=self.PERIOD_YEAR, year=year)
return self._format_response(response, neto)
def _fetch_remote_data(self, **kwargs):
"""
Do authentification and retrieve the data
:param kwargs:
:return:
"""
response = self.session.get(self._get_data_url(**kwargs))
root = PyQuery(response.text)
fields = (
'_token',
'returnUrl',
'login',
'password'
)
values = {}
for field in fields:
values[field] = root(f'input[name={field}]').attr('value')
values['login'] = self.login
values['password'] = self.password
response = self.session.post(self.LOGIN_URL, data=values)
root = PyQuery(response.text)
values = root('div.chart').attr('data-values')
return json.loads(values)
def main():
"""
Main function
"""
parser = argparse.ArgumentParser(
description='mans.e-st.lv electricity consumption data scraper'
)
parser.add_argument('--username', default=None, help='Website username')
parser.add_argument('--password', default=None, help='Website password')
parser.add_argument('--objectid', default=None, help='Object ID')
parser.add_argument('--meter', default=None, help='Electricity meter ID')
parser.add_argument('--period', default='month', help='Report data time period')
parser.add_argument('--year', default=None)
parser.add_argument('--month', default=None)
parser.add_argument('--day', default=None)
parser.add_argument('--neto', default=True, help="Include generation data")
parser.add_argument('--outfile', default=None, help='Save data in specified file')
opts = parser.parse_args()
if not opts.username or not opts.password:
raise TypeError('Username and/or password must be set')
if not opts.objectid:
raise TypeError('Object ID must be set')
if not opts.meter:
raise TypeError('Electricity meter ID must be set')
scraper = STScraper(opts.username, opts.password, opts.objectid, opts.meter)
# Adjusted call to data retrieval methods based on selected period
if opts.period == 'year':
data = scraper.get_year_data(opts.neto, opts.year)
elif opts.period == 'month':
data = scraper.get_month_data(opts.neto, opts.year, opts.month)
elif opts.period == 'day':
data = scraper.get_day_data(opts.neto, opts.year, opts.month, opts.day)
else:
raise ValueError("Invalid period specified")
if opts.outfile:
with open(opts.outfile, 'w', encoding="utf8") as f:
json.dump(data, f, indent=4)
else:
print(json.dumps(data, indent=4))
if __name__ == '__main__':
main()