update new opt algorithm

app.py

@@ -34,7 +34,8 @@ with gr.Blocks() as demo:
     with gr.Group() as main_group:
         with gr.Tab("EV Charger Stations"):
             plot_type = gr.Radio(['Provider', 'Charging Type'], value="Provider", label='Map type', 
-                                 info="Map type plots separatly EV station by providers, charging type or both.", interactive=True)
+                                 info="Map type plots separatly EV station by providers, charging type or both.", 
+                                 interactive=True)
             gr_fmap = Folium(fmap_provider)
 
         with gr.Tab("EV Charger along the Route"):
@@ -47,10 +48,12 @@ with gr.Blocks() as demo:
                 gr.DownloadButton(label="Setting Configuration", interactive=False)
             with gr.Row():
                 search_type =  gr.Radio(['Place Name', 'Coordinate'], value="Place Name", label='Search by', 
-                                        info="Search address by place name or coordinate in format of (lat,long).", interactive=True)
+                                        info="Search address by place name or coordinate in format of (lat,long).", 
+                                        interactive=True)
                 plotcircle_radius =  gr.Radio(['Without Circle', 'With Circle'], value="Without Circle", label='Plot Circle', 
                                               info="Plot a map with(or without) a circle for each point returned by the Google API.", interactive=True)
-                radius = gr.Slider(minimum=0.1, maximum=10, value=1, step=0.1, label="Circle Radius (Km.)", info='Adjust the radius of the circle.', interactive=True)
+                radius = gr.Slider(minimum=0.1, maximum=10, value=3.5, step=0.1, label="Circle Radius (Km.)", 
+                                   info='Adjust the radius of the circle.', interactive=True)
             with gr.Row():
                 submit_btn = gr.Button("Submit", variant='primary')
             with gr.Row():
@@ -83,12 +86,25 @@ with gr.Blocks() as demo:
             with gr.Row():
                 gr.DownloadButton(label="Input Battery Details", interactive=False)
             with gr.Row():
-                opt_battery_capacity = gr.Number(value=100, label='Battery capacity (kWh):', info='Input battery capacity in kWh.')
-                opt_battery_initial = gr.Slider(minimum=0, maximum=100, value=50, step=0.1, label="Current battery level (%)", info='Input your battery level in %.', interactive=True)
-                opt_battery_arrival = gr.Slider(minimum=0, maximum=100, value=50, step=0.1, label="Desired battery level at destination (%)", info='Input your desired battery level at destination in %.', interactive=True)
-            with gr.Row():
-                opt_charging_ports = gr.CheckboxGroup(["Type1", "Type2", "CCS2", "CHAdeMO", 'Superchargers'], value=["Type2", "CCS2", "CHAdeMO"], label="Usable charging connectors", info="Select your usable charging connectors.",interactive=True)
-                opt_method = gr.Radio(['forward', 'backward'], value="forward", label='Optimization Method', info="Select your optimization method.", interactive=True)
+                opt_battery_capacity = gr.Number(value=60, label='Battery capacity (kWh):', 
+                                                 info='Input battery capacity in kWh.')
+                opt_battery_initial = gr.Slider(minimum=0, maximum=100, value=50, step=0.1, 
+                                                label="Current battery level (%)", info='Input your current battery level in %.', 
+                                                interactive=True)
+                opt_battery_arrival = gr.Slider(minimum=0, maximum=100, value=50, step=0.1, 
+                                                label="Desired battery level at destination (%)", 
+                                                info='Input your desired battery level at destination in %.', interactive=True)
+                opt_reserve_battery = gr.Slider(minimum=0, maximum=100, value=10, step=0.1, label="Reserve battery level (%)",
+                                                info='Battery arrival of each station will be always grater than reserve battery.', 
+                                                interactive=True)
+            with gr.Row():
+                opt_charging_ports = gr.CheckboxGroup(["Type2", "CCS2", "CHAdeMO"], 
+                                                      value=["Type2", "CCS2", "CHAdeMO"], label="Usable charging connectors", 
+                                                      info="Select your usable charging connectors.",interactive=True)
+                opt_provider_filter = gr.CheckboxGroup(['PTT', 'PEA', 'EleX', 'Altervim', 'EA'], 
+                                                       value=['PTT', 'PEA', 'EleX', 'Altervim', 'EA'], 
+                                                       label="Charging providers", info="Select your charging providers.",
+                                                       interactive=True)
             with gr.Row():
                 opt_submit_btn = gr.Button("Submit", variant='primary')
             with gr.Row():
@@ -111,9 +127,9 @@ with gr.Blocks() as demo:
                      [origin_address, destination_address, search_type, radius, plotcircle_radius], 
                      [fmap_route, total_distance, total_duration, route_df, stations_df, file1, file2])
     
-    opt_submit_btn.click(lambda originaddress, destinationaddress, batcapacity, batinit, batarival, ports, method :
-                         plot_optimization(originaddress, destinationaddress, batcapacity, batinit, batarival, ports, method,gmaps=gmaps, data=data),
-                         [opt_origin_address, opt_destination_address, opt_battery_capacity, opt_battery_initial, opt_battery_arrival, opt_charging_ports, opt_method], 
+    opt_submit_btn.click(lambda originaddress, destinationaddress, batcapacity, batinit, batarival, batreserve, ports, provider :
+                         plot_optimization(originaddress, destinationaddress, batcapacity, batinit, batarival, batreserve, ports, provider, gmaps=gmaps, data=data),
+                         [opt_origin_address, opt_destination_address, opt_battery_capacity, opt_battery_initial, opt_battery_arrival, opt_reserve_battery, opt_charging_ports, opt_provider_filter], 
                          [opt_fmap, opt_distance, opt_driving_time, opt_charging_time, opt_total_time])
 
 demo.launch(debug=False)

requirements.txt

@@ -9,4 +9,5 @@ xlsxwriter
 python-dotenv
 requests
 numpy
+geopy
 https://gradio-builds.s3.amazonaws.com/a57e34ef87d24f40f09380b7b71a052f120a19fe/gradio-4.19.2-py3-none-any.whl

utils.py

@@ -14,6 +14,7 @@ import re
 import os
 import requests
 import numpy as np
+from geopy import distance
 from typing import Any, Union
 from dotenv import load_dotenv
 
@@ -145,20 +146,17 @@ def convert_string_to_tuple(input_string:str) -> tuple[float,float]:
         except:
             gr.Warning(f"'{input_string}' Worng Format. coordinate should be (lat, long). Please check your coordinate or try to use Place Name option instead.")
 
-def geocode_place(place_name:str, gmaps:googlemaps.Client, warning_text:str='Location Not Found. Please check your place name or try to use Coordinate option instead.') -> tuple[float,float]:
-    try:
-        # Geocode the place name to get its latitude and longitude
-        geocode_result = gmaps.geocode(place_name)
+def geocode_place(place_name:str, gmaps:googlemaps.Client, warning_text:str) -> tuple[float,float]:
+    # Geocode the place name to get its latitude and longitude
+    geocode_result = gmaps.geocode(place_name)
 
-        if geocode_result:
-            location = geocode_result[0]['geometry']['location']
-            lat, lng = location['lat'], location['lng']
-            return tuple([lat, lng])
-        else:
-            raise ValueError()
-        
-    except:
+    if geocode_result:
+        location = geocode_result[0]['geometry']['location']
+        lat, lng = location['lat'], location['lng']
+        return tuple([lat, lng])
+    else:
         gr.Warning(f"'{place_name}' {warning_text}")
+        return None
 
 def get_polyline(origin_address:tuple[float,float], destination_address:tuple[float,float], gmaps:googlemaps.Client) -> list[float]:
     directions_result = gmaps.directions(origin_address, destination_address, mode='driving')
@@ -171,25 +169,65 @@ def get_polyline(origin_address:tuple[float,float], destination_address:tuple[fl
 
     return all_polyline
 
+# ********************** Utils Fucntion for Optimization **********************
+def request_optimize(origin_address: str, destination_address: str, battery_capacity: float, battery_initial: float, 
+                     battery_destination: float, reserve_battery: float, usable_battype: str, provider_filter:str) -> requests.Response:
+    
+    optimizer_api_url = os.getenv('OPTIMIZER_API_URL')
+    headers = {
+        'Content-Type': 'application/json' 
+    }
+    data = {
+                "origin_address": origin_address,
+                "destination_address": destination_address,
+                "usable_battype": usable_battype,
+                "provider_filter": provider_filter,
+                "battery_capacity": battery_capacity,
+                "battery_initial": battery_initial,
+                "battery_destination": battery_destination,
+                "reserve_battery": reserve_battery,
+                "epochs": 10,
+                "temperature": 10,
+                "top_k": 10,
+                "radius_km": 3.5
+            }
+
+    response = requests.post(optimizer_api_url, json=data, headers=headers)
+
+    return response
+
+def hours_to_hours_minutes(time_in_hours):
+    # Extract whole hours
+    hours = int(time_in_hours)
+    # Compute remaining minutes
+    minutes = int((time_in_hours - hours) * 60)
+    return f"{hours} hours {minutes} minutes"
+
 # ********************** Optimization Function **********************
 def plot_optimization(origin_address: str, destination_address: str, battery_capacity: float, 
-                      battery_initial: float, battery_arrival: float, charging_ports: list, method: str,  gmaps:googlemaps.Client,
-                      data:pd.DataFrame, progress=gr.Progress()) -> tuple:
+                      battery_initial: float, battery_destination: float, reserve_battery: float, 
+                      usable_battype: list, provider_filter:list,  gmaps:googlemaps.Client,
+                      data:pd.DataFrame, progress:gr.Progress=gr.Progress()) -> tuple:
     
-    def validate_inputs(origin_address, destination_address, battery_capacity, battery_initial, battery_arrival, charging_ports):
+    def validate_inputs(origin_address, destination_address, battery_capacity, battery_initial, 
+                        battery_destination, usable_battype, provider_filter):
         errors = []
         if origin_address == '':
             errors.append("Origin address is required.")
         if destination_address == '':
             errors.append("Destination address is required.")
+        if origin_address == destination_address:
+            errors.append('Origin and destination address must be different.')
         if battery_capacity <= 0:
             errors.append("Battery capacity must be greater than 0.")
         if battery_initial == 0:
             errors.append("Battery initial state must be greater than 0.")
-        if battery_arrival == 0:
-            errors.append("Battery arrival state must be greater than 0.")
-        if len(charging_ports) == 0:
+        if battery_destination == 0:
+            errors.append("Battery destination state must be greater than 0.")
+        if len(usable_battype) == 0:
             errors.append("At least one charging port must be selected.")
+        if len(provider_filter) == 0:
+            errors.append("At least one provider must be selected.")
 
         if errors:
             error_message = "Please correct the following errors:\n" + "\n".join(errors)
@@ -221,7 +259,8 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
         # Format the result as a string
         return f"{total_hours} hours {remaining_minutes} minutes"
     
-    error = validate_inputs(origin_address, destination_address, battery_capacity, battery_initial, battery_arrival, charging_ports)
+    error = validate_inputs(origin_address, destination_address, battery_capacity, battery_initial, 
+                            battery_destination, usable_battype, provider_filter)
     if error is not None:
         gr.Warning(error)
 
@@ -229,15 +268,23 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
     
     else:
         
-        ports_str = ''
-        for i in charging_ports:
-            ports_str = ports_str + i + ', '
-        ports_str = ports_str[:-2].replace(' ','')
-
-        responce = request_optimize(origin_address, destination_address, battery_capacity, battery_initial, battery_arrival, ports_str, method)
-
-        if responce is None or responce['solution data'] is None:
-            gr.Warning('API ERROR. Please try again later or check your input.')
+        usable_battype_str = ''
+        for i in usable_battype:
+            usable_battype_str = usable_battype_str + i + ', '
+        usable_battype_str = usable_battype_str[:-2].replace(' ','')
+
+        provider_str = ''
+        for i in provider_filter:
+            provider_str = provider_str + i + ', '
+        provider_str = provider_str[:-2].replace(' ','')
+
+        responce = request_optimize(origin_address=origin_address, destination_address=destination_address, 
+                                    battery_capacity=battery_capacity, battery_initial=battery_initial, 
+                                    battery_destination=battery_destination, reserve_battery=reserve_battery,
+                                    usable_battype=usable_battype_str, provider_filter=provider_str)
+        
+        if responce.status_code == 500:
+            gr.Warning('INTERNAL API SERVER ERROR. Please try again later or check your input.')
             
             try:
                 geocode_place(origin_address, gmaps, warning_text='Location Not Found. Please check your place name.')
@@ -245,9 +292,24 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
             except:
                 pass
 
+            return None, None, None, None, None
+        
+        if responce.status_code == 404:
+            origin_address_coordinate = geocode_place(origin_address, gmaps, warning_text='Location Not Found. Please check your place name.')
+            destination_address_coordinate = geocode_place(destination_address, gmaps, warning_text='Location Not Found. Please check your place name.')
+            distance_km = distance.distance(origin_address_coordinate, destination_address_coordinate).km
+
+            if battery_initial - (distance_km*0.2) >= battery_destination:
+                gr.Warning('Your battery initial state can reach your destination without charging.')
+            elif battery_initial - (distance_km*0.2) < battery_destination and origin_address_coordinate is not None and destination_address_coordinate is not None:
+                gr.Warning('Solution not found. Try to change your input with this suggestion. \n1.Try to increse battery initial state. \n2. Try to decrease battery destination state. \n3. Try to decrease reserve battery. \n4. Try to expand usable charging connectors. \n5. Try to expand provider filter.')
+            
+                return None, None, None, None, None
+
         # get data
         ########################## data preprocessing ###########################
         try:
+            responce = responce.json()
             solution_data = responce['solution data'] 
             solution_detail = responce['solution detail']
             solution_path = responce['solution path']
@@ -262,15 +324,18 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
         ######################### plot map ###########################
         icon_url = icons()
         
-        origin_address = geocode_place(origin_address, gmaps)
-        destination_address = geocode_place(destination_address, gmaps)
+        if solution_path is not None:
+            origin_address = geocode_place(origin_address, gmaps, 'Location Not Found. Please check your place name.')
+            destination_address = geocode_place(destination_address, gmaps, 'Location Not Found. Please check your place name.')
         try:
             directions_result = gmaps.directions(origin_address, destination_address, mode='driving')
         except:
-            pass
-
+            return None, None, None, None, None
+        
         total_distance = directions_result[0]['legs'][0]['distance']['text']
         total_duration = directions_result[0]['legs'][0]['duration']['text']
+        if 'hours' not in total_duration:
+            total_duration = '0 hours ' + total_duration
 
         fmap = folium.Map(location=origin_address, tiles=None, zoom_start=13)
         # add Openstreetmap layer
@@ -279,7 +344,7 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
         ######################## plot start-stop location ###########################
         addresses = [directions_result[0]['legs'][0]['start_address'], directions_result[0]['legs'][0]['end_address']]
         locations = [origin_address, destination_address]
-        battery_state = [round(battery_initial, 1), round(battery_arrival, 1)]
+        battery_state = [round(battery_initial, 1), round(battery_destination, 1)]
         start_stop = ['start', 'stop']
         icon_start_stop = ['house', 'flag']
 
@@ -337,8 +402,8 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
                         
                         html = f"""
                                     <h3>{solution_detail[i]['Name']}</h3>
-                                    <p>Arival battery : {round((solution_data['arrival battery'][i]/battery_capacity)*100, 1)}</p>
-                                    <p>Target battery : {round((solution_data['target battery'][i]/battery_capacity)*100, 1)}</p>
+                                    <p>Arival battery : {round(solution_data['arrival battery'][i], 1)}</p>
+                                    <p>Target battery : {round(solution_data['target battery'][i], 1)}</p>
                                     <p>Charging time : {hours_to_hours_minutes(solution_data['charging time'][i])}</p>
                                     <p>Address : {solution_detail[i]['Address']}</p>
                                     <p>Coordinates : {solution_detail[i]['Coordinates']}</p>
@@ -364,61 +429,28 @@ def plot_optimization(origin_address: str, destination_address: str, battery_cap
 
         if solution_path == ['START', 'STOP']:
             gr.Warning('You can reach the destination with your current battery. There is no need to optimize.')
-            charging_time = 0
+            charging_time = 'No charging time'
             total_time = total_duration
         elif solution_path != ['START', 'STOP'] and solution_data is not None:
             charging_time = hours_to_hours_minutes(np.array(solution_data['charging time']).sum())
             total_time = add_time_strings(charging_time, total_duration)
         else:
-            charging_time = 0
+            charging_time = 'No charging time'
             total_time = total_duration
 
         return fig, total_distance, total_duration, charging_time, total_time
-
-# ********************** Utils Fucntion for Optimization **********************
-def request_optimize(origin_address: str, destination_address: str, battery_capacity: float, battery_initial: float, 
-                     battery_arrival: float, charging_ports: str, method: str) -> dict:
-    
-    optimizer_api_url = os.getenv('OPTIMIZER_API_URL')
-    headers = {
-        'Content-Type': 'application/json' 
-    }
-    data = {
-        "origin_address": origin_address,
-        "destination_address": destination_address,
-        "battery_initial": battery_initial,
-        "battery_arrival": battery_arrival,
-        "battery_capacity": battery_capacity,
-        "charging_ports": charging_ports,
-        "reserve_batterry": 10,
-        "provider_filter": "PTT,PEA,EleX,Altervim,EA",
-        "method": method,
-    }
-
-    response = requests.post(optimizer_api_url, json=data, headers=headers)
-
-    if response.status_code == 200:
-        return response.json()
-    else:
-        return None
-
-def hours_to_hours_minutes(time_in_hours):
-    # Extract whole hours
-    hours = int(time_in_hours)
-    # Compute remaining minutes
-    minutes = int((time_in_hours - hours) * 60)
-    return f"{hours} hours {minutes} minutes"
     
 # ********************** Find all station near route **********************
 def find_all_station_near_route(origin_address:str, destination_address:str, data:pd.DataFrame, 
                                 search_type:str, gmaps:googlemaps.Client, radius_km:int=1, 
-                                plotcircle_radius:str='Without Circle', progress=gr.Progress()) -> branca.element.Figure:
+                                plotcircle_radius:str='Without Circle', 
+                                progress:gr.Progress=gr.Progress()) -> branca.element.Figure:
 
     icon_url = icons()
     
     if search_type == 'Place Name':
-        origin_address = geocode_place(origin_address, gmaps)
-        destination_address = geocode_place(destination_address, gmaps)
+        origin_address = geocode_place(origin_address, gmaps, 'Location Not Found. Please check your place name or try to use Coordinate option instead.')
+        destination_address = geocode_place(destination_address, gmaps, 'Location Not Found. Please check your place name or try to use Coordinate option instead.')
     
     elif search_type == 'Coordinate':
         origin_address = convert_string_to_tuple(origin_address)
@@ -567,7 +599,8 @@ def fetchdata(collection_name:str) -> pd.DataFrame:
     data = clean_data(data)
     return data
 
-def plot_fmap_provider(data:pd.DataFrame, center:list[float,float], show_progress:bool=False, progress=gr.Progress()) -> branca.element.Figure:
+def plot_fmap_provider(data:pd.DataFrame, center:list[float,float], show_progress:bool=False, 
+                       progress:gr.Progress=gr.Progress()) -> branca.element.Figure:
     #***************************** fmap1 for Provider Map *****************************
 
     icon_url = icons()
@@ -625,7 +658,8 @@ def plot_fmap_provider(data:pd.DataFrame, center:list[float,float], show_progres
     
     return fig1
 
-def plot_fmap_chargetype(data:pd.DataFrame, center:list[float,float], show_progress:bool=False, progress=gr.Progress()) -> branca.element.Figure:
+def plot_fmap_chargetype(data:pd.DataFrame, center:list[float,float], show_progress:bool=False, 
+                         progress:gr.Progress=gr.Progress()) -> branca.element.Figure:
     #***************************** fmap2 for Chage Type Map *****************************
     icon_url = icons()
     # create charge type feature groups