-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathrterequests.py
95 lines (84 loc) · 3.71 KB
/
rterequests.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
import requests
from datetime import datetime, timedelta, date
import json
# API user guide : https://data.rte-france.com/catalog/-/api/user_guide/231896
# This base64 encoded client:secret is generated by RTE for each app
# see : https://data.rte-france.com/group/guest/apps
RTE_ID64= 'NTE1MzdiNTktMjhhZi00ZjFmLWE5YWMtN2M4MTJkNGIwYmE5OjYwMzEzNTU2LTQxNzQtNDA4My1hOGVlLWQxN2QyZTIwOWIwZQ=='
# a temporary valid token
RTE_TOKEN = 'TYWJyVJuvZxHGGxjWsZqhNUlTdAJd4qXNmNYwlBo7ukPZNPG8fytWN'
def requestToken():
"""
OAuth v.2 token request to RTE identification server
:return: a token valid for 2 hours
"""
header = {'Authorization' : 'Basic ' + RTE_ID64}
r = requests.post('https://digital.iservices.rte-france.com/token/oauth/', headers=header)
# print(r.status_code)
# print(r.headers)
# print(r.text)
if r.status_code == requests.codes.ok :
token = r.json()['access_token']
print('Gathered token : ' + token)
return token
else :
return ''
def headerWithToken():
return {'Authorization' : 'Bearer ' + RTE_TOKEN}
def getActualGenerationPerProductionType(period=timedelta(days=30)):
"""
Cette ressource a pour objectif de permettre de récupérer la production infrajournalière réalisée agrégée par filière
Doc : https: // data.rte - france.com / catalog / - / api / user_guide / 226953
:return: json structure
"""
url = 'https://digital.iservices.rte-france.com/open_api/actual_generation/v1/actual_generations_per_production_type'
sandbox_url = 'https://digital.iservices.rte-france.com/open_api/actual_generation/v1/sandbox/actual_generations_per_production_type'
datefmt = '%Y-%m-%dT%H:%M:%S%z'
today = datetime.today()
start_date = (today-period).strftime(datefmt)
end_date = today.strftime(datefmt)
payload = '?start_date={}%2B02:00&end_date={}%2B02:00'.format(start_date, end_date)
print(payload)
try :
r = requests.get(url+payload, headers=headerWithToken())
r.raise_for_status()
except requests.exceptions.HTTPError as e :
print(e)
print('request headers : {}'.format(r.headers))
return
return r.json()
def getActualGenerationPerUnit(period=timedelta(days=7)) :
"""
Cette ressource a pour objectif de permettre de récupérer la production infrajournalièreagrégée par groupe.
Doc : https://data.rte-france.com/catalog/-/api/user_guide/226953
:return: json structure
"""
url = 'https://digital.iservices.rte-france.com/open_api/actual_generation/v1/actual_generations_per_unit'
sandbox_url = 'https://digital.iservices.rte-france.com/open_api/actual_generation/v1/sandbox/actual_generations_per_production_type'
datefmt = '%Y-%m-%dT%H:%M:%S%z'
today = date.today() - timedelta(days=1)
start_date = (today-period).strftime(datefmt)
end_date = today.strftime(datefmt)
eic_code=''
#due to the ':' character in the request parameters, we cant' use the parameters option of requests module.
payload = '?start_date={}%2B02:00&end_date={}%2B02:00'.format(start_date, end_date)
print(payload)
# payload='?start_date=2015-06-08T00:00:00%2B02:00&end_date=2015-06-11T00:00:00%2B02:00'
# print(payload)
try :
r = requests.get(url+payload, headers=headerWithToken())
r.raise_for_status()
except requests.exceptions.HTTPError as e :
print(e)
print('request headers : {}'.format(r.headers))
return
return r.json()
if __name__ == '__main__' :
RTE_TOKEN = requestToken()
period = timedelta(days=7)
# gen = getActualGenerationPerProductionType(period)
# with open('ActualGenerationPerProductionType.json','w') as outfile:
# json.dump(gen, outfile, sort_keys=True, indent=4, separators=(',', ': '), ensure_ascii=False)
gen = getActualGenerationPerUnit(period)
with open('ActualGenerationPerUnit.json','w') as outfile:
json.dump(gen, outfile, sort_keys=True, indent=4, separators=(',', ': '), ensure_ascii=False)