diff --git a/flask_server.py b/flask_server.py index ef682b7..67444e2 100755 --- a/flask_server.py +++ b/flask_server.py @@ -31,118 +31,95 @@ app = Flask(__name__) opt_class = optimization_problem(prediction_hours=prediction_hours, strafe=10, optimization_hours=optimization_hours) - - # @app.route('/last_correction', methods=['GET']) # def flask_last_correction(): - # if request.method == 'GET': - # year_energy = float(request.args.get("year_energy")) - # date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date()) - # ############### - # # Load Forecast - # ############### - # 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] # Nur Erwartungswert! - - # gesamtlast = Gesamtlast(prediction_hours=prediction_hours) - # gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) - - # # ############### - # # # WP - # # ############## - # # leistung_wp = wp.simulate_24h(temperature_forecast) - # # gesamtlast.hinzufuegen("Heatpump", leistung_wp) - - # last = gesamtlast.gesamtlast_berechnen() - # print(last) - # #print(specific_date_prices) - # return jsonify(last.tolist()) - +# if request.method == 'GET': +# year_energy = float(request.args.get("year_energy")) +# date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date()) +# ############### +# # Load Forecast +# ############### +# lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) +# leistung_haushalt = lf.get_stats_for_date_range(date_now, date)[0] # Only the expected value! +# +# gesamtlast = Gesamtlast(prediction_hours=prediction_hours) +# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) +# # ############### +# # Heat Pump (WP) +# # ############## +# # leistung_wp = wp.simulate_24h(temperature_forecast) +# # gesamtlast.hinzufuegen("Heatpump", leistung_wp) +# last = gesamtlast.gesamtlast_berechnen() +# print(last) +# return jsonify(last.tolist()) @app.route('/soc', methods=['GET']) def flask_soc(): - - # MariaDB Verbindungsdetails + # MariaDB connection details config = db_config - # Parameter festlegen + # Set parameters for SOC (State of Charge) calculation voltage_high_threshold = 55.4 # 100% SoC voltage_low_threshold = 46.5 # 0% SoC - current_low_threshold = 2 # Niedriger Strom für beide Zustände - gap = 30 # Zeitlücke in Minuten zum Gruppieren von Maxima/Minima - bat_capacity = 33 * 1000 / 48 + current_low_threshold = 2 # Low current threshold for both states + gap = 30 # Time gap in minutes for grouping maxima/minima + 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') - - # BatteryDataProcessor instanziieren und verwenden - processor = BatteryDataProcessor(config, voltage_high_threshold, voltage_low_threshold, current_low_threshold, gap,bat_capacity) + # Instantiate BatteryDataProcessor and perform calculations + processor = BatteryDataProcessor(config, voltage_high_threshold, voltage_low_threshold, current_low_threshold, gap, bat_capacity) processor.connect_db() processor.fetch_data(zeitpunkt_x) processor.process_data() 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) - #soh_df = processor.calculate_soh(integration_results) - processor.update_database_with_soc(soc_df) - #processor.plot_data(last_points_100_df, last_points_0_df, soc_df) - processor.disconnect_db() + # soh_df = processor.calculate_soh(integration_results) # Optional State of Health calculation + 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) # Optional data visualization + processor.disconnect_db() # Disconnect from the database return jsonify("Done") - - - - @app.route('/strompreis', methods=['GET']) def flask_strompreis(): - date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date()) - filepath = os.path.join (r'test_data', r'strompreise_akkudokAPI.json') # Pfad zur JSON-Datei anpassen - #price_forecast = HourlyElectricityPriceForecast(source=filepath) - price_forecast = HourlyElectricityPriceForecast(source="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) - #print(specific_date_prices) - return jsonify(specific_date_prices.tolist()) + # Get the current date and the end date based on prediction hours + date_now, date = get_start_enddate(prediction_hours, startdate=datetime.now().date()) + filepath = os.path.join(r'test_data', r'strompreise_akkudokAPI.json') # Adjust the path to the JSON file + 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) # Fetch prices for the specified date range + return jsonify(specific_date_prices.tolist()) - - -# 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: +# Endpoint to handle total load calculation based on the latest measured data @app.route('/gesamtlast', methods=['POST']) def flask_gesamtlast(): - # Daten aus dem JSON-Body abrufen + # Retrieve data from the JSON body data = request.get_json() # Extract year_energy and prediction_hours from the request JSON year_energy = float(data.get("year_energy")) 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") - - # Convert JSON data into a Pandas DataFrame 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']) - # 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: - # Treat it as local time and localize it measured_data['time'] = measured_data['time'].dt.tz_localize('Europe/Berlin') else: - # Convert the time to local timezone (e.g., '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) # Instantiate LoadForecast and generate forecast data lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) - - # Generate forecast data based on the measured data time range 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()): date_str = single_date.strftime('%Y-%m-%d') daily_forecast = lf.get_daily_stats(date_str) @@ -153,207 +130,196 @@ def flask_gesamtlast(): # Concatenate all daily forecasts into a single DataFrame 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.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 - 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 + # Extract household power predictions leistung_haushalt = future_predictions['Adjusted Pred'].values - - # Instantiate Gesamtlast and add household power predictions gesamtlast = Gesamtlast(prediction_hours=prediction_hours) - gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) - - # ############### - # # WP (optional) - # ############### - # leistung_wp = wp.simulate_24h(temperature_forecast) - # gesamtlast.hinzufuegen("Heatpump", leistung_wp) + gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation # Calculate the total load - last = gesamtlast.gesamtlast_berechnen() - - # Return the calculated load as JSON + last = gesamtlast.gesamtlast_berechnen() # Compute total load return jsonify(last.tolist()) - - - # @app.route('/gesamtlast', methods=['GET']) # def flask_gesamtlast(): - # if request.method == 'GET': - # year_energy = float(request.args.get("year_energy")) - # prediction_hours = int(request.args.get("hours", 48)) # Default to 24 hours if not specified - # date_now = datetime.now() - # end_date = (date_now + timedelta(hours=prediction_hours)).strftime('%Y-%m-%d %H:%M:%S') +# if request.method == 'GET': +# 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 48 hours if not specified +# 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') # Calculate end date based on prediction hours - # ############### - # # Load Forecast - # ############### - # # Instantiate LastEstimator and get measured data - # estimator = LastEstimator() - # start_date = (date_now - timedelta(days=60)).strftime('%Y-%m-%d') # Example: last 60 days - # end_date = date_now.strftime('%Y-%m-%d') # Current date +# ############### +# # Load Forecast +# ############### +# # Instantiate LastEstimator to retrieve measured data +# estimator = LastEstimator() +# 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 - # 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['time'] = pd.to_datetime(selected_columns['timestamp']).dt.floor('H') - # selected_columns['Last'] = pd.to_numeric(selected_columns['Last'], errors='coerce') - # cleaned_data = selected_columns.dropna() +# selected_columns = last_df[['timestamp', 'Last']] # Select relevant columns +# 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') # Convert 'Last' to numeric, coerce errors +# cleaned_data = selected_columns.dropna() # Clean data by dropping NaN values - # # Instantiate LoadForecast - # lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) +# # Instantiate LoadForecast +# lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) - # # Generate forecast data - # forecast_list = [] - # for single_date in pd.date_range(cleaned_data['time'].min().date(), cleaned_data['time'].max().date()): - # date_str = single_date.strftime('%Y-%m-%d') - # daily_forecast = lf.get_daily_stats(date_str) - # mean_values = daily_forecast[0] - # hours = [single_date + pd.Timedelta(hours=i) for i in range(24)] - # daily_forecast_df = pd.DataFrame({'time': hours, 'Last Pred': mean_values}) - # forecast_list.append(daily_forecast_df) +# # Generate forecast data +# forecast_list = [] # List to hold daily forecasts +# 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') # Format date +# daily_forecast = lf.get_daily_stats(date_str) # Get daily stats from LoadForecast +# mean_values = daily_forecast[0] # Extract mean values +# 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}) # Create DataFrame for daily forecast +# 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 - # adjuster = LoadPredictionAdjuster(cleaned_data, forecast_df, lf) - # adjuster.calculate_weighted_mean() - # adjuster.adjust_predictions() +# # Create LoadPredictionAdjuster instance +# adjuster = LoadPredictionAdjuster(cleaned_data, forecast_df, lf) +# adjuster.calculate_weighted_mean() # Calculate weighted mean for adjustments +# adjuster.adjust_predictions() # Adjust predictions based on measured data - # # Predict the next hours - # future_predictions = adjuster.predict_next_hours(prediction_hours) +# # Predict the next 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.hinzufuegen("Haushalt", leistung_haushalt) +# gesamtlast = Gesamtlast(prediction_hours=prediction_hours) # Create Gesamtlast instance +# gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation - # # ############### - # # # WP - # # ############## - # # leistung_wp = wp.simulate_24h(temperature_forecast) - # # gesamtlast.hinzufuegen("Heatpump", leistung_wp) - - # last = gesamtlast.gesamtlast_berechnen() - # print(last) - # return jsonify(last.tolist()) +# # ############### +# # # WP (Heat Pump) +# # ############## +# # leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours +# # 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 @app.route('/gesamtlast_simple', methods=['GET']) def flask_gesamtlast_simple(): if request.method == 'GET': - year_energy = float(request.args.get("year_energy")) - date_now,date = get_start_enddate(prediction_hours,startdate=datetime.now().date()) + 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()) # Get the current date and prediction end date + ############### # Load Forecast ############### - 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] # Nur Erwartungswert! + lf = LoadForecast(filepath=r'load_profiles.npz', year_energy=year_energy) # Instantiate LoadForecast with specified parameters + leistung_haushalt = lf.get_stats_for_date_range(date_now, date)[0] # Get expected household load for the date range - gesamtlast = Gesamtlast(prediction_hours=prediction_hours) - gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) + gesamtlast = Gesamtlast(prediction_hours=prediction_hours) # Create Gesamtlast instance + gesamtlast.hinzufuegen("Haushalt", leistung_haushalt) # Add household load to total load calculation # ############### - # # WP + # # WP (Heat Pump) # ############## - # leistung_wp = wp.simulate_24h(temperature_forecast) - # gesamtlast.hinzufuegen("Heatpump", leistung_wp) + # leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours + # 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']) def flask_pvprognose(): if request.method == 'GET': + # Retrieve URL and AC power measurement from query parameters url = request.args.get("url") 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 ############### - PVforecast = PVForecast(prediction_hours = prediction_hours, url=url) - #print("PVPOWER",parameter['pvpowernow']) - if isfloat(ac_power_measurement): - PVforecast.update_ac_power_measurement(date_time=datetime.now(), ac_power_measurement=float(ac_power_measurement) ) - #PVforecast.print_ac_power_and_measurement() + PVforecast = PVForecast(prediction_hours=prediction_hours, url=url) # Instantiate PVForecast with given parameters + if isfloat(ac_power_measurement): # Check if the AC power measurement is a valid float + PVforecast.update_ac_power_measurement(date_time=datetime.now(), ac_power_measurement=float(ac_power_measurement)) # Update measurement + + # Get PV forecast and temperature forecast for the specified date range + pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now, date) + temperature_forecast = PVforecast.get_temperature_for_date_range(date_now, date) - pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now,date) #get_forecast_for_date(date) - temperature_forecast = PVforecast.get_temperature_for_date_range(date_now,date) - - #print(specific_date_prices) - ret = {"temperature":temperature_forecast.tolist(),"pvpower":pv_forecast.tolist()} + # Return both forecasts as a JSON response + ret = {"temperature": temperature_forecast.tolist(), "pvpower": pv_forecast.tolist()} return jsonify(ret) - @app.route('/optimize', methods=['POST']) def flask_optimize(): if request.method == 'POST': + # Retrieve optimization parameters from the request JSON parameter = request.json - # Erforderliche Parameter prüfen - 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"] - for p in erforderliche_parameter: - if p not in parameter: - return jsonify({"error": f"Fehlender Parameter: {p}"}), 400 + # Check for required parameters + required_parameters = ['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"] + # 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 - ergebnis = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour) # , startdate = datetime.now().date() - timedelta(days = 1) - - return jsonify(ergebnis) + # Perform optimization simulation + result = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour) + return jsonify(result) # Return optimization results as JSON @app.route('/visualisierungsergebnisse.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") def site_map(): + # Function to generate a site map of valid routes in the application def print_links(links): - ### This is lazy. Use templates content = "

Valid routes