Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-04-19 08:55:15 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Strompreise aus URL möglich, Source unsere API + pers. Cache

Browse Source
This commit is contained in:
Bla Bla 2024-02-25 15:32:43 +01:00
parent 270eca6104
commit 33a6d59bcb
2 changed files with 160 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

220
modules/class_strompreis.py
View File

@ -1,82 +1,150 @@
import json import json
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import numpy as np import numpy as np
import json, os
from datetime import datetime
import hashlib, requests
class HourlyElectricityPriceForecast: class HourlyElectricityPriceForecast:
class PriceData: def __init__(self, source, cache_dir='cache'):
def __init__(self, total, energy, tax, starts_at, currency, level): self.cache_dir = cache_dir
self.total = total/1000.0 if not os.path.exists(self.cache_dir):
self.energy = energy/1000.0 os.makedirs(self.cache_dir)
self.tax = tax/1000.0 self.prices = self.load_data(source)
self.starts_at = datetime.strptime(starts_at, '%Y-%m-%dT%H:%M:%S.%f%z')
self.currency = currency
self.level = level
# Getter-Methoden
def get_total(self):
return self.total
def get_energy(self):
return self.energy
def get_tax(self):
return self.tax
def get_starts_at(self):
return self.starts_at
def get_currency(self):
return self.currency
def get_level(self):
return self.level
def __init__(self, filepath):
self.filepath = filepath
self.price_data = []
self.load_data()
def get_prices_for_date(self, query_date):
query_date = datetime.strptime(query_date, '%Y-%m-%d').date()
prices_for_date = [price.get_total() for price in self.price_data if price.starts_at.date() == query_date]
return np.array(prices_for_date)
def get_price_for_datetime(self, query_datetime):
query_datetime = datetime.strptime(query_datetime, '%Y-%m-%d %H').replace(minute=0, second=0, microsecond=0)
query_datetime = query_datetime.replace(tzinfo=timezone(timedelta(hours=1)))
for price in self.price_data:
#print(price.starts_at.replace(minute=0, second=0, microsecond=0) , " ", query_datetime, " == ",price.starts_at.replace(minute=0, second=0, microsecond=0) == query_datetime)
if price.starts_at.replace(minute=0, second=0, microsecond=0) == query_datetime:
return np.array(price)
return None
def load_data(self):
with open(self.filepath, 'r') as file:
data = json.load(file)
for item in data['payload']:
self.price_data.append(self.PriceData(
total=item['total'],
energy=item['energy'],
tax=item['tax'],
starts_at=item['startsAt'],
currency=item['currency'],
level=item['level']
))
def get_price_data(self):
return self.price_data
# Beispiel für die Verwendung der Klasse
if __name__ == '__main__':
filepath = r'..\test_data\strompreis.json' # Pfad zur JSON-Datei anpassen
price_forecast = HourlyElectricityPriceForecast(filepath)
specific_date_prices = price_forecast.get_prices_for_date('2024-02-16') # Datum anpassen
specific_date_prices = price_forecast.get_price_for_datetime('2024-02-16 12') def load_data(self, source):
print(specific_date_prices) if source.startswith('http'):
#for price in price_forecast.get_price_data(): cache_filename = self.get_cache_filename(source)
# print(price.get_starts_at(), price.get_total(), price.get_currency()) if os.path.exists(cache_filename):
print("Lade Daten aus dem Cache...")
with open(cache_filename, 'r') as file:
data = json.load(file)
else:
print("Lade Daten von der URL...")
response = requests.get(source)
if response.status_code == 200:
data = response.json()
with open(cache_filename, 'w') as file:
json.dump(data, file)
else:
raise Exception(f"Fehler beim Abrufen der Daten: {response.status_code}")
else:
with open(source, 'r') as file:
data = json.load(file)
return data['values']
def get_cache_filename(self, url):
hash_object = hashlib.sha256(url.encode())
hex_dig = hash_object.hexdigest()
return os.path.join(self.cache_dir, f"cache_{hex_dig}.json")
def get_price_for_date(self, date_str):
"""Gibt alle Preise für das spezifizierte Datum zurück."""
date_prices = [entry["marketpriceEurocentPerKWh"] for entry in self.prices if date_str in entry['start']]
return date_prices
# def get_price_for_hour(self, datetime_str):
# """Gibt den Preis für die spezifizierte Stunde zurück."""
# hour_price = [entry for entry in self.prices if datetime_str in entry['start']]
# return hour_price[0] if hour_price else None
# # Beispiel zur Verwendung der Klasse
# filepath = '/mnt/data/strompreise_akkudokAPI.json' # Pfad zur JSON-Datei
# strompreise = Strompreise(filepath)
# # Preise für ein spezifisches Datum erhalten
# date_str = '2024-02-25'
# prices_for_date = strompreise.get_price_for_date(date_str)
# print(f"Preise für {date_str}: {prices_for_date}")
# # Preis für eine spezifische Stunde erhalten
# datetime_str = '2024-02-25T15:00:00.000Z'
# price_for_hour = strompreise.get_price_for_hour(datetime_str)
# print(f"Preis für {datetime_str}: {price_for_hour}")
# class HourlyElectricityPriceForecast:
# class PriceData:
# def __init__(self, total, energy, tax, starts_at, currency, level):
# self.total = total/1000.0
# self.energy = energy/1000.0
# self.tax = tax/1000.0
# self.starts_at = datetime.strptime(starts_at, '%Y-%m-%dT%H:%M:%S.%f%z')
# self.currency = currency
# self.level = level
# # Getter-Methoden
# def get_total(self):
# return self.total
# def get_energy(self):
# return self.energy
# def get_tax(self):
# return self.tax
# def get_starts_at(self):
# return self.starts_at
# def get_currency(self):
# return self.currency
# def get_level(self):
# return self.level
# def __init__(self, filepath):
# self.filepath = filepath
# self.price_data = []
# self.load_data()
# def get_prices_for_date(self, query_date):
# query_date = datetime.strptime(query_date, '%Y-%m-%d').date()
# prices_for_date = [price.get_total() for price in self.price_data if price.starts_at.date() == query_date]
# return np.array(prices_for_date)
# def get_price_for_datetime(self, query_datetime):
# query_datetime = datetime.strptime(query_datetime, '%Y-%m-%d %H').replace(minute=0, second=0, microsecond=0)
# query_datetime = query_datetime.replace(tzinfo=timezone(timedelta(hours=1)))
# for price in self.price_data:
# #print(price.starts_at.replace(minute=0, second=0, microsecond=0) , " ", query_datetime, " == ",price.starts_at.replace(minute=0, second=0, microsecond=0) == query_datetime)
# if price.starts_at.replace(minute=0, second=0, microsecond=0) == query_datetime:
# return np.array(price)
# return None
# def load_data(self):
# with open(self.filepath, 'r') as file:
# data = json.load(file)
# for item in data['payload']:
# self.price_data.append(self.PriceData(
# total=item['total'],
# energy=item['energy'],
# tax=item['tax'],
# starts_at=item['startsAt'],
# currency=item['currency'],
# level=item['level']
# ))
# def get_price_data(self):
# return self.price_data
# # Beispiel für die Verwendung der Klasse
# if __name__ == '__main__':
# filepath = r'..\test_data\strompreis.json' # Pfad zur JSON-Datei anpassen
# price_forecast = HourlyElectricityPriceForecast(filepath)
# specific_date_prices = price_forecast.get_prices_for_date('2024-02-16') # Datum anpassen
# specific_date_prices = price_forecast.get_price_for_datetime('2024-02-16 12')
# print(specific_date_prices)
# #for price in price_forecast.get_price_data():
# # print(price.get_starts_at(), price.get_total(), price.get_currency())

22
test.py
View File

@ -28,22 +28,32 @@ akku = PVAkku(akku_size)
discharge_array = np.full(24,1) discharge_array = np.full(24,1)
# Load Forecast # Load Forecast
###############
lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy)
leistung_haushalt = lf.get_daily_stats(date)[0,...] # Datum anpassen leistung_haushalt = lf.get_daily_stats(date)[0,...] # Datum anpassen
pprint(leistung_haushalt.shape) pprint(leistung_haushalt.shape)
# PV Forecast # PV Forecast
###############
#PVforecast = PVForecast(filepath=os.path.join(r'test_data', r'pvprognose.json')) #PVforecast = PVForecast(filepath=os.path.join(r'test_data', r'pvprognose.json'))
PVforecast = PVForecast(url="https://api.akkudoktor.net/forecast?lat=50.8588&lon=7.3747&power=5400&azimuth=-10&tilt=7&powerInvertor=2500&horizont=20,40,30,30&power=4800&azimuth=-90&tilt=7&powerInvertor=2500&horizont=20,40,45,50&power=1480&azimuth=-90&tilt=70&powerInvertor=1120&horizont=60,45,30,70&power=1600&azimuth=5&tilt=60&powerInvertor=1200&horizont=60,45,30,70&past_days=5&cellCoEff=-0.36&inverterEfficiency=0.8&albedo=0.25&timezone=Europe%2FBerlin&hourly=relativehumidity_2m%2Cwindspeed_10m") PVforecast = PVForecast(url="https://api.akkudoktor.net/forecast?lat=50.8588&lon=7.3747&power=5400&azimuth=-10&tilt=7&powerInvertor=2500&horizont=20,40,30,30&power=4800&azimuth=-90&tilt=7&powerInvertor=2500&horizont=20,40,45,50&power=1480&azimuth=-90&tilt=70&powerInvertor=1120&horizont=60,45,30,70&power=1600&azimuth=5&tilt=60&powerInvertor=1200&horizont=60,45,30,70&past_days=5&cellCoEff=-0.36&inverterEfficiency=0.8&albedo=0.25&timezone=Europe%2FBerlin&hourly=relativehumidity_2m%2Cwindspeed_10m")
pv_forecast = PVforecast.get_forecast_for_date(date) pv_forecast = PVforecast.get_forecast_for_date(date)
temperature_forecast = PVforecast.get_temperature_forecast_for_date(date) temperature_forecast = PVforecast.get_temperature_forecast_for_date(date)
pprint(pv_forecast) pprint(pv_forecast.shape)
sys.exit()
# Strompreise # Strompreise
filepath = os.path.join (r'test_data', r'strompreis.json') # Pfad zur JSON-Datei anpassen ###############
price_forecast = HourlyElectricityPriceForecast(filepath) filepath = os.path.join (r'test_data', r'strompreise_akkudokAPI.json') # Pfad zur JSON-Datei anpassen
specific_date_prices = price_forecast.get_prices_for_date(date) #price_forecast = HourlyElectricityPriceForecast(source=filepath)
price_forecast = HourlyElectricityPriceForecast(source="https://api.akkudoktor.net/prices?start="+date+"&end="+date+"")
specific_date_prices = price_forecast.get_price_for_date(date)
pprint(f"Preise für {date}: {specific_date_prices}")
# WP # WP
leistung_wp = wp.simulate_24h(temperature_forecast) leistung_wp = wp.simulate_24h(temperature_forecast)