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
|
#!/usr/bin/env python3
""" Get energyprices from Entsoe """
import os
import sys
from datetime import datetime, timedelta
import common
import requests
import xmltodict
from dateutil import tz
from tzlocal import get_localzone
# variables
# Getting an api-key isn't very well documented. The documentation [1] points
# to a pdf [2], which says the following:
# > In order to request the access to the Restful API, please register on the
# > Transparency Platform and send an email to transparency@entsoe.eu with
# > “Restful API access” in the subject line. Indicate the email address you
# > entered during registration in the email body. We will make our best to
# > respond to your request.
# 1: https://transparency.entsoe.eu/content/static_content/Static%20content/web%20api/Guide.html#_authentication_and_authorisation
# 2: https://transparency.entsoe.eu/content/static_content/download?path=/Static%20content/API-Token-Management.pdf
apiKey = os.environ["el_entsoe_token"]
# https://transparency.entsoe.eu/content/static_content/Static%20content/web%20api/Guide.html
apiUrl = "https://web-api.tp.entsoe.eu/api?securityToken=" + apiKey
startTime = datetime.now(get_localzone()) - timedelta(days=7)
startTime = startTime.strftime("%Y%m%d")
# startTime = '20230523' # <- edit for manual starttime. Like when filling in missing info.
endTime = datetime.now(get_localzone()) + timedelta(days=1)
endTime = endTime.strftime("%Y%m%d")
jobname = os.path.splitext(os.path.basename(__file__))[0]
# https://transparency.entsoe.eu/content/static_content/Static%20content/web%20api/Guide.html#_areas
areas = [
{"name": "NO-0", "code": "10YNO-0--------C"},
{"name": "NO-1", "code": "10YNO-1--------2"},
{"name": "NO-2", "code": "10YNO-2--------T"},
{"name": "NO-3", "code": "10YNO-3--------J"},
{"name": "NO-4", "code": "10YNO-4--------9"},
]
UTC = tz.gettz("UTC")
CET = tz.gettz("Europe/Oslo")
# Get the data
values = []
for area in areas:
try:
url = (
apiUrl
+ "&documentType=A44&in_Domain="
+ area["code"]
+ "&out_Domain="
+ area["code"]
+ "&periodStart="
+ startTime
+ "0000&periodEnd="
+ endTime
+ "0000"
)
print("Getting data for " + area["code"])
response = requests.get(url, timeout=10)
if response.status_code != 200:
print(response.status_code)
print("Oh shit")
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("oh lol")
sys.exit(e)
data_dict = xmltodict.parse(response.content)
items = 0
if "Publication_MarketDocument" in data_dict:
for lista in data_dict["Publication_MarketDocument"]["TimeSeries"]:
utctime = datetime.strptime(
lista["Period"]["timeInterval"]["start"], "%Y-%m-%dT%H:%MZ"
)
utctime = utctime.replace(tzinfo=UTC)
cettime = utctime.astimezone(CET)
items += len(lista["Period"]["Point"])
for item in lista["Period"]["Point"]:
# the response contains timerange, but not timestamp for every price, so we must calculate it
time = str(cettime + timedelta(hours=int(item["position"]) - 1))
# append values
values.append((time, area["name"], item["price.amount"]))
print("Got " + str(items) + " records")
# SQL
sql = """ INSERT INTO entsoe
VALUES(%s, %s, %s)
ON CONFLICT (starttime, zone) DO NOTHING"""
common.dbi(sql, values, verbose=True)
|
