import pandas as pd import numpy as np from flask import Flask, jsonify, request from flask_restx import Api, Resource, Namespace # from flask_httpauth import HTTPBasicAuth import requests import base64 import json import datetime from calendar import monthrange import pymongo from mongoengine import StringField, ListField, DateTimeField, DictField """ This script creates an api that connects to the MongoDB database. This api will eventually allow connection between the database and the frontend """ # Connect to MongoDB # For some reason none of this works when im connected to VPN app = Flask(__name__) api = Api(app, version='1.0', title='Haya Energy NucPy API', description=""" API endpoints used to communicate NucPy with MongoDB """, contact="Diego", endpoint="/nucpy/v1") def mongo_unavs_call(user_input_start_date, user_input_end_date, user_input_photo_date, user_input_past_date): print("Starting mongo_unavs_call") # Connect to the MongoDB database user = "dmarroquin" passw = "tN9XpCCQM2MtYDme" host = "nucmonitordata.xxcwx9k.mongodb.net" client = pymongo.MongoClient( f"mongodb+srv://{user}:{passw}@{host}/?retryWrites=true&w=majority" ) db = client["data"] collection = db["unavs"] start_date = f"{user_input_start_date}T00:00:00" end_date = f"{user_input_end_date}T23:59:59" pipeline = [ { "$unwind": "$results" }, { "$unwind": "$results.generation_unavailabilities" }, { "$match": { "results.generation_unavailabilities.production_type": "NUCLEAR", "results.generation_unavailabilities.start_date": {"$lte": end_date}, "results.generation_unavailabilities.end_date": {"$gte": start_date}, "results.generation_unavailabilities.updated_date": {"$lte": end_date} } }, { "$project": { "_id": 0, "generation_unavailabilities": "$results.generation_unavailabilities" } } ] result = collection.aggregate(pipeline) return list(result) # --------------------------------------------------------------------------------------- # # Convert the dictionary of dictionaries to JSON def convert_to_json(item): if isinstance(item, dict): return {str(k): convert_to_json(v) for k, v in item.items()} elif isinstance(item, list): return [convert_to_json(i) for i in item] elif isinstance(item, ObjectId): return str(item) else: return item # --------------------------------------------------------------------------------------- # # Function gives the total of the data. When printed as dataframe/excel, # Will give a final row with the total for each plant and the total overall def add_total(data): total_values = {} for key in data: daily_values = data[key] total = sum(daily_values.values()) daily_values["Total"] = total for date, value in daily_values.items(): if date not in total_values: total_values[date] = value else: total_values[date] += value data["Total"] = total_values # --------------------------------------------------------------------------------------- # # This file will simply connect to the rte and get the data directly from there # Function to create an authentication token. This token is then used in the HTTP requests to the API for authentication. # It is necessary to receive data from RTE. def get_oauth(): # ID from the user. This is encoded to base64 and sent in an HTTP request to receive the oauth token. # This ID is from my account (RMP). However, another account can be created in the RTE API portal and get another ID. joined_ID = '057e2984-edb3-4706-984b-9ea0176e74db:dc9df9f7-9f91-4c7a-910c-15c4832fb7bc' b64_ID = base64.b64encode(joined_ID.encode('utf-8')) b64_ID_decoded = b64_ID.decode('utf-8') # Headers for the HTTP request headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Authorization': f'Basic {b64_ID_decoded}'} api_url = 'https://digital.iservices.rte-france.com/token/oauth/' # Call to the API and if successful, the response will be 200. response = requests.post(api_url, headers=headers) # When positive response, the token is retrieved data = response.json() oauth = data['access_token'] return(oauth) # --------------------------------------------------------------------------------------- # # This function does severall calls to the RTE API (because maximum time between start_date and end_date is 1 month) # the argument past_photo is a boolean (True, False) that indicates if we want to make a photo from the past or not # However, the past_photo part and past_date is not yet implemented. def get_unavailabilities(usr_start_date, usr_end_date): oauth = get_oauth() print("Get Oauth done") date_type = 'APPLICATION_DATE' # Current year/month/day/hour/minute/second is calculated for the last call to the API. For instance, if today is 05/05/2023, # the last call of the API will be from 01/05/2023 to 05/05/2023 (+current hour,minute,second). current_datetime = datetime.datetime.now() current_year = current_datetime.strftime('%Y') current_month = current_datetime.strftime('%m') current_day = current_datetime.strftime('%d') current_hour = current_datetime.strftime('%H') current_minute = current_datetime.strftime('%M') current_second = current_datetime.strftime('%S') # Headers for the HTTP request headers = {'Host': 'digital.iservices.rte-france.com', 'Authorization': f'Bearer {oauth}' } # the responses object is where we are going to store all the responses from the API. # Initially, current_datetime is included to know when we have called the API and all the # individual results of the API (because each call is Maz 1 month) are stored in responses["results"] responses = {"current_datetime": current_datetime.strftime("%m/%d/%Y, %H:%M:%S"), "results":[] } # --------------------------- HERE HAVE TO GET THE RANGE OF DATES FROM START AND END AND PUT THEM INTO LIST --------------------------- # # Convert start_date and end_date to datetime objects start_date_obj = datetime.datetime.strptime(usr_start_date, "%Y-%m-%d").date() end_date_obj = datetime.datetime.strptime(usr_end_date, "%Y-%m-%d").date() # Initialize lists to store years and months years = [] months = [] # Generate the range of years and months current_date = start_date_obj while current_date <= end_date_obj: years.append(current_date.year) months.append(current_date.month) current_date += datetime.timedelta(days=1) # Remove duplicates from the lists years = list(set(years)) months = list(set(months)) years.sort() months.sort() print(years) print(months) # --------------------------- HERE HAVE TO GET THE RANGE OF DATES FROM START AND END AND PUT THEM INTO LIST --------------------------- # # Loop to call the API all the necessary times. for i in range(len(years)): for j in range(len(months)): # start_year and start_month of the current call to the API start_year = years[i] start_month = months[j] # start_date is constructed. Now we only need to construct the end_date. start_date = f'{start_year}-{start_month}-01T00:00:00%2B02:00' if True: # Calculate the number of days in the current month _, num_days = monthrange(int(start_year), int(start_month)) end_date = f'{start_year}-{start_month}-{num_days}T23:59:59%2B02:00' print(f'start date is {start_date}') print(f'end date is {end_date}') # Call to the API api_url = f'https://digital.iservices.rte-france.com/open_api/unavailability_additional_information/v4/generation_unavailabilities?date_type={date_type}&start_date={start_date}&end_date={end_date}' response = requests.get(api_url, headers=headers) json_response = response.json() responses["results"].append(json_response) # print(responses) return responses # --------------------------------------------------------------------------------------- # def nuc_monitor(rte_data, mongo_data, usr_start_date, usr_end_date, photo_date, past_date): # # Slightly changed metadata to fit the data from the RTE API: ST-LAURENT B 2 --> ST LAURENT 2, .... plants_metadata = {"BELLEVILLE 1": 1310.0, "BELLEVILLE 2": 1310.0, "BLAYAIS 1": 910.0, "BLAYAIS 2": 910.0, "BLAYAIS 3": 910.0, "BLAYAIS 4": 910.0, "BUGEY 2": 910.0, "BUGEY 3": 910.0, "BUGEY 4": 880.0, "BUGEY 5": 880.0, "CATTENOM 1": 1300.0, "CATTENOM 2": 1300.0, "CATTENOM 3": 1300.0, "CATTENOM 4": 1300.0, "CHINON 1": 905.0, "CHINON 2": 905.0, "CHINON 3": 905.0, "CHINON 4": 905.0, "CHOOZ 1": 1500.0, "CHOOZ 2": 1500.0, "CIVAUX 1": 1495.0, "CIVAUX 2": 1495.0, "CRUAS 1": 915.0, "CRUAS 2": 915.0, "CRUAS 3": 915.0, "CRUAS 4": 915.0, "DAMPIERRE 1": 890.0, "DAMPIERRE 2": 890.0, "DAMPIERRE 3": 890.0, "DAMPIERRE 4": 890.0, "FLAMANVILLE 1": 1330.0, "FLAMANVILLE 2": 1330.0, "GOLFECH 1": 1310.0, "GOLFECH 2": 1310.0, "GRAVELINES 1": 910.0, "GRAVELINES 2": 910.0, "GRAVELINES 3": 910.0, "GRAVELINES 4": 910.0, "GRAVELINES 5": 910.0, "GRAVELINES 6": 910.0, "NOGENT 1": 1310.0, "NOGENT 2": 1310.0, "PALUEL 1": 1330.0, "PALUEL 2": 1330.0, "PALUEL 3": 1330.0, "PALUEL 4": 1330.0, "PENLY 1": 1330.0, "PENLY 2": 1330.0, "ST ALBAN 1": 1335.0, "ST ALBAN 2": 1335.0, "ST LAURENT 1": 915.0, "ST LAURENT 2": 915.0, "TRICASTIN 1": 915.0, "TRICASTIN 2": 915.0, "TRICASTIN 3": 915.0, "TRICASTIN 4": 915.0, "FESSENHEIM 1": 880.0, "FESSENHEIM 2": 880.0} # --------------------- INITIAL DATA CLEANING FOR RTE DATA ------------------------ # unav_API = rte_data.json() # print(unav_API) # Store the unavailabilities in a list unavailabilities = [] print("Unav") for unavailabilities_API in unav_API['results']: try: unavailabilities.extend(unavailabilities_API.get('generation_unavailabilities', [])) except: print('There was an error') # print(unavailabilities_API) rte_df = pd.DataFrame(unavailabilities) def unpack_values(row): if isinstance(row["values"], list): for key, value in row["values"][0].items(): row[key] = value return row # Apply the function to each row in the DataFrame rte_df = rte_df.apply(unpack_values, axis=1) # Drop the original "values" column rte_df.drop("values", axis=1, inplace=True) # Unpack the unit column rte_df2 = pd.concat([rte_df, pd.json_normalize(rte_df['unit'])], axis=1) rte_df2.drop('unit', axis=1, inplace=True) rte_nuclear_unav = rte_df2[(rte_df2["production_type"] == "NUCLEAR")] # --------------------- INITIAL DATA CLEANING FOR RTE DATA ------------------------ # # --------------------- INITIAL DATA CLEANING FOR MONGO DATA ------------------------ # # # Create a DataFrame mongo_df = pd.DataFrame(mongo_data) # Unpack the dictionaries into separate columns mongo_df_unpacked = pd.json_normalize(mongo_df['generation_unavailabilities']) # Concatenate the unpacked columns with the original DataFrame mongo_df_result = pd.concat([mongo_df, mongo_df_unpacked], axis=1) # Drop the original column mongo_df_result.drop(columns=['generation_unavailabilities'], inplace=True) mongo_df_columns = mongo_df_result.columns mongo_df_result['start_date'] = mongo_df_result['values'].apply(lambda x: x[0]['start_date']) mongo_df_result['end_date'] = mongo_df_result['values'].apply(lambda x: x[0]['end_date']) mongo_df_result['available_capacity'] = mongo_df_result['values'].apply(lambda x: x[0]['available_capacity']) mongo_df_result['unavailable_capacity'] = mongo_df_result['values'].apply(lambda x: x[0]['unavailable_capacity']) # print(mongo_df_result) # print(mongo_df_result.columns) # Drop the original 'values' column mongo_df_result.drop('values', axis=1, inplace=True) mongo_df2 = mongo_df_result mongo_df2.rename(columns=lambda col: col.replace('unit.', ''), inplace=True) # --------------------- INITIAL DATA CLEANING FOR MONGO DATA ------------------------ # # Make the two dataframes have the same columns mongo_unavs = mongo_df2.copy() mongo_unavs.drop(columns="type", inplace=True) rte_unavs = rte_nuclear_unav.copy() rte_unavs.drop(columns="type", inplace=True) # Merge dataframes column_order = mongo_unavs.columns # print(column_order) merged_df = pd.concat([mongo_unavs[column_order], rte_unavs[column_order]], ignore_index=True) # --------------------------- HERE IS THE CHANGE TO GET ONLY ACTIVE OR ACTIVE AND INACTIVE --------------------------- # # start_date_str = usr_start_date.strftime("%Y-%m-%d") start_date_str = usr_start_date # end_date_str = usr_end_date.strftime("%Y-%m-%d") end_date_str = usr_end_date current_datetime = datetime.datetime.now() current_datetime_str = current_datetime.strftime("%Y-%m-%d") if photo_date == True: nuclear_unav = merged_df.copy()[(merged_df.copy()["production_type"] == "NUCLEAR") & (merged_df.copy()["updated_date"] <= past_date)] photo_date = True else: # need to add updated_date as a conditional to get the newest for that day nuclear_unav = merged_df.copy()[(merged_df.copy()["production_type"] == "NUCLEAR") & (merged_df.copy()["updated_date"] <= end_date_str)] # --------------------------- HERE IS THE CHANGE TO GET ONLY ACTIVE OR ACTIVE AND INACTIVE --------------------------- # # --------------------- SECOND DATA CLEANING ------------------------ # # This filter should take only the most recent id and discard the rest # Sort by updated date sorted_df = nuclear_unav.copy().sort_values(by='updated_date') sorted_df = sorted_df.copy().reset_index(drop=True) # Filter to get identifiers filtered_id_df = sorted_df.copy() filtered_id_df.drop_duplicates(subset='identifier', keep='last', inplace=True) filtered_id_df = filtered_id_df.copy().reset_index(drop=True) # This filter should take all the dates with unavs that include days with unavs in the range of the start and end date filtered_df = filtered_id_df.copy()[(filtered_id_df.copy()['start_date'] <= end_date_str) & (filtered_id_df.copy()['end_date'] >= start_date_str)] # Standardize datetime in dataframe filtered_df2 = filtered_df.copy() # This code will just standardize datetime stuff filtered_df2['creation_date'] = pd.to_datetime(filtered_df2['creation_date'], utc=True) filtered_df2['updated_date'] = pd.to_datetime(filtered_df2['updated_date'], utc=True) filtered_df2['start_date'] = pd.to_datetime(filtered_df2['start_date'], utc=True) filtered_df2['end_date'] = pd.to_datetime(filtered_df2['end_date'], utc=True) # Drop the duplicates filtered_df3 = filtered_df2.copy().drop_duplicates() # start_date_datetime = pd.to_datetime(start_date_str, utc=True) # Remove timezone info start_date_datetime = pd.Timestamp(start_date_str, tz='UTC') # end_date_datetime = pd.to_datetime(end_date_str, utc=True) end_date_datetime = pd.Timestamp(end_date_str, tz='UTC') # Turn df into dict for json processing filtered_unavs = filtered_df3.copy().to_dict(orient='records') results = {} for unav in filtered_unavs: plant_name = unav['name'] if plant_name in results: # If the key is already in the dictionary, append unavailability to the list results[plant_name].append({'status': unav['status'], 'id': unav['message_id'], 'creation_date': unav['creation_date'], 'updated_date': unav['updated_date'], 'start_date': unav['start_date'], 'end_date': unav['end_date'], 'available_capacity': unav['available_capacity']}) else: # if the key of the plant is not there yet, create a new element of the dictionary # Get message_id instead of identifier, easier to identify stuff with it results[plant_name] = [{'status': unav['status'], 'id': unav['message_id'], 'creation_date': unav['creation_date'], 'updated_date': unav['updated_date'], 'start_date': unav['start_date'], 'end_date': unav['end_date'], 'available_capacity': unav['available_capacity']}] # Custom encoder to handle datetime objects class DateTimeEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, datetime.datetime): return o.isoformat() return super().default(o) results_holder = results # Create new dict with each plant only having start_date less than user_end_date and an end_date greater than user_start_date # should just be doing the same as above in the df for filtering only dates that inclued the start and end date start_date = start_date_datetime.date() end_date = end_date_datetime.date() results_filtered = results_holder for key, value in results_filtered.items(): filtered_values = [] for item in value: item_start_date = item['start_date'].date() item_end_date = item['end_date'].date() identifier = item['id'] if item_start_date < end_date and item_end_date > start_date and identifier not in filtered_values: filtered_values.append(item) results_filtered[key] = filtered_values sorted_results = results_filtered # --------------------- SECOND DATA CLEANING ------------------------ # # --------------------------- HERE IS THE FINAL PROCESS --------------------------- # for key, value in sorted_results.items(): sorted_results[key] = sorted(value, key=lambda x: x['updated_date']) results_sorted = sorted_results dates_of_interest = [start_date] # We are creating a list of dates ranging from user specified start and end dates date_plus_one = start_date while date_plus_one < end_date: date_plus_one = date_plus_one + datetime.timedelta(days=1) dates_of_interest.append(date_plus_one) # This is to standardize the datetimes. Without this, the datetime calculations for each power plant will not work results_plants = {plant_name: {date: {"available_capacity": power, "updated_date": pd.to_datetime("1970-01-01", utc=True)} for date in dates_of_interest} for plant_name, power in plants_metadata.items()} for plant, unavailabilities in results_sorted.items(): original_power = plants_metadata[plant] # Get all the unavailabilities scheduled for the plant. results_current_plant = results_plants[plant] for unavailability in unavailabilities: # For each unavailability, the resulting power, start and end datetime are collected. Need to collect updated_date power_unavailability = unavailability["available_capacity"] updated_date_unav = unavailability["updated_date"] # The date comes as a string start_datetime_unav = unavailability["start_date"] end_datetime_unav = unavailability["end_date"] start_date_unav = start_datetime_unav.date() # Extract date part end_date_unav = end_datetime_unav.date() # Extract date part # For the current unavailability, we want to find which days it affects for day in dates_of_interest: start_hour = start_datetime_unav.hour start_minute = start_datetime_unav.minute end_hour = end_datetime_unav.hour end_minute = end_datetime_unav.minute if start_date_unav <= day <= end_date_unav: # Check if the day is already updated with a later update_date if day in results_current_plant and updated_date_unav <= results_current_plant[day]["updated_date"]: continue # Skip to the next loop if there is already information for a later update_date # Calculate the % of the day that the plant is under maintenance if start_date_unav == day and day == end_date_unav: # The unavailability starts and ends on the same day percentage_of_day = (end_hour * 60 + end_minute - start_hour * 60 - start_minute) / (24 * 60) elif start_date_unav == day: # The unavailability starts on the current day but ends on a later day percentage_of_day = (24 * 60 - (start_hour * 60 + start_minute)) / (24 * 60) elif day == end_date_unav: # The unavailability starts on a previous day and ends on the current day percentage_of_day = (end_hour * 60 + end_minute) / (24 * 60) else: # The unavailability covers the entire day percentage_of_day = 1 # The average power of the day is calculated power_of_day = percentage_of_day * power_unavailability + (1 - percentage_of_day) * original_power # Update the available_capacity for the day only if it's not already updated with a later update_date if day not in results_current_plant or updated_date_unav > results_current_plant[day]["updated_date"]: results_current_plant[day] = {"available_capacity": power_of_day, "updated_date": updated_date_unav} output_results = {} for plant, plant_data in results_plants.items(): available_capacity_per_day = {str(date): data["available_capacity"] for date, data in plant_data.items()} output_results[plant] = available_capacity_per_day # print(output_results) add_total(output_results) # print("Done") # print(results_plants) # Convert datetime key to string to store in mongodb output_results = {plant: {str(date): power for date, power in plant_data.items()} for plant, plant_data in output_results.items()} # print(output_results) # ------------------------------------------------- if photo_date == False: json_data = json.dumps(output_results) # print(json_data) return json_data else: json_data = json.dumps(output_results) # print(json_data) return json_data # ------------------------------------------------- return # Namespaces # Get raw data stuff raw_ns = Namespace('raw', description='Raw Data', path='/nucpy/v1') api.add_namespace(raw_ns) @raw_ns.route('/raw', methods=["GET"]) @raw_ns.doc(params= {"start_date": "Start date", "end_date": "end date", "photo_date": "True False", "past_date": "Cutoff date"}) class Raw(Resource): # @auth.login_required def get(self): # raw_data = merge_gridfs_files_to_json() print("Applying request") mongo_start_date = request.args.get("start_date") mongo_end_date = request.args.get("end_date") mongo_photo_date = request.args.get("photo_date") mongo_past_date = request.args.get("past_date") print("Getting raw_data") raw_data = mongo_unavs_call(mongo_start_date, mongo_end_date, mongo_past_date, mongo_photo_date) print("Returning raw_data") # print(raw_data) return raw_data # Get RTE data rte_ns = Namespace('rte', description='RTE Data', path='/nucpy/v1') api.add_namespace(rte_ns) @rte_ns.route('/rte', methods=["GET"]) @rte_ns.doc(params= {"start_date": "Start date", "end_date": "end date"}) class RTEDATA(Resource): # @auth.login_required def get(self): rte_start_date = request.args.get("start_date") rte_end_date = request.args.get("end_date") print(rte_start_date) print(rte_end_date) # Process the user input and retrieve data data = get_unavailabilities(rte_start_date, rte_end_date) return data # Get processed data nucmonitor_ns = Namespace('nucmonitor', description='Nucmonitor', path='/nucpy/v1') api.add_namespace(nucmonitor_ns) @nucmonitor_ns.route('/nucmonitor', methods=['GET']) class Nucmonitor(Resource): # @auth.login_required def get(self): # Retrieve input parameters from request.args start_date = request.args.get("start_date") end_date = request.args.get("end_date") photo_date = request.args.get("photo_date") past_date = request.args.get("past_date") # Call the /rte endpoint to get RTE data rte_data = self.get_rte_data(start_date, end_date) print("Got RTE data") print("Getting Mongo data") mongo_data = self.get_mongo_data(start_date, end_date, photo_date, past_date) print("Got Mongo data") # print(mongo_data) # Process data using nuc_monitor nucmonitor_response = nuc_monitor(rte_data, mongo_data, start_date, end_date, photo_date, past_date) # print(nucmonitor_response) return (nucmonitor_response) def get_rte_data(self, start_date, end_date): rte_url = "http://0.0.0.0:7860/nucpy/v1/rte" # RTE endpoint URL rte_params = {"start_date": start_date, "end_date": end_date} rte_response = requests.get(rte_url, params=rte_params) # rte_data = rte_response.json() return rte_response def get_mongo_data(self, start_date, end_date, photo_date, past_date): print("Getting url") mongo_url = "http://0.0.0.0:7860/nucpy/v1/raw" # Mongo endpoint URL print("Getting params") mongo_params = {"start_date": start_date, "end_date": end_date, "photo_date": photo_date, "past_date": past_date} print("Getting request") mongo_response = requests.get(mongo_url, params=mongo_params) # mongo_data = mongo_response.json() print("Returning response") return mongo_response if __name__ == '__main__': app.run(host='0.0.0.0', debug=True, port=7860)