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
|
#!/usr/bin/env python3
""" Get energyprices from Entsoe """
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 = common.env("el_entsoe_token")
# https://transparency.entsoe.eu/content/static_content/Static%20content/web%20api/Guide.html
apiUrl = "https://web-api.tp.entsoe.eu/api"
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")
# 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")
# Create a session object for making requests
session = requests.Session()
# Get the data
values = []
for area in areas:
try:
url = (
f"{apiUrl}"
f"?securityToken={apiKey}"
f"&documentType=A44"
f"&in_Domain={area['code']}"
f"&out_Domain={area['code']}"
f"&periodStart={startTime}0000"
f"&periodEnd={endTime}0000"
)
print("Getting data for " + area["code"])
# Use the session object to make the request
response = session.get(url, timeout=10)
# Handle any exceptions or non-200 status code
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Error: {e}")
sys.exit(1)
data_dict = xmltodict.parse(response.content)
items = 0
if "Publication_MarketDocument" in data_dict:
for lista in data_dict["Publication_MarketDocument"]["TimeSeries"]["Period"]:
utctime = datetime.strptime(
lista["timeInterval"]["start"], "%Y-%m-%dT%H:%MZ"
)
utctime = utctime.replace(tzinfo=UTC)
cettime = utctime.astimezone(CET)
items += len(lista["Point"])
for item in lista["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(f"Got {items} records")
# SQL
sql = """ INSERT INTO entsoe
VALUES(%s, %s, %s)
ON CONFLICT (starttime, zone) DO NOTHING"""
common.dbi(sql, values, verbose=True)
|
