mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2025-06-28 00:46:53 +00:00
Update flask_server.py
first step of cleanup. only formatting and very minor code changes To Do: - implement proper config - move parameters into config - remove not needed commented code (idk which is useful) - logic in has_no_empty_params is unnecessary complicated
This commit is contained in:
parent
af7b745d15
commit
29315427f0
334
flask_server.py
334
flask_server.py
@ -31,118 +31,95 @@ app = Flask(__name__)
|
|||||||
|
|
||||||
opt_class = optimization_problem(prediction_hours=prediction_hours, strafe=10, optimization_hours=optimization_hours)
|
opt_class = optimization_problem(prediction_hours=prediction_hours, strafe=10, optimization_hours=optimization_hours)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
# @app.route('/last_correction', methods=['GET'])
|
# @app.route('/last_correction', methods=['GET'])
|
||||||
# def flask_last_correction():
|
# def flask_last_correction():
|
||||||
# if request.method == 'GET':
|
# if request.method == 'GET':
|
||||||
# year_energy = float(request.args.get("year_energy"))
|
# year_energy = float(request.args.get("year_energy"))
|
||||||
# date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date())
|
# date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date())
|
||||||
# ###############
|
# ###############
|
||||||
# # 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_stats_for_date_range(date_now, date)[0] # Only the expected value!
|
||||||
# leistung_haushalt = lf.get_stats_for_date_range(date_now,date)[0] # Nur Erwartungswert!
|
#
|
||||||
|
# gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
||||||
# gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt)
|
||||||
# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt)
|
# # ###############
|
||||||
|
# # Heat Pump (WP)
|
||||||
# # ###############
|
# # ##############
|
||||||
# # # WP
|
# # leistung_wp = wp.simulate_24h(temperature_forecast)
|
||||||
# # ##############
|
# # gesamtlast.hinzufuegen("Heatpump", leistung_wp)
|
||||||
# # leistung_wp = wp.simulate_24h(temperature_forecast)
|
# last = gesamtlast.gesamtlast_berechnen()
|
||||||
# # gesamtlast.hinzufuegen("Heatpump", leistung_wp)
|
# print(last)
|
||||||
|
# return jsonify(last.tolist())
|
||||||
# last = gesamtlast.gesamtlast_berechnen()
|
|
||||||
# print(last)
|
|
||||||
# #print(specific_date_prices)
|
|
||||||
# return jsonify(last.tolist())
|
|
||||||
|
|
||||||
|
|
||||||
@app.route('/soc', methods=['GET'])
|
@app.route('/soc', methods=['GET'])
|
||||||
def flask_soc():
|
def flask_soc():
|
||||||
|
# MariaDB connection details
|
||||||
# MariaDB Verbindungsdetails
|
|
||||||
config = db_config
|
config = db_config
|
||||||
|
|
||||||
# Parameter festlegen
|
# Set parameters for SOC (State of Charge) calculation
|
||||||
voltage_high_threshold = 55.4 # 100% SoC
|
voltage_high_threshold = 55.4 # 100% SoC
|
||||||
voltage_low_threshold = 46.5 # 0% SoC
|
voltage_low_threshold = 46.5 # 0% SoC
|
||||||
current_low_threshold = 2 # Niedriger Strom für beide Zustände
|
current_low_threshold = 2 # Low current threshold for both states
|
||||||
gap = 30 # Zeitlücke in Minuten zum Gruppieren von Maxima/Minima
|
gap = 30 # Time gap in minutes for grouping maxima/minima
|
||||||
bat_capacity = 33 * 1000 / 48
|
bat_capacity = 33 * 1000 / 48 # Battery capacity in watt-hours
|
||||||
|
|
||||||
# Zeitpunkt X definieren
|
# Define the reference time point (3 weeks ago)
|
||||||
zeitpunkt_x = (datetime.now() - timedelta(weeks=3)).strftime('%Y-%m-%d %H:%M:%S')
|
zeitpunkt_x = (datetime.now() - timedelta(weeks=3)).strftime('%Y-%m-%d %H:%M:%S')
|
||||||
|
|
||||||
|
# Instantiate BatteryDataProcessor and perform calculations
|
||||||
# BatteryDataProcessor instanziieren und verwenden
|
processor = BatteryDataProcessor(config, voltage_high_threshold, voltage_low_threshold, current_low_threshold, gap, bat_capacity)
|
||||||
processor = BatteryDataProcessor(config, voltage_high_threshold, voltage_low_threshold, current_low_threshold, gap,bat_capacity)
|
|
||||||
processor.connect_db()
|
processor.connect_db()
|
||||||
processor.fetch_data(zeitpunkt_x)
|
processor.fetch_data(zeitpunkt_x)
|
||||||
processor.process_data()
|
processor.process_data()
|
||||||
last_points_100_df, last_points_0_df = processor.find_soc_points()
|
last_points_100_df, last_points_0_df = processor.find_soc_points()
|
||||||
soc_df, integration_results = processor.calculate_resetting_soc(last_points_100_df, last_points_0_df)
|
soc_df, integration_results = processor.calculate_resetting_soc(last_points_100_df, last_points_0_df)
|
||||||
#soh_df = processor.calculate_soh(integration_results)
|
# soh_df = processor.calculate_soh(integration_results) # Optional State of Health calculation
|
||||||
processor.update_database_with_soc(soc_df)
|
processor.update_database_with_soc(soc_df) # Update database with SOC data
|
||||||
#processor.plot_data(last_points_100_df, last_points_0_df, soc_df)
|
# processor.plot_data(last_points_100_df, last_points_0_df, soc_df) # Optional data visualization
|
||||||
processor.disconnect_db()
|
processor.disconnect_db() # Disconnect from the database
|
||||||
|
|
||||||
return jsonify("Done")
|
return jsonify("Done")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@app.route('/strompreis', methods=['GET'])
|
@app.route('/strompreis', methods=['GET'])
|
||||||
def flask_strompreis():
|
def flask_strompreis():
|
||||||
date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date())
|
# Get the current date and the end date based on prediction hours
|
||||||
filepath = os.path.join (r'test_data', r'strompreise_akkudokAPI.json') # Pfad zur JSON-Datei anpassen
|
date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date())
|
||||||
#price_forecast = HourlyElectricityPriceForecast(source=filepath)
|
filepath = os.path.join(r'test_data', r'strompreise_akkudokAPI.json') # Adjust the path to the JSON file
|
||||||
price_forecast = HourlyElectricityPriceForecast(source="https://api.akkudoktor.net/prices?start="+date_now+"&end="+date+"", prediction_hours=prediction_hours)
|
price_forecast = HourlyElectricityPriceForecast(source=f"https://api.akkudoktor.net/prices?start={date_now}&end={date}", prediction_hours=prediction_hours)
|
||||||
specific_date_prices = price_forecast.get_price_for_daterange(date_now,date)
|
specific_date_prices = price_forecast.get_price_for_daterange(date_now, date) # Fetch prices for the specified date range
|
||||||
#print(specific_date_prices)
|
|
||||||
return jsonify(specific_date_prices.tolist())
|
return jsonify(specific_date_prices.tolist())
|
||||||
|
|
||||||
|
# Endpoint to handle total load calculation based on the latest measured data
|
||||||
|
|
||||||
# Die letzten X gemessenen Daten + gesamtlast Simple oder eine andere Schätung als Input
|
|
||||||
# Daraus wird dann eine neuen Lastprognose erstellt welche korrigiert ist.
|
|
||||||
# Input:
|
|
||||||
@app.route('/gesamtlast', methods=['POST'])
|
@app.route('/gesamtlast', methods=['POST'])
|
||||||
def flask_gesamtlast():
|
def flask_gesamtlast():
|
||||||
# Daten aus dem JSON-Body abrufen
|
# Retrieve data from the JSON body
|
||||||
data = request.get_json()
|
data = request.get_json()
|
||||||
|
|
||||||
# Extract year_energy and prediction_hours from the request JSON
|
# Extract year_energy and prediction_hours from the request JSON
|
||||||
year_energy = float(data.get("year_energy"))
|
year_energy = float(data.get("year_energy"))
|
||||||
prediction_hours = int(data.get("hours", 48)) # Default to 48 hours if not specified
|
prediction_hours = int(data.get("hours", 48)) # Default to 48 hours if not specified
|
||||||
|
|
||||||
# Measured data as JSON
|
# Measured data in JSON format
|
||||||
measured_data_json = data.get("measured_data")
|
measured_data_json = data.get("measured_data")
|
||||||
|
|
||||||
# Convert JSON data into a Pandas DataFrame
|
|
||||||
measured_data = pd.DataFrame(measured_data_json)
|
measured_data = pd.DataFrame(measured_data_json)
|
||||||
# Make sure the 'time' column is in datetime format
|
|
||||||
measured_data['time'] = pd.to_datetime(measured_data['time'])
|
measured_data['time'] = pd.to_datetime(measured_data['time'])
|
||||||
|
|
||||||
# Check if the datetime has timezone info, if not, assume it's local time
|
# Ensure datetime has timezone info for accurate calculations
|
||||||
if measured_data['time'].dt.tz is None:
|
if measured_data['time'].dt.tz is None:
|
||||||
# Treat it as local time and localize it
|
|
||||||
measured_data['time'] = measured_data['time'].dt.tz_localize('Europe/Berlin')
|
measured_data['time'] = measured_data['time'].dt.tz_localize('Europe/Berlin')
|
||||||
else:
|
else:
|
||||||
# Convert the time to local timezone (e.g., 'Europe/Berlin')
|
|
||||||
measured_data['time'] = measured_data['time'].dt.tz_convert('Europe/Berlin')
|
measured_data['time'] = measured_data['time'].dt.tz_convert('Europe/Berlin')
|
||||||
|
|
||||||
# Remove timezone info after conversion
|
# Remove timezone info after conversion to simplify further processing
|
||||||
measured_data['time'] = measured_data['time'].dt.tz_localize(None)
|
measured_data['time'] = measured_data['time'].dt.tz_localize(None)
|
||||||
|
|
||||||
# Instantiate LoadForecast and generate forecast data
|
# Instantiate LoadForecast and generate forecast data
|
||||||
lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy)
|
lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy)
|
||||||
|
|
||||||
# Generate forecast data based on the measured data time range
|
|
||||||
forecast_list = []
|
forecast_list = []
|
||||||
|
|
||||||
|
# Generate daily forecasts for the date range based on measured data
|
||||||
for single_date in pd.date_range(measured_data['time'].min().date(), measured_data['time'].max().date()):
|
for single_date in pd.date_range(measured_data['time'].min().date(), measured_data['time'].max().date()):
|
||||||
date_str = single_date.strftime('%Y-%m-%d')
|
date_str = single_date.strftime('%Y-%m-%d')
|
||||||
daily_forecast = lf.get_daily_stats(date_str)
|
daily_forecast = lf.get_daily_stats(date_str)
|
||||||
@ -153,207 +130,196 @@ def flask_gesamtlast():
|
|||||||
|
|
||||||
# Concatenate all daily forecasts into a single DataFrame
|
# Concatenate all daily forecasts into a single DataFrame
|
||||||
predicted_data = pd.concat(forecast_list, ignore_index=True)
|
predicted_data = pd.concat(forecast_list, ignore_index=True)
|
||||||
#print(predicted_data)
|
|
||||||
# Create LoadPredictionAdjuster instance
|
# Create LoadPredictionAdjuster instance to adjust the predictions based on measured data
|
||||||
adjuster = LoadPredictionAdjuster(measured_data, predicted_data, lf)
|
adjuster = LoadPredictionAdjuster(measured_data, predicted_data, lf)
|
||||||
|
adjuster.calculate_weighted_mean() # Calculate weighted mean for adjustment
|
||||||
|
adjuster.adjust_predictions() # Adjust predictions based on measured data
|
||||||
|
future_predictions = adjuster.predict_next_hours(prediction_hours) # Predict future load
|
||||||
|
|
||||||
# Calculate weighted mean and adjust predictions
|
# Extract household power predictions
|
||||||
adjuster.calculate_weighted_mean()
|
|
||||||
adjuster.adjust_predictions()
|
|
||||||
|
|
||||||
# Predict the next x hours
|
|
||||||
future_predictions = adjuster.predict_next_hours(prediction_hours)
|
|
||||||
|
|
||||||
# Extract the household power predictions
|
|
||||||
leistung_haushalt = future_predictions['Adjusted Pred'].values
|
leistung_haushalt = future_predictions['Adjusted Pred'].values
|
||||||
|
|
||||||
# Instantiate Gesamtlast and add household power predictions
|
|
||||||
gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
||||||
gesamtlast.hinzufuegen("Haushalt", leistung_haushalt)
|
gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation
|
||||||
|
|
||||||
# ###############
|
|
||||||
# # WP (optional)
|
|
||||||
# ###############
|
|
||||||
# leistung_wp = wp.simulate_24h(temperature_forecast)
|
|
||||||
# gesamtlast.hinzufuegen("Heatpump", leistung_wp)
|
|
||||||
|
|
||||||
# Calculate the total load
|
# Calculate the total load
|
||||||
last = gesamtlast.gesamtlast_berechnen()
|
last = gesamtlast.gesamtlast_berechnen() # Compute total load
|
||||||
|
|
||||||
# Return the calculated load as JSON
|
|
||||||
return jsonify(last.tolist())
|
return jsonify(last.tolist())
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
# @app.route('/gesamtlast', methods=['GET'])
|
# @app.route('/gesamtlast', methods=['GET'])
|
||||||
# def flask_gesamtlast():
|
# def flask_gesamtlast():
|
||||||
# if request.method == 'GET':
|
# if request.method == 'GET':
|
||||||
# year_energy = float(request.args.get("year_energy"))
|
# year_energy = float(request.args.get("year_energy")) # Get annual energy value from query parameters
|
||||||
# prediction_hours = int(request.args.get("hours", 48)) # Default to 24 hours if not specified
|
# prediction_hours = int(request.args.get("hours", 48)) # Default to 48 hours if not specified
|
||||||
# date_now = datetime.now()
|
# date_now = datetime.now() # Get the current date and time
|
||||||
# end_date = (date_now + timedelta(hours=prediction_hours)).strftime('%Y-%m-%d %H:%M:%S')
|
# end_date = (date_now + timedelta(hours=prediction_hours)).strftime('%Y-%m-%d %H:%M:%S') # Calculate end date based on prediction hours
|
||||||
|
|
||||||
# ###############
|
# ###############
|
||||||
# # Load Forecast
|
# # Load Forecast
|
||||||
# ###############
|
# ###############
|
||||||
# # Instantiate LastEstimator and get measured data
|
# # Instantiate LastEstimator to retrieve measured data
|
||||||
# estimator = LastEstimator()
|
# estimator = LastEstimator()
|
||||||
# start_date = (date_now - timedelta(days=60)).strftime('%Y-%m-%d') # Example: last 60 days
|
# start_date = (date_now - timedelta(days=60)).strftime('%Y-%m-%d') # Start date: last 60 days
|
||||||
# end_date = date_now.strftime('%Y-%m-%d') # Current date
|
# end_date = date_now.strftime('%Y-%m-%d') # Current date
|
||||||
|
|
||||||
# last_df = estimator.get_last(start_date, end_date)
|
# last_df = estimator.get_last(start_date, end_date) # Get last load data
|
||||||
|
|
||||||
# selected_columns = last_df[['timestamp', 'Last']]
|
# selected_columns = last_df[['timestamp', 'Last']] # Select relevant columns
|
||||||
# selected_columns['time'] = pd.to_datetime(selected_columns['timestamp']).dt.floor('H')
|
# selected_columns['time'] = pd.to_datetime(selected_columns['timestamp']).dt.floor('H') # Floor timestamps to the nearest hour
|
||||||
# selected_columns['Last'] = pd.to_numeric(selected_columns['Last'], errors='coerce')
|
# selected_columns['Last'] = pd.to_numeric(selected_columns['Last'], errors='coerce') # Convert 'Last' to numeric, coerce errors
|
||||||
# cleaned_data = selected_columns.dropna()
|
# cleaned_data = selected_columns.dropna() # Clean data by dropping NaN values
|
||||||
|
|
||||||
# # Instantiate LoadForecast
|
# # Instantiate LoadForecast
|
||||||
# lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy)
|
# lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy)
|
||||||
|
|
||||||
# # Generate forecast data
|
# # Generate forecast data
|
||||||
# forecast_list = []
|
# forecast_list = [] # List to hold daily forecasts
|
||||||
# for single_date in pd.date_range(cleaned_data['time'].min().date(), cleaned_data['time'].max().date()):
|
# for single_date in pd.date_range(cleaned_data['time'].min().date(), cleaned_data['time'].max().date()): # Iterate over date range
|
||||||
# date_str = single_date.strftime('%Y-%m-%d')
|
# date_str = single_date.strftime('%Y-%m-%d') # Format date
|
||||||
# daily_forecast = lf.get_daily_stats(date_str)
|
# daily_forecast = lf.get_daily_stats(date_str) # Get daily stats from LoadForecast
|
||||||
# mean_values = daily_forecast[0]
|
# mean_values = daily_forecast[0] # Extract mean values
|
||||||
# hours = [single_date + pd.Timedelta(hours=i) for i in range(24)]
|
# hours = [single_date + pd.Timedelta(hours=i) for i in range(24)] # Generate hours for the day
|
||||||
# daily_forecast_df = pd.DataFrame({'time': hours, 'Last Pred': mean_values})
|
# daily_forecast_df = pd.DataFrame({'time': hours, 'Last Pred': mean_values}) # Create DataFrame for daily forecast
|
||||||
# forecast_list.append(daily_forecast_df)
|
# forecast_list.append(daily_forecast_df) # Append to the list
|
||||||
|
|
||||||
# forecast_df = pd.concat(forecast_list, ignore_index=True)
|
# forecast_df = pd.concat(forecast_list, ignore_index=True) # Concatenate all daily forecasts
|
||||||
|
|
||||||
# # Create LoadPredictionAdjuster instance
|
# # Create LoadPredictionAdjuster instance
|
||||||
# adjuster = LoadPredictionAdjuster(cleaned_data, forecast_df, lf)
|
# adjuster = LoadPredictionAdjuster(cleaned_data, forecast_df, lf)
|
||||||
# adjuster.calculate_weighted_mean()
|
# adjuster.calculate_weighted_mean() # Calculate weighted mean for adjustments
|
||||||
# adjuster.adjust_predictions()
|
# adjuster.adjust_predictions() # Adjust predictions based on measured data
|
||||||
|
|
||||||
# # Predict the next hours
|
# # Predict the next hours
|
||||||
# future_predictions = adjuster.predict_next_hours(prediction_hours)
|
# future_predictions = adjuster.predict_next_hours(prediction_hours) # Predict future load
|
||||||
|
|
||||||
# leistung_haushalt = future_predictions['Adjusted Pred'].values
|
# leistung_haushalt = future_predictions['Adjusted Pred'].values # Extract household power predictions
|
||||||
|
|
||||||
# gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
# gesamtlast = Gesamtlast(prediction_hours=prediction_hours) # Create Gesamtlast instance
|
||||||
# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt)
|
# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation
|
||||||
|
|
||||||
# # ###############
|
# # ###############
|
||||||
# # # WP
|
# # # WP (Heat Pump)
|
||||||
# # ##############
|
# # ##############
|
||||||
# # leistung_wp = wp.simulate_24h(temperature_forecast)
|
# # leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours
|
||||||
# # gesamtlast.hinzufuegen("Heatpump", leistung_wp)
|
# # gesamtlast.hinzufuegen("Heatpump", leistung_wp) # Add heat pump load to total load calculation
|
||||||
|
|
||||||
# last = gesamtlast.gesamtlast_berechnen()
|
|
||||||
# print(last)
|
|
||||||
# return jsonify(last.tolist())
|
|
||||||
|
|
||||||
|
# last = gesamtlast.gesamtlast_berechnen() # Calculate total load
|
||||||
|
# print(last) # Output total load
|
||||||
|
# return jsonify(last.tolist()) # Return total load as JSON
|
||||||
|
|
||||||
@app.route('/gesamtlast_simple', methods=['GET'])
|
@app.route('/gesamtlast_simple', methods=['GET'])
|
||||||
def flask_gesamtlast_simple():
|
def flask_gesamtlast_simple():
|
||||||
if request.method == 'GET':
|
if request.method == 'GET':
|
||||||
year_energy = float(request.args.get("year_energy"))
|
year_energy = float(request.args.get("year_energy")) # Get annual energy value from query parameters
|
||||||
date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date())
|
date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date()) # Get the current date and prediction end date
|
||||||
|
|
||||||
###############
|
###############
|
||||||
# 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) # Instantiate LoadForecast with specified parameters
|
||||||
#leistung_haushalt = lf.get_daily_stats(date)[0,...] # Datum anpassen
|
leistung_haushalt = lf.get_stats_for_date_range(date_now, date)[0] # Get expected household load for the date range
|
||||||
leistung_haushalt = lf.get_stats_for_date_range(date_now,date)[0] # Nur Erwartungswert!
|
|
||||||
|
|
||||||
gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
|
gesamtlast = Gesamtlast(prediction_hours=prediction_hours) # Create Gesamtlast instance
|
||||||
gesamtlast.hinzufuegen("Haushalt", leistung_haushalt)
|
gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation
|
||||||
|
|
||||||
# ###############
|
# ###############
|
||||||
# # WP
|
# # WP (Heat Pump)
|
||||||
# ##############
|
# ##############
|
||||||
# leistung_wp = wp.simulate_24h(temperature_forecast)
|
# leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours
|
||||||
# gesamtlast.hinzufuegen("Heatpump", leistung_wp)
|
# gesamtlast.hinzufuegen("Heatpump", leistung_wp) # Add heat pump load to total load calculation
|
||||||
|
|
||||||
|
last = gesamtlast.gesamtlast_berechnen() # Calculate total load
|
||||||
|
print(last) # Output total load
|
||||||
|
return jsonify(last.tolist()) # Return total load as JSON
|
||||||
|
|
||||||
last = gesamtlast.gesamtlast_berechnen()
|
|
||||||
print(last)
|
|
||||||
#print(specific_date_prices)
|
|
||||||
return jsonify(last.tolist())
|
|
||||||
|
|
||||||
@app.route('/pvforecast', methods=['GET'])
|
@app.route('/pvforecast', methods=['GET'])
|
||||||
def flask_pvprognose():
|
def flask_pvprognose():
|
||||||
if request.method == 'GET':
|
if request.method == 'GET':
|
||||||
|
# Retrieve URL and AC power measurement from query parameters
|
||||||
url = request.args.get("url")
|
url = request.args.get("url")
|
||||||
ac_power_measurement = request.args.get("ac_power_measurement")
|
ac_power_measurement = request.args.get("ac_power_measurement")
|
||||||
date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date())
|
date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date())
|
||||||
|
|
||||||
###############
|
###############
|
||||||
# PV Forecast
|
# PV Forecast
|
||||||
###############
|
###############
|
||||||
PVforecast = PVForecast(prediction_hours = prediction_hours, url=url)
|
PVforecast = PVForecast(prediction_hours=prediction_hours, url=url) # Instantiate PVForecast with given parameters
|
||||||
#print("PVPOWER",parameter['pvpowernow'])
|
if isfloat(ac_power_measurement): # Check if the AC power measurement is a valid float
|
||||||
if isfloat(ac_power_measurement):
|
PVforecast.update_ac_power_measurement(date_time=datetime.now(), ac_power_measurement=float(ac_power_measurement)) # Update measurement
|
||||||
PVforecast.update_ac_power_measurement(date_time=datetime.now(), ac_power_measurement=float(ac_power_measurement) )
|
|
||||||
#PVforecast.print_ac_power_and_measurement()
|
|
||||||
|
|
||||||
pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now,date) #get_forecast_for_date(date)
|
# Get PV forecast and temperature forecast for the specified date range
|
||||||
temperature_forecast = PVforecast.get_temperature_for_date_range(date_now,date)
|
pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now, date)
|
||||||
|
temperature_forecast = PVforecast.get_temperature_for_date_range(date_now, date)
|
||||||
|
|
||||||
#print(specific_date_prices)
|
# Return both forecasts as a JSON response
|
||||||
ret = {"temperature":temperature_forecast.tolist(),"pvpower":pv_forecast.tolist()}
|
ret = {"temperature": temperature_forecast.tolist(), "pvpower": pv_forecast.tolist()}
|
||||||
return jsonify(ret)
|
return jsonify(ret)
|
||||||
|
|
||||||
|
|
||||||
@app.route('/optimize', methods=['POST'])
|
@app.route('/optimize', methods=['POST'])
|
||||||
def flask_optimize():
|
def flask_optimize():
|
||||||
if request.method == 'POST':
|
if request.method == 'POST':
|
||||||
|
# Retrieve optimization parameters from the request JSON
|
||||||
parameter = request.json
|
parameter = request.json
|
||||||
|
|
||||||
# Erforderliche Parameter prüfen
|
# Check for required parameters
|
||||||
erforderliche_parameter = [ 'preis_euro_pro_wh_akku','strompreis_euro_pro_wh', "gesamtlast",'pv_akku_cap', "einspeiseverguetung_euro_pro_wh", 'pv_forecast','temperature_forecast', 'eauto_min_soc', "eauto_cap","eauto_charge_efficiency","eauto_charge_power","eauto_soc","pv_soc","start_solution","haushaltsgeraet_dauer","haushaltsgeraet_wh"]
|
required_parameters = ['preis_euro_pro_wh_akku', 'strompreis_euro_pro_wh', "gesamtlast", 'pv_akku_cap',
|
||||||
for p in erforderliche_parameter:
|
"einspeiseverguetung_euro_pro_wh", 'pv_forecast', 'temperature_forecast',
|
||||||
if p not in parameter:
|
'eauto_min_soc', "eauto_cap", "eauto_charge_efficiency", "eauto_charge_power",
|
||||||
return jsonify({"error": f"Fehlender Parameter: {p}"}), 400
|
"eauto_soc", "pv_soc", "start_solution", "haushaltsgeraet_dauer",
|
||||||
|
"haushaltsgeraet_wh"]
|
||||||
|
# Identify any missing parameters
|
||||||
|
missing_params = [p for p in required_parameters if p not in parameter]
|
||||||
|
if missing_params:
|
||||||
|
return jsonify({"error": f"Missing parameter: {', '.join(missing_params)}"}), 400 # Return error for missing parameters
|
||||||
|
|
||||||
# Simulation durchführen
|
# Perform optimization simulation
|
||||||
ergebnis = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour) # , startdate = datetime.now().date() - timedelta(days = 1)
|
result = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour)
|
||||||
|
|
||||||
return jsonify(ergebnis)
|
|
||||||
|
|
||||||
|
return jsonify(result) # Return optimization results as JSON
|
||||||
|
|
||||||
@app.route('/visualisierungsergebnisse.pdf')
|
@app.route('/visualisierungsergebnisse.pdf')
|
||||||
def get_pdf():
|
def get_pdf():
|
||||||
return send_from_directory('', 'visualisierungsergebnisse.pdf')
|
# Endpoint to serve the generated PDF with visualization results
|
||||||
|
return send_from_directory('', 'visualisierungsergebnisse.pdf') # Adjust the directory if needed
|
||||||
|
|
||||||
@app.route("/site-map")
|
@app.route("/site-map")
|
||||||
def site_map():
|
def site_map():
|
||||||
|
# Function to generate a site map of valid routes in the application
|
||||||
def print_links(links):
|
def print_links(links):
|
||||||
### This is lazy. Use templates
|
|
||||||
content = "<h1>Valid routes</h1><ul>"
|
content = "<h1>Valid routes</h1><ul>"
|
||||||
for link in links:
|
for link in links:
|
||||||
content += f"<li><a href='{link}'>{link}</li>"
|
content += f"<li><a href='{link}'>{link}</a></li>"
|
||||||
content = content + "<ul>"
|
content += "</ul>"
|
||||||
return content
|
return content
|
||||||
|
|
||||||
|
# Check if the route has no empty parameters
|
||||||
def has_no_empty_params(rule):
|
def has_no_empty_params(rule):
|
||||||
defaults = rule.defaults if rule.defaults is not None else ()
|
defaults = rule.defaults if rule.defaults is not None else ()
|
||||||
arguments = rule.arguments if rule.arguments is not None else ()
|
arguments = rule.arguments if rule.arguments is not None else ()
|
||||||
return len(defaults) >= len(arguments)
|
return len(defaults) >= len(arguments)
|
||||||
|
# Collect all valid GET routes without empty parameters
|
||||||
links = []
|
links = []
|
||||||
for rule in app.url_map.iter_rules():
|
for rule in app.url_map.iter_rules():
|
||||||
if "GET" in rule.methods and has_no_empty_params(rule):
|
if "GET" in rule.methods and has_no_empty_params(rule):
|
||||||
url = url_for(rule.endpoint, **(rule.defaults or {}))
|
url = url_for(rule.endpoint, **(rule.defaults or {}))
|
||||||
links.append(url)
|
links.append(url)
|
||||||
return print_links(sorted(links))
|
return print_links(sorted(links))# Return the sorted links as HTML
|
||||||
|
|
||||||
|
|
||||||
@app.route('/')
|
@app.route('/')
|
||||||
def root():
|
def root():
|
||||||
|
# Redirect the root URL to the site map
|
||||||
return redirect("/site-map", code=302)
|
return redirect("/site-map", code=302)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
try:
|
try:
|
||||||
host= os.getenv("FLASK_RUN_HOST", "0.0.0.0")
|
# Set host and port from environment variables or defaults
|
||||||
|
host = os.getenv("FLASK_RUN_HOST", "0.0.0.0")
|
||||||
port = os.getenv("FLASK_RUN_PORT", 8503)
|
port = os.getenv("FLASK_RUN_PORT", 8503)
|
||||||
app.run(debug=True, host=host, port=port)
|
app.run(debug=True, host=host, port=port) # Run the Flask application
|
||||||
except:
|
except Exception as e:
|
||||||
print(f"Coud not bind to host {host}:{port}, set FLASK_RUN_HOST and/or FLASK_RUN_PORT.")
|
print(f"Could not bind to host {host}:{port}. Error: {e}") # Error handling for binding issues
|
||||||
|
|
||||||
|
|
||||||
# PV Forecast:
|
# PV Forecast:
|
||||||
# object {
|
# object {
|
||||||
|
Loading…
x
Reference in New Issue
Block a user