from models import Location from db_connector import database,engine,db_session import pandas as pd import openmeteo_requests import requests_cache from retry_requests import retry def initialize_database(): """ A function to initialize the database by checking table availability and creating it if it does not exist. """ db_tables = ['locations'] # check table availability and create it if it does not exist for tb in db_tables: if not engine.dialect.has_table(engine.connect(), tb): print("Creating table: {}\n".format(tb)) database.metadata.create_all(engine) table_data = pd.read_csv ('./table_data/locations.csv', index_col=None, header=0) for i in range(len(table_data)): add_location(Location(name=table_data.loc[i, "Capital City"], latitude=float(table_data.loc[i, "Latitude"]), longitude=float(table_data.loc[i, "Longitude"]), ), no_commit=True ) db_session.commit() def add_location(location:Location , no_commit=False): """ A function that adds a location to the database session. Parameters: location (Location): The location object to be added to the database session. no_commit (bool): Flag indicating whether to commit the transaction immediately. Returns: None """ db_session.add(location) #print("Adding location: {}".format(location)) if not no_commit: db_session.commit() def delete_location(id): """ Deletes a location from the database based on the provided ID. Parameters: id (int): The ID of the location to be deleted. Returns: dict: A dictionary with the key "id" indicating that the location was successfully deleted. """ db_session.query(Location).filter(Location.id == id).delete() db_session.commit() return {"id": "Deleted"} def chunkify_data(data, chunk_size): """ A function that chunks the input data into smaller pieces of the specified chunk size. Parameters: - data: the input data to be chunked - chunk_size: the size of each chunk Returns: - A generator that yields chunks of the input data """ for i in range(0, len(data), chunk_size): yield data[i:i + chunk_size] def retrieve_weather_data(location_id=None): """ A function to retrieve weather data based on location IDs. Parameters: - location_id: an optional parameter to specify the location ID(s) to retrieve weather data for. If 'all' is provided, data for all locations is returned. Returns: - If location_id is not 'all' and any specified location ID does not exist, a dictionary of invalid location IDs is returned. - If location_id is None, None is returned. - If location_id is 'all', weather data for all locations is returned in a dictionary. - Otherwise, weather data for the specified location IDs is returned in a dictionary. """ print('xaz') max_id = db_session.query(Location).count() #NOTE: Disable location check if location_id is 'all' (returns all locations), debugging purposes if location_id != 'all': loc_check = {} for loc_id in location_id: if loc_id > max_id: loc_check[loc_id] = "The location does not exist" print(len(loc_check)) if len(loc_check) > 0: return {'error': loc_check} print('az') weather_data = {} #NOTE: Get weather data for all locations if location_id is 'all' (debugging purposes), otherwise get weather data for specified locations if location_id == 'all': locations = list(chunkify_data(db_session.query(Location).all(),100)) else: locations = list(chunkify_data(db_session.query(Location).filter(Location.id.in_(location_id)).all(),100)) print(locations) for chunk in locations: coordinates = [[loc.id, loc.latitude, loc.longitude, loc.name] for loc in chunk] weather_data.update(get_openmeteo_data(coordinates)) #print(weather_data) #exit(0) return weather_data def get_openmeteo_data(coordinates): """ Retrieves weather data from the OpenMeteo API for the given coordinates. Parameters: coordinates (list): List of tuples containing latitude, longitude, and location name. Returns: dict: A dictionary containing the retrieved weather data for each location. """ data_dict = {} cache_session = requests_cache.CachedSession('.cache', expire_after = 3600) retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2) openmeteo_client = openmeteo_requests.Client(session = retry_session) latitude = [coords[1] for coords in coordinates] longitude = [coords[2] for coords in coordinates] # NOTE: Uncomment to get all available weather data for each location # data_names = [ # "weather_code", "temperature_2m_max", "temperature_2m_min", # "apparent_temperature_max", "apparent_temperature_min", # "sunrise", "sunset", "daylight_duration", "sunshine_duration", # "uv_index_max", "uv_index_clear_sky_max", "precipitation_sum", # "rain_sum", "showers_sum", "snowfall_sum", "precipitation_hours", # "precipitation_probability_max", "wind_speed_10m_max", # "wind_gusts_10m_max", "wind_direction_10m_dominant", # "shortwave_radiation_sum", "et0_fao_evapotranspiration" # ] data_names = ["temperature_2m_max", "temperature_2m_min", "precipitation_sum"] #NOTE: OpenMeteo API has a limit of 10000 requests per day, # so in a production environment it would be wise to change to # an enterprise account OR utilize the clients for the requests url = "https://api.open-meteo.com/v1/forecast" params = { "latitude": latitude, "longitude": longitude, "daily": data_names } responses = openmeteo_client.weather_api(url, params=params) for idx, response in enumerate(responses, start=1): location = coordinates[idx-1][3] idx = coordinates[idx-1][0] daily = response.Daily() daily_data = { variable: daily.Variables(i).ValuesAsNumpy().tolist() if variable not in ["sunrise", "sunset"] else daily.Variables(i).ValuesAsNumpy() for i, variable in enumerate(data_names) } dates = pd.date_range(start = pd.to_datetime(daily.Time(), unit = "s", utc = True), end = pd.to_datetime(daily.TimeEnd(), unit = "s", utc = True), freq = pd.Timedelta(seconds = daily.Interval()), inclusive = "left").tolist() daily_data["date"] = [date.strftime("%d/%m/%Y") for date in dates] daily_data["location_name"] = location daily_data["resp_coordinates"] = [response.Latitude(), response.Longitude()] data_dict[idx] = daily_data return data_dict