chore: Update TTS dependencies and remove unused imports

kitt/core/__init__.py

@@ -1,12 +1,12 @@
 import os
-from collections import namedtuple
-import time
 import pathlib
+import time
+from collections import namedtuple
 from typing import List
 
 import numpy as np
 import torch
-# from TTS.api import TTS
+from TTS.api import TTS
 
 os.environ["COQUI_TOS_AGREED"] = "1"
 
@@ -18,7 +18,10 @@ file_full_path = pathlib.Path(os.path.realpath(__file__)).parent
 
 voices = [
     Voice(
-        "Fast", neutral=None, angry=None, speed=1.0,
+        "Fast",
+        neutral="empty",
+        angry=None,
+        speed=1.0,
     ),
     Voice(
         "Attenborough",

kitt/core/model.py

@@ -2,20 +2,21 @@ import ast
 import json
 import re
 import uuid
+import xml.etree.ElementTree as ET
 from enum import Enum
 from typing import List
-import xml.etree.ElementTree as ET
 
 from langchain.memory import ChatMessageHistory
-from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
-from langchain_core.utils.function_calling import convert_to_openai_tool
 from langchain.tools.base import StructuredTool
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.utils.function_calling import convert_to_openai_tool
+from loguru import logger
 from ollama import Client
 from pydantic import BaseModel
-from loguru import logger
 
 from kitt.skills import vehicle_status
 from kitt.skills.common import config
+
 from .validator import validate_function_call_schema
 
 
@@ -83,8 +84,9 @@ Once you have called a function, results will be fed back to you within <tool_re
 Don't make assumptions about tool results if <tool_response> XML tags are not present since function hasn't been executed yet.
 Analyze the data once you get the results and call another function.
 At each iteration please continue adding the your analysis to previous summary.
-Your final response should directly answer the user query. Don't tell what you are doing, just do it.
-Keep your responses very concise and to the point. Don't provide any unnecessary information. Don't refer to user preferences as <user_preferences>.
+Your final response should directly answer the user query. Don't tell what you are doing, just do it. Do your best to keep your responses to about 1 line. Avoid asking follow up questions as much as possible.
+Keep your responses very concise and to the point. Don't provide any unnecessary information. Do not offer to help with anything other than the user query.
+Don't refer to user preferences as <user_preferences>.
 
 
 Tools:
@@ -131,6 +133,16 @@ Assistant:
 {{"arguments": {{"destination": "Paris"}}, "name": "set_vehicle_destination"}}
 </tool_call>
 
+Example 5:
+User: Which place is warmer and by how much, dubai or tokyo?
+Assistant:
+<tool_call>
+{{"arguments": {{"location": "Tokyo"}}, "name": "get_weather"}}
+</tool_call>
+<tool_call>
+{{"arguments": {{"location": "Dubai"}}, "name": "get_weather"}}
+</tool_call>
+
 
 Instructions:
 At the very first turn you don't have <tool_results> so you shouldn't not make up the results.
@@ -228,9 +240,6 @@ def get_prompt(template, history, tools, schema, user_preferences, car_status=No
     return prompt
 
 
-
-
-
 def run_inference_ollama(prompt):
     data = {
         "prompt": prompt,

kitt/core/tts.py

@@ -1,14 +1,14 @@
 from collections import namedtuple
-from replicate import Client
-from loguru import logger
-from kitt.skills.common import config
-import torch
 
-from parler_tts import ParlerTTSForConditionalGeneration
-from transformers import AutoTokenizer
 import soundfile as sf
+import torch
+from loguru import logger
 from melo.api import TTS as MeloTTS
+from parler_tts import ParlerTTSForConditionalGeneration
+from replicate import Client
+from transformers import AutoTokenizer
 
+from kitt.skills.common import config
 
 replicate = Client(api_token=config.REPLICATE_API_KEY)
 
@@ -16,7 +16,10 @@ Voice = namedtuple("voice", ["name", "neutral", "angry", "speed"])
 
 voices_replicate = [
     Voice(
-        "Fast", neutral=None, angry=None, speed=1.0,
+        "Fast",
+        neutral="empty",
+        angry=None,
+        speed=1.0,
     ),
     Voice(
         "Attenborough",

kitt/core/utils.py

@@ -1,11 +1,11 @@
 import json
 import re
-from typing import List, Tuple, Optional, Union
+from typing import List, Optional, Tuple, Union
 
 
 def plot_route(points, vehicle: Union[tuple[float, float], None] = None):
     import plotly.express as px
-    
+
     lats = []
     lons = []
 
@@ -15,9 +15,7 @@ def plot_route(points, vehicle: Union[tuple[float, float], None] = None):
     # fig = px.line_geo(lat=lats, lon=lons)
     # fig.update_geos(fitbounds="locations")
 
-    fig = px.line_mapbox(
-        lat=lats, lon=lons, zoom=12, height=600, color_discrete_sequence=["red"]
-    )
+    fig = px.line_mapbox(lat=lats, lon=lons, color_discrete_sequence=["red"])
 
     if vehicle:
         fig.add_trace(
@@ -33,21 +31,21 @@ def plot_route(points, vehicle: Union[tuple[float, float], None] = None):
         # mapbox_zoom=12,
     )
     fig.update_geos(fitbounds="locations")
-    fig.update_layout(margin={"r": 20, "t": 20, "l": 20, "b": 20})
+    fig.update_layout(height=600, margin={"r": 20, "t": 20, "l": 20, "b": 20})
     return fig
 
 
 def extract_json_from_markdown(text):
     """
     Extracts the JSON string from the given text using a regular expression pattern.
-    
+
     Args:
         text (str): The input text containing the JSON string.
-        
+
     Returns:
         dict: The JSON data loaded from the extracted string, or None if the JSON string is not found.
     """
-    json_pattern = r'```json\r?\n(.*?)\r?\n```'
+    json_pattern = r"```json\r?\n(.*?)\r?\n```"
     match = re.search(json_pattern, text, re.DOTALL)
     if match:
         json_string = match.group(1)
@@ -58,4 +56,4 @@ def extract_json_from_markdown(text):
             print(f"Error decoding JSON string: {e}")
     else:
         print("JSON string not found in the text.")
-    return None
+    return None

kitt/skills/poi.py

@@ -1,8 +1,10 @@
 import json
 import urllib.parse
+
 import requests
-from loguru import logger
 from langchain.tools import tool
+from loguru import logger
+
 from .common import config, vehicle
 
 
@@ -20,7 +22,7 @@ def _select_equally_spaced_coordinates(coords, number_of_points=10):
 
 
 @tool
-def search_points_of_interest(search_query: str ="french restaurant"):
+def search_points_of_interest(search_query: str = "french restaurant"):
     """
     Get some of the closest points of interest matching the query.
 
@@ -47,7 +49,7 @@ def search_points_of_interest(search_query: str ="french restaurant"):
         "lon": lon,
         "radius": 5000,
         "idxSet": "POI",
-        "limit": 50
+        "limit": 50,
     }
 
     r = requests.get(url, params=params, timeout=5)
@@ -76,7 +78,7 @@ def search_points_of_interest(search_query: str ="french restaurant"):
     output = (
         f"There are {len(results)} options in the vicinity. The most relevant are: "
     )
-    return output + ".\n ".join(formatted_results), results[:3]
+    return output + ".\n ".join(formatted_results), [x["poi"] for x in results[:3]]
 
 
 def find_points_of_interest(lat="0", lon="0", type_of_poi="restaurant"):
@@ -96,7 +98,6 @@ def find_points_of_interest(lat="0", lon="0", type_of_poi="restaurant"):
 
     r = requests.get(url, timeout=5)
 
-
     # Parse JSON from the response
     data = r.json()
     # print(data)

kitt/skills/routing.py

@@ -1,7 +1,9 @@
 from datetime import datetime
+
 import requests
-from loguru import logger
 from langchain.tools import tool
+from loguru import logger
+
 from .common import config, vehicle
 
 
@@ -12,13 +14,29 @@ def find_coordinates(address):
     """
     # https://developer.tomtom.com/geocoding-api/documentation/geocode
     url = f"https://api.tomtom.com/search/2/geocode/{address}.json?key={config.TOMTOM_API_KEY}"
-    response = requests.get(url)
+    response = requests.get(url, timeout=5)
     data = response.json()
     lat = data["results"][0]["position"]["lat"]
     lon = data["results"][0]["position"]["lon"]
     return lat, lon
 
 
+def find_address(lat, lon):
+    """
+    Find the address of a specific location.
+
+    Args:
+    lat (string): Required. The latitude
+    lon (string): Required. The longitude
+    """
+    # https://developer.tomtom.com/search-api/documentation/reverse-geocoding
+    url = f"https://api.tomtom.com/search/2/reverseGeocode/{lat},{lon}.json?key={config.TOMTOM_API_KEY}"
+    response = requests.get(url, timeout=5)
+    data = response.json()
+    address = data["addresses"][0]["address"]["freeformAddress"]
+    return address
+
+
 def calculate_route(origin, destination):
     """This function is called when the origin or destination is updated in the GUI. It calculates the route between the origin and destination."""
     print(f"calculate_route(origin: {origin}, destination: {destination})")
@@ -37,7 +55,7 @@ def calculate_route(origin, destination):
     # destination = "49.586745,6.140002"
 
     url = f"https://api.tomtom.com/routing/1/calculateRoute/{orig_coords_str}:{dest_coords_str}/json?key={config.TOMTOM_API_KEY}"
-    response = requests.get(url)
+    response = requests.get(url, timeout=5)
     data = response.json()
     points = data["routes"][0]["legs"][0]["points"]
 
@@ -150,7 +168,6 @@ def find_route(destination):
     )
     return _format_tomtom_trip_info(trip_info, destination)
 
-    
     # raw_response["routes"][0]["legs"][0]["points"]
 
 
@@ -178,4 +195,4 @@ def _format_tomtom_trip_info(trip_info, destination="destination"):
     arrival_hour_display = arrival_time.strftime("%H:%M")
 
     # return the distance and time
-    return f"The route to {destination} is {distance_km:.2f} km which takes {time_display}. Leaving now, the arrival time is estimated at {arrival_hour_display}."
+    return f"The route to {destination} is {distance_km:.2f} km which takes {time_display}. Leaving now, the arrival time is estimated at {arrival_hour_display}."

kitt/skills/weather.py

@@ -1,6 +1,6 @@
 import requests
-from loguru import logger
 from langchain.tools import tool
+from loguru import logger
 
 from .common import config, vehicle
 

main.py

@@ -1,49 +1,65 @@
 import time
+
 import gradio as gr
 import numpy as np
+import ollama
 import torch
 import torchaudio
-from transformers import pipeline
 import typer
-
-from kitt.skills.common import config, vehicle
-from kitt.skills.routing import calculate_route
-from kitt.core.tts import run_tts_replicate, run_tts_fast, run_melo_tts
-import ollama
-
-from langchain.tools.base import StructuredTool
 from langchain.memory import ChatMessageHistory
-from langchain_core.utils.function_calling import convert_to_openai_tool
 from langchain.tools import tool
+from langchain.tools.base import StructuredTool
+from langchain_core.utils.function_calling import convert_to_openai_tool
 from loguru import logger
+from transformers import pipeline
 
+from kitt.core import tts_gradio
+from kitt.core import utils as kitt_utils
+from kitt.core import voice_options
 
+# from kitt.core.model import process_query
+from kitt.core.model import generate_function_call as process_query
+from kitt.core.tts import run_melo_tts, run_tts_fast, run_tts_replicate
 from kitt.skills import (
-    get_weather,
+    code_interpreter,
+    date_time_info,
+    do_anything_else,
+    extract_func_args,
     find_route,
     get_forecast,
-    vehicle_status as vehicle_status_fn,
-    set_vehicle_speed,
-    search_points_of_interest,
+    get_weather,
+    get_weather_current_location,
     search_along_route_w_coordinates,
+    search_points_of_interest,
     set_vehicle_destination,
-    do_anything_else,
-    date_time_info,
-    get_weather_current_location,
-    code_interpreter,
+    set_vehicle_speed,
 )
-from kitt.skills import extract_func_args
-from kitt.core import voice_options, tts_gradio
-
-# from kitt.core.model import process_query
-from kitt.core.model import generate_function_call as process_query
-from kitt.core import utils as kitt_utils
+from kitt.skills import vehicle_status as vehicle_status_fn
+from kitt.skills.common import config, vehicle
+from kitt.skills.routing import calculate_route, find_address
 
+ORIGIN = "Mondorf-les-Bains, Luxembourg"
+DESTINATION = "Rue Alphonse Weicker, Luxembourg"
+DEFAULT_LLM_BACKEND = "ollama"
+ENABLE_HISTORY = True
+ENABLE_TTS = True
+TTS_BACKEND = "local"
+USER_PREFERENCES = "User loves italian food."
 
 global_context = {
     "vehicle": vehicle,
     "query": "How is the weather?",
     "route_points": [],
+    "origin": ORIGIN,
+    "destination": DESTINATION,
+    "enable_history": ENABLE_HISTORY,
+    "tts_enabled": ENABLE_TTS,
+    "tts_backend": TTS_BACKEND,
+    "llm_backend": DEFAULT_LLM_BACKEND,
+    "map_origin": ORIGIN,
+    "map_destination": DESTINATION,
+    "update_proxy": 0,
+    "map": None,
 }
 
 speaker_embedding_cache = {}
@@ -72,8 +88,6 @@ Answer questions concisely and do not mention what you base your reply on.<|im_e
 <|im_start|>assistant
 """
 
-USER_PREFERENCES = "I love italian food\nI like doing sports"
-
 
 def get_prompt(template, input, history, tools):
     # "vehicle_status": vehicle_status_fn()[0]
@@ -221,7 +235,7 @@ def run_llama3_model(query, voice_character, state):
     if state["tts_enabled"]:
         # voice_out = run_tts_replicate(output_text, voice_character)
         # voice_out = run_tts_fast(output_text)[0]
-        voice_out = run_melo_tts(output_text, voice_character)    
+        voice_out = run_melo_tts(output_text, voice_character)
         # voice_out = tts_gradio(output_text, voice_character, speaker_embedding_cache)[0]
     return (
         output_text,
@@ -245,33 +259,47 @@ def run_model(query, voice_character, state):
 
     if not state["enable_history"]:
         history.clear()
-    return text, voice, vehicle.model_dump_json()
+    global_context["update_proxy"] += 1
+
+    return (
+        text,
+        voice,
+        vehicle.model_dump_json(),
+        state,
+        dict(update_proxy=global_context["update_proxy"]),
+    )
 
 
 def calculate_route_gradio(origin, destination):
     vehicle_status, points = calculate_route(origin, destination)
     plot = kitt_utils.plot_route(points, vehicle=vehicle.location_coordinates)
     global_context["route_points"] = points
+    # state.value["route_points"] = points
     vehicle.location_coordinates = points[0]["latitude"], points[0]["longitude"]
     return plot, vehicle_status, 0
 
 
-def update_vehicle_status(trip_progress, origin, destination):
+def update_vehicle_status(trip_progress, origin, destination, state):
     if not global_context["route_points"]:
         vehicle_status, points = calculate_route(origin, destination)
         global_context["route_points"] = points
+    global_context["destination"] = destination
+    global_context["route_points"] = global_context["route_points"]
     n_points = len(global_context["route_points"])
     index = min(int(trip_progress / 100 * n_points), n_points - 1)
-    print(f"Trip progress: {trip_progress} len: {n_points}, index: {index}")
+    logger.info(f"Trip progress: {trip_progress} len: {n_points}, index: {index}")
     new_coords = global_context["route_points"][index]
     new_coords = new_coords["latitude"], new_coords["longitude"]
-    print(f"Trip progress: {trip_progress}, len: {n_points}, new_coords: {new_coords}")
+    logger.info(
+        f"Trip progress: {trip_progress}, len: {n_points}, new_coords: {new_coords}"
+    )
     vehicle.location_coordinates = new_coords
-    vehicle.location = ""
+    new_vehicle_location = find_address(new_coords[0], new_coords[1])
+    vehicle.location = new_vehicle_location
     plot = kitt_utils.plot_route(
         global_context["route_points"], vehicle=vehicle.location_coordinates
     )
-    return vehicle.model_dump_json(), plot
+    return vehicle.model_dump_json(), plot, state
 
 
 device = "cuda" if torch.cuda.is_available() else "cpu"
@@ -314,8 +342,10 @@ def save_and_transcribe_audio(audio):
 
 def save_and_transcribe_run_model(audio, voice_character, state):
     text = save_and_transcribe_audio(audio)
-    out_text, out_voice, vehicle_status = run_model(text, voice_character, state)
-    return text, out_text, out_voice, vehicle_status
+    out_text, out_voice, vehicle_status, state, update_proxy = run_model(
+        text, voice_character, state
+    )
+    return None, text, out_text, out_voice, vehicle_status, state, update_proxy
 
 
 def set_tts_enabled(tts_enabled, state):
@@ -324,6 +354,7 @@ def set_tts_enabled(tts_enabled, state):
         f"TTS enabled was {state['tts_enabled']} and changed to {new_tts_enabled}"
     )
     state["tts_enabled"] = new_tts_enabled
+    global_context["tts_enabled"] = new_tts_enabled
     return state
 
 
@@ -333,6 +364,7 @@ def set_llm_backend(llm_backend, state):
         f"LLM backend was {state['llm_backend']} and changed to {new_llm_backend}"
     )
     state["llm_backend"] = new_llm_backend
+    global_context["llm_backend"] = new_llm_backend
     return state
 
 
@@ -340,6 +372,7 @@ def set_user_preferences(preferences, state):
     new_preferences = preferences
     logger.info(f"User preferences changed to: {new_preferences}")
     state["user_preferences"] = new_preferences
+    global_context["user_preferences"] = new_preferences
     return state
 
 
@@ -349,9 +382,40 @@ def set_enable_history(enable_history, state):
         f"Enable history was {state['enable_history']} and changed to {new_enable_history}"
     )
     state["enable_history"] = new_enable_history
+    global_context["enable_history"] = new_enable_history
+    return state
+
+
+def set_tts_backend(tts_backend, state):
+    new_tts_backend = tts_backend.lower()
+    logger.info(
+        f"TTS backend was {state['tts_backend']} and changed to {new_tts_backend}"
+    )
+    state["tts_backend"] = new_tts_backend
+    global_context["tts_backend"] = new_tts_backend
     return state
 
 
+def conditional_update():
+    if global_context["destination"] != vehicle.destination:
+        global_context["destination"] = vehicle.destination
+
+    if global_context["origin"] != vehicle.location:
+        global_context["origin"] = vehicle.location
+
+    if (
+        global_context["map_origin"] != vehicle.location
+        or global_context["map_destination"] != vehicle.destination
+        or global_context["update_proxy"] == 0
+    ):
+        logger.info(f"Updating the map plot... in conditional_update")
+        map_plot, vehicle_status, _ = calculate_route_gradio(
+            vehicle.location, vehicle.destination
+        )
+        global_context["map"] = map_plot
+    return global_context["map"]
+
+
 # to be able to use the microphone on chrome, you will have to go to chrome://flags/#unsafely-treat-insecure-origin-as-secure and enter http://10.186.115.21:7860/
 # in "Insecure origins treated as secure", enable it and relaunch chrome
 
@@ -360,13 +424,6 @@ def set_enable_history(enable_history, state):
 # What's the closest restaurant from here?
 
 
-ORIGIN = "Mondorf-les-Bains, Luxembourg"
-DESTINATION = "Rue Alphonse Weicker, Luxembourg"
-DEFAULT_LLM_BACKEND = "ollama"
-ENABLE_HISTORY = True
-ENABLE_TTS = True
-
-
 def create_demo(tts_server: bool = False, model="llama3"):
     print(f"Running the demo with model: {model} and TTSServer: {tts_server}")
     with gr.Blocks(theme=gr.themes.Default()) as demo:
@@ -380,10 +437,13 @@ def create_demo(tts_server: bool = False, model="llama3"):
                 "llm_backend": DEFAULT_LLM_BACKEND,
                 "user_preferences": USER_PREFERENCES,
                 "enable_history": ENABLE_HISTORY,
+                "tts_backend": TTS_BACKEND,
+                "destination": DESTINATION,
             }
         )
-        trip_points = gr.State(value=[])
+
         plot, vehicle_status, _ = calculate_route_gradio(ORIGIN, DESTINATION)
+        global_context["map"] = plot
 
         with gr.Row():
             with gr.Column(scale=1, min_width=300):
@@ -452,6 +512,10 @@ def create_demo(tts_server: bool = False, model="llama3"):
                         label="Input text",
                         interactive=True,
                     )
+                    update_proxy = gr.JSON(
+                        value=dict(update_proxy=0),
+                        label="Global context",
+                    )
                 vehicle_status = gr.JSON(
                     value=vehicle.model_dump_json(), label="Vehicle status"
                 )
@@ -462,6 +526,12 @@ def create_demo(tts_server: bool = False, model="llama3"):
                         value="Yes" if ENABLE_TTS else "No",
                         interactive=True,
                     )
+                    tts_backend = gr.Radio(
+                        ["Local", "Replicate"],
+                        label="TTS Backend",
+                        value=TTS_BACKEND.title(),
+                        interactive=True,
+                    )
                     llm_backend = gr.Radio(
                         choices=["Ollama", "Replicate"],
                         label="LLM Backend",
@@ -505,26 +575,34 @@ def create_demo(tts_server: bool = False, model="llama3"):
         input_text.submit(
             fn=run_model,
             inputs=[input_text, voice_character, state],
-            outputs=[output_text, output_audio, vehicle_status],
+            outputs=[output_text, output_audio, vehicle_status, state, update_proxy],
         )
         input_text_debug.submit(
             fn=run_model,
             inputs=[input_text_debug, voice_character, state],
-            outputs=[output_text, output_audio, vehicle_status],
+            outputs=[output_text, output_audio, vehicle_status, state, update_proxy],
         )
 
         # Set the vehicle status based on the trip progress
         trip_progress.release(
             fn=update_vehicle_status,
-            inputs=[trip_progress, origin, destination],
-            outputs=[vehicle_status, map_plot],
+            inputs=[trip_progress, origin, destination, state],
+            outputs=[vehicle_status, map_plot, state],
         )
 
         # Save and transcribe the audio
         input_audio.stop_recording(
             fn=save_and_transcribe_run_model,
             inputs=[input_audio, voice_character, state],
-            outputs=[input_text, output_text, output_audio, vehicle_status],
+            outputs=[
+                input_audio,
+                input_text,
+                output_text,
+                output_audio,
+                vehicle_status,
+                state,
+                update_proxy,
+            ],
         )
         input_audio_debug.stop_recording(
             fn=save_and_transcribe_audio,
@@ -539,12 +617,16 @@ def create_demo(tts_server: bool = False, model="llama3"):
         tts_enabled.change(
             fn=set_tts_enabled, inputs=[tts_enabled, state], outputs=[state]
         )
+        tts_backend.change(
+            fn=set_tts_backend, inputs=[tts_backend, state], outputs=[state]
+        )
         llm_backend.change(
             fn=set_llm_backend, inputs=[llm_backend, state], outputs=[state]
         )
         enable_history.change(
             fn=set_enable_history, inputs=[enable_history, state], outputs=[state]
         )
+        update_proxy.change(fn=conditional_update, inputs=[], outputs=[map_plot])
 
     return demo