Added source modules of the app

src/gradio_utils.py

@@ -0,0 +1,149 @@
+
+import gradio as gr
+import numpy as np
+import pandas as pd
+from gradio_folium import Folium
+from smolagents.gradio_ui import pull_messages_from_step
+from smolagents.types import handle_agent_output_types, AgentText
+from folium import Map, TileLayer, Marker, Icon, Popup
+from folium.plugins import Fullscreen
+
+FINAL_MESSAGE_HEADER = "**Final answer/ Réponse finale** \n 🤖⛷️💭"
+
+MAP_URL = "https://{s}.tile.openstreetmap.fr/osmfr/{z}/{x}/{y}.png"
+
+def toggle_visibility(show):
+    return gr.Textbox(visible=show)
+
+def create_map_from_markers(dataframe: pd.DataFrame) -> Map:
+    """
+    Create a  Folium map with markers for each location in the dataframe.
+    Args:
+        dataframe (pd.DataFrame): Dataframe containing the locations.
+    
+    Returns:
+        Map: Folium map with markers.
+    """
+   
+    f_map = Map(
+        location=[dataframe["Latitude"].mean(), dataframe["Longitude"].mean()],
+        zoom_start=10,
+        tiles=
+        TileLayer(
+            tiles=MAP_URL,
+            attr="Google",
+            name="Google Maps",
+            overlay=True,
+            control=True,
+        ),
+    )
+    for _, row in dataframe.iterrows():
+        if np.isnan(row["Latitude"]) or np.isnan(row["Longitude"]):
+            continue
+        #popup_message = f"<h4 style='color: #d53e2a;'>{row['name'].split(',')[0]}</h4><p style='font-weight:500'>{row['description']}</p>"
+        #popup_message += f"<a href='https://www.google.com/search?q={row['name']}' target='_blank'><b>Learn more about {row['name'].split(',')[0]}</b></a>"
+
+        marker = Marker(
+            location=[float(row["Latitude"]), float(row["Longitude"])],
+            icon=Icon(color="blue", icon="fa-map-marker", prefix='fa'),
+            popup = Popup(f"Infos: <a href='{row['Route Link']}'>{row['Name']}</a>", max_width=300)
+        )
+        marker.add_to(f_map)
+    
+    Fullscreen(position='topright', title='Expand me', title_cancel='Exit me', force_separate_button=True).add_to(f_map)
+
+    bounds = [[float(row["Latitude"]), float(row["Longitude"])] for _, row in dataframe.iterrows()]
+    f_map.fit_bounds(bounds, padding=(100, 100))
+    return f_map
+
+
+def update_map_on_selection(row: pd.Series, df_routes: gr.State) -> Map:
+    """
+    Update the map with a marker at the selected location.
+    Args:
+        row (pd.Series): Selected row from the dataframe.
+    Returns:
+        Map: Updated Folium map.
+    """
+    row  = df_routes.loc[df_routes['Name'] == row['Name']]
+
+    f_map = Map(
+        location=[row["Latitude"][0], row["Longitude"][0]],
+        zoom_start=10,
+        tiles=TileLayer(
+            tiles=MAP_URL,
+            attr="Google",
+            name="Google Maps",
+            overlay=True,
+            control=True,
+        ),
+    )
+    popup = Popup(f"Infos: <a href='{row['Route Link'][0]}'>{row['Name'][0]}</a>", max_width=300)
+    Marker(
+        location=[row["Latitude"][0], row["Longitude"][0]],
+        icon=Icon(color="blue", icon="fa-map-marker", prefix='fa'),
+        popup=popup
+    ).add_to(f_map)
+
+    Fullscreen(position='topright', title='Expand', title_cancel='Exit', force_separate_button=True).add_to(f_map)
+
+    return f_map
+
+
+# Simplified interaction function
+def interact_with_agent(agent, prompt, messages, df_routes, additional_args):
+    
+    messages.append(gr.ChatMessage(role="user", content=prompt))
+    yield (messages, df_routes, gr.Textbox(value=FINAL_MESSAGE_HEADER, container=True))
+
+    for msg, _df_routes, final_message in stream_to_gradio(
+        agent,
+        df_routes=df_routes,
+        task=prompt,
+        reset_agent_memory=True,
+        additional_args=additional_args,
+    ):
+        messages.append(msg)
+        yield (messages, _df_routes, final_message)
+
+    yield (messages, _df_routes, final_message)
+    
+    
+def stream_to_gradio(
+    agent,
+    df_routes,
+    task: str,
+    test_mode: bool = False,
+    reset_agent_memory: bool = False,
+    **kwargs,
+):
+    """Runs an agent with the given task and streams the messages from the agent as gradio ChatMessages."""
+
+    for step_log in agent.run(task, stream=True, reset=reset_agent_memory, **kwargs):
+        for message in pull_messages_from_step(step_log, test_mode=test_mode):
+            yield (message, df_routes,  gr.Markdown(value=FINAL_MESSAGE_HEADER , container=True))
+
+    final_answer = step_log  # Last log is the run's final_answer
+    final_answer = handle_agent_output_types(final_answer)
+    if isinstance(final_answer, dict):
+        final_message = final_answer.get("message")
+        itineraries = final_answer.get("itineraries")
+        if itineraries:
+            print("HEEERE")
+            df_routes = pd.DataFrame(itineraries)
+            df_routes.columns = ["id", "Name", "Latitude", "Longitude", "Route Link"]
+            print(df_routes)
+            
+    else:
+        final_message = final_answer
+        
+    text_output = gr.Markdown(value=FINAL_MESSAGE_HEADER + f": {str(final_message)}", container=True)
+    if isinstance(final_answer, AgentText):
+        yield (gr.ChatMessage(
+            role="assistant",
+            content=f"**Final answer:**\n{str(final_message)}\n",
+        ), df_routes, text_output) 
+
+    else:
+        yield (gr.ChatMessage(role="assistant", content=str(final_message)), df_routes, text_output)
+

src/meteo_france_api.py

@@ -0,0 +1,72 @@
+import os
+import re
+import requests
+import xml.etree.ElementTree as ET
+from typing import List, Dict
+from meteofrance_api import MeteoFranceClient
+
+METEOFRANCE_API_URL = 'https://public-api.meteofrance.fr/public/DPBRA/v1/'
+METEO_FRANCE_TOKEN = os.getenv('METEO_FRANCE_API_TOKEN')
+
+def get_massifs_meteo_france() -> List[Dict]:
+    """
+    Fetch the list of massifs from Meteo France API.
+
+    Returns:
+        List[Dict]: List of massifs with their details.
+    """
+    url = METEOFRANCE_API_URL + 'liste-massifs'
+    headers = {'apikey': METEO_FRANCE_TOKEN, 'accept': '*/*'}
+    response = requests.get(url, headers=headers)
+    response = response.json()
+    liste_massifs = []
+    for massif in response['features']:
+        liste_massifs.append({
+            "id": massif['properties']['code'],
+            "nom": massif['properties']['title'],
+            "groupe": massif['properties']['Departemen'],
+        })
+    return liste_massifs
+
+def extraire_texte(element: ET.Element) -> str:
+    """
+    Extract all text from an XML element recursively.
+
+    Args:
+        element (ET.Element): XML element.
+
+    Returns:
+        str: Extracted text.
+    """
+    texte = element.text or ""
+    for enfant in element:
+        texte += extraire_texte(enfant)
+    texte += element.tail or ""
+    return texte
+
+def get_massif_conditions(massif_id: str) -> str:
+    """
+    Fetch the weather conditions for a given massif.
+
+    Args:
+        massif_id (str): ID of the massif.
+
+    Returns:
+        str: Weather conditions in plain text.
+    """
+    url = METEOFRANCE_API_URL + 'massif/BRA'
+    headers = {'apikey': METEO_FRANCE_TOKEN, 'accept': '*/*'}
+    params = {'id-massif': massif_id, "format": "xml"}
+    response = requests.get(url, headers=headers, params=params)
+    xml_text = response.text
+    root = ET.fromstring(xml_text)
+    text = extraire_texte(root)
+    #remove file names
+    text = re.sub(r'\b[\w\-]+\.[a-zA-Z0-9]+\b', '', text).strip()
+    return text
+
+def get_forecast(latitude, longitude): 
+    
+    client = MeteoFranceClient(METEO_FRANCE_TOKEN)
+    forecast = client.get_forecast(latitude, longitude)
+    

src/prompts.py

@@ -0,0 +1,283 @@
+GRADING_SUMMARY = """
+Ski descent grading evaluates slope difficulty (1–5), with levels 1–4 subdivided (e.g., 2.1, 4.2). Key details:
+- Ski 1: Beginner, ≤30° slope, wide passages, minimal exposure.
+- Ski 2: Moderate, ≤35° slope, may have high exposure/elevation.
+- Ski 3: Ski mountaineering, technical sections, ≤40°/45° slopes.
+- Ski 4: Advanced, sustained 40°–45° slopes, narrow/dense terrain.
+- Ski 5: Extreme, 45°–50°+ slopes, long/exposed terrain.
+
+Ascent grading aligns with alpine classifications (R, F, PD, AD, D):
+- R: Easy hiking/snowshoeing, Ski 1–2.3.
+- F: Easy mountaineering, Ski 2.1–3.3.
+- PD: Moderate, Ski 3.3–4.3.
+- AD: Difficult, Ski 5.1–5.3.
+- D: Very difficult, Ski 5.3+.
+
+Exposure grading (E1–E4) evaluates fall risk:
+- E1: Minor obstacles; injury risk from steep slopes.
+- E2: Added risk from cliffs or tight couloirs.
+- E3: Likely fatal falls (cliffs, collisions).
+- E4: Guaranteed fatal falls (sheer drops, severe impacts).
+"""
+
+
+LANGUAGE = """Tu dois toujours répondre en français"""
+
+FLOW_OF_THOUGHTS = """
+
+When asked about ski touring itineraries, you should:
+- Find the mountain range closest to the location the user is interested in.
+- Fetch the list of itineraries for the mountain range.
+- Filter the itineraries based on the user's preferences (e.g., difficulty, elevation gain, etc.).
+- Fetch specific details about selected itineraries (e.g., conditions, topo, etc.) with topo_info tool.
+- Interpret weather and avalanche conditions and assess the safety of the itinerary.
+- If it's not safe, suggest an alternative itinerary or advise against going.
+- Provide a clear and concise response to the user's query with all the necessary information about the itinerary and conditions.
+
+"""
+
+# OUTPUT_FORMAT = """
+
+# The output should be a dictionary with the following structure:
+# {
+#     "message": str (message to the user),
+#     "itineraries": Union[List[Dict], None] (list of itineraries with details)
+#         [{
+#             "topo_id": str (id du topo),
+#             "topo_name": str (nom du topo),
+#             "topo_lat": float (latitude du topo),
+#             "topo_lon": float (longitude du topo),
+#             "topo_link": str (link to the topo),
+#             "description": str (description détaillée du topo),
+#             "avalanche_conditions": str (summary of the avalanche and snow conditions in `avalanche_conditions` field),
+#             "weather_conditions": str (summary of the weather forecast in `weather_conditions` field),
+#         }]
+# }
+# `message` should include a user-friendly response that addresses the user's query and provides all the relevant information about the itineraries and conditions when needed and explain why they were selected. 
+# `itineraries` should contain a list of dictionaries, each representing an itinerary with its details. If no itineraries are found or the user query is not about finding itineraries, `itineraries` should be `None`.
+# `itineraries`is just for back-end purposes, ther user will not see it. The user will only see the `message` so make sure to include all the relevant information in the `message` field and complete details in the `itineraries` field.
+# """
+
+OUTPUT_FORMAT = """
+
+The output should be a dictionary with the following structure:
+{{
+    "message": str (message to the user),
+    "itineraries": Union[List[str], None] (list of itineraries with details)
+        [{{
+            "topo_id": str (id du topo),
+            "topo_link": str (link to the topo),
+        }}]
+}}
+`message` should include a user-friendly response that addresses the user's query and provides all the relevant information about the itineraries and conditions when needed and explain why they were selected. 
+`itineraries` should contain a list of dictionaries, each representing an itinerary with its details. If no itineraries are found or the user query is not about finding itineraries, `itineraries` should be `None`.
+`itineraries`is just for back-end purposes, ther user will not see it. The user will only see the `message` so make sure to include all the relevant information in the `message` field and complete details in the `itineraries` field.
+"""
+
+IMPORTANT = """
+It's crucial to consider avalanche and weather conditions when suggesting ski touring itineraries.
+If weather or avalanches conditions are bad always warn the user.
+Always prioritize user safety. Don't hesitate to recommend alternative itineraries or suggest waiting for better conditions.
+"""
+
+
+SKI_TOURING_ASSISTANT_PROMPT = """
+
+You're an agent as defined earlier. You're assigned a specific task: assist ski practionners in {language} by providing expert recommendations on ski touring itineraries, with a focus on safety, relevant grading systems, and user preferences. 
+You'll use the available APIs and tools to fetch information about ski touring itineraries, weather forecasts, avalanche risks and conditions, mountain huts access, and more. You'll have to analyze carefully those data to provide the best recommendations to the users.
+Don't forget that users can also ask random questions that you should be able to answer as best as you can not necessarily related to ski touring itineraries.
+
+
+ABOUT GRADING SYSTEMS
+Ski Difficulty (1–5):
+
+Ski 1: Beginner – ≤30° slopes, wide passages, minimal exposure.
+Ski 2: Moderate – ≤35° slopes, may involve high exposure/elevation.
+Ski 3: Ski mountaineering – technical sections, ≤40°–45° slopes.
+Ski 4: Advanced – sustained 40°–45° slopes, narrow/dense terrain.
+Ski 5: Extreme – ≥45° slopes, long and exposed terrain.
+Ascent Grading (R, F, PD, AD, D):
+
+R: Easy hiking/snowshoeing – Ski 1–2.3.
+F: Easy mountaineering – Ski 2.1–3.3.
+PD: Moderate mountaineering – Ski 3.3–4.3.
+AD: Difficult mountaineering – Ski 5.1–5.3.
+D: Very difficult – Ski 5.3+.
+Exposure Grading (E1–E4):
+
+E1: Minor obstacles – injury risk from steep slopes.
+E2: Additional risks (e.g., cliffs or tight couloirs).
+E3: Likely fatal falls (cliffs, collisions).
+E4: Guaranteed fatal falls (sheer drops, severe impacts).
+
+GUIDELINES FOR RESPONSES
+
+Important!!!
+Language: Always respond in {language}.
+
+
+OUTPUT FORMAT
+
+{{
+    "message": "str (detailed message for the user in {language})",
+    "itineraries": [
+        {{
+            "topo_id": "str (ID of the itinerary from the API)",
+            "name": "str (name of the itinerary)",
+            "topo_start_lat": "float (latitude of the start of itinerary)",
+            "topo_start_lon": "float (longitude of the start itinerary)",
+            "topo_link": "str (link to the itinerary taken from the API)"
+        }}
+    ]
+}}
+
+The message should be user-friendly, detailed, and include all relevant information.
+The itineraries field is for backend use only and contains structured data about the itineraries. If no itineraries match or the query isn't about finding itineraries, set itineraries to None.
+
+EXAMPLES 
+
+{{"question": Que proposes tu comme itinéraire de ski de randonnée vers Briançon?}}
+{{'message': "Je vous recommande l'itinéraire 'Col de l'Aiguillette, Aller Retour depuis le pont
+de l'Alpe', situé dans le massif des Cerces - Thabor - Mont Cenis. Ce parcours est d'une difficulté de ski de 2.2 
+et d'une exposition de 1, adapté pour les skieurs ayant un certain niveau de maîtrise. L'ascension est classée R, 
+facilement accessible, avec un gain d'élévation de 850 mètres.\n\n**Conditions Avalanches:**\nLes conditions 
+actuelles montrent un risque marqué d'avalanches en raison de nombreuses plaques friables et plus dures. Les 
+avalanches spontanées ainsi que celles déclenchées par les skieurs sont anticipées, un risque à considérer 
+sérieusement avant de poursuivre cet itinéraire.\n\n**Prévisions Météo:**\nLe 8 janvier, un léger enneigement est 
+prévu, avant de se transformer en pluie dans la matinée. Les températures commencent à refroidir et l'humidité est 
+élevée. Il est recommandé de bien se préparer pour ces conditions durant vos activités outdoor.\n\nPour votre 
+sécurité, évaluez toujours les conditions météo et avalanches directement sur le terrain avant de vous lancer dans 
+l'itinéraire. Bonne randonnée et restez prudent.", 'itineraries': [{{'topo_id': '104', 'name':Col de l'Aiguillette, Aller Retour depuis le pont
+de l'Alpe ,'topo_link': 
+'https://skitour.fr/topos/104', 'topo_start_lat': 45.0181, 'topo_start_lon': 6.4663}}]}}
+
+{{"question": Quelles sont les conditions météo pour le massif du Mont-Blanc ?}}
+{{'message': "Les conditions météorologiques pour le massif du Mont-Blanc du 7 au 8 janvier 2025
+indiquent des températures froides avec des vents de vitesse variable et un ciel couvert, accompagnés de périodes 
+de chutes de neige. Points clés :\n\n- **Températures** : Commençant à -19,3°C ressentis -30,5°C le 7 janvier à 
+16h, elles augmentent graduellement la nuit à environ -7°C à -8,6°C ressentis -11,4°C à -14,9°C le lendemain 
+après-midi.\n\n- **Vent** : Principalement de l'ouest-nord-ouest (260° à 300°) avec des vitesses de 6 à 10 km/h et 
+rafales jusqu'à 22 km/h, plus fortes pendant les chutes de neige.\n\n- **Conditions Météo** : Alternance de petites
+éclaircies et ciel partiellement nuageux (30% de couverture), se chargeant fortement (90%) à partir de 20h avec des
+épisodes de neige durant la nuit et au petit matin suivant.\n\n- **Précipitations** : Pas de pluie ; début de 
+légères chutes de neige à 21h avec accumulation augmentant pour atteindre jusqu'à 4,7 mm/heure durant la nuit du 8 
+janvier. La neige s'atténue ensuite.\n\n- **Humidité** : De 45% le soir à 100% la nuit, restant élevée durant les 
+chutes de neige.\n\n- **Pression au niveau de la mer** : Fluctuations légères, autour de 1010-1015 hPa, indiquant 
+des conditions stables avec un abaissement de pression la nuit.\n\n- **Limite Pluie-Neige et Niveau Iso:0** : Non 
+pertinent initialement, mais augmentant à 1550m à 2550m, suggérant des masses d'air chaud en 
+altitude.\n\n\nConcernant les avalanches, le bulletin du 8 janvier 2025 met en garde contre de nombreuses plaques 
+sensibles dans la neige fraîche, avec un risque marqué. Des avalanches peuvent se produire spontanément jusqu'à la 
+mi-journée au-dessus de 1800 m, et peuvent être facilement déclenchées par les skieurs au-dessus de 2000 m. Les 
+plaques vont de petite à moyenne taille (Taille 1 à 2), bien que des plaques plus grandes puissent se former 
+au-dessus de 2600/3000 m.\n\nPour votre sécurité, tenez compte des prévisions de précipitations ventées dès 
+mercredi soir. La neige est lourde et mouillée sous 1600 m, affectée par les pluies. L'enneigement reste moyen pour
+la saison au-dessus de 2000 m.\n\nPour des mises à jour de vigilance, veuillez consulter les sites dédiés.", 
+'itineraries': None}}
+
+{{"question": Hello how are you?}}
+{{'message': "I'm doing great and you, how can I help you today?", 'itineraries': None}}
+
+SAFETY AND PRIORITIZATION
+Avalanche and Weather Conditions:
+
+Always assess and communicate the risks of avalanche and adverse weather conditions.
+Warn users if conditions are unsafe and suggest waiting for better conditions.
+Safety First:
+
+Prioritize user safety in all recommendations.
+Recommend alternatives or explicitly advise against outings if necessary.
+
+"""
+
+SKI_TOURING_ASSISTANT_PROMPT = """
+
+ROLE
+You are an expert assistant specializing in ski touring. Your primary task is to assist users in {language} by providing tailored recommendations for ski touring itineraries, focusing on safety, grading systems, and user preferences.
+
+TASK REQUIREMENTS
+
+Use APIs and tools to gather data on:
+Ski touring itineraries
+Weather forecasts
+Avalanche risks and conditions
+Mountain hut access
+Analyze the data and deliver user-friendly, detailed recommendations.
+Always interogate yourself about the routes access, snow and weather conditions before suggesting them to the users. `describe_route` tool will be useful for that. It's the most important part of your job.
+Answer general queries unrelated to ski touring to the best of your ability.
+
+GRADING SYSTEMS
+
+Ski Difficulty (1–5): Define skill levels based on slope gradient, exposure, and terrain.
+Ascent Grading (R, F, PD, AD, D): Classify ascents based on technical challenges.
+Exposure Grading (E1–E4): Highlight the severity of risks, from minor obstacles to life-threatening falls.
+RESPONSE FORMAT
+
+json
+Copy code
+{{
+  "message": "string (a detailed, user-friendly response in {language})",
+  "itineraries": [
+    {{
+      "topo_id": "string (ID of the itinerary from the API)",
+      "name": "string (name of the itinerary)",
+      "topo_start_lat": "float (latitude of itinerary start)",
+      "topo_start_lon": "float (longitude of itinerary start)",
+      "topo_link": "string (link to the itinerary from the API)"
+    }}
+  ])
+}}
+Use the message field to explain recommendations and safety precautions clearly.
+If no itineraries match or the query isn’t itinerary-related, set itineraries to None.
+GUIDELINES
+
+Always prioritize user safety. Warn users of unsafe conditions and suggest alternatives.
+Respond only in {language}.
+Structure messages to include relevant information such as:
+Difficulty, ascent, and exposure grading.
+Weather and avalanche conditions.
+Recommendations for preparation and safety.
+
+EXAMPLES
+
+{{"question": Que proposes tu comme itinéraire de ski de randonnée vers Briançon?}}
+{{'message': "Je vous recommande l'itinéraire 'Col de l'Aiguillette, Aller Retour depuis le pont
+de l'Alpe', situé dans le massif des Cerces - Thabor - Mont Cenis. Ce parcours est d'une difficulté de ski de 2.2 
+et d'une exposition de 1, adapté pour les skieurs ayant un certain niveau de maîtrise. L'ascension est classée R, 
+facilement accessible, avec un gain d'élévation de 850 mètres.\n\n**Conditions Avalanches:**\nLes conditions 
+actuelles montrent un risque marqué d'avalanches en raison de nombreuses plaques friables et plus dures. Les 
+avalanches spontanées ainsi que celles déclenchées par les skieurs sont anticipées, un risque à considérer 
+sérieusement avant de poursuivre cet itinéraire.\n\n**Prévisions Météo:**\nLe 8 janvier, un léger enneigement est 
+prévu, avant de se transformer en pluie dans la matinée. Les températures commencent à refroidir et l'humidité est 
+élevée. Il est recommandé de bien se préparer pour ces conditions durant vos activités outdoor.\n\nPour votre 
+sécurité, évaluez toujours les conditions météo et avalanches directement sur le terrain avant de vous lancer dans 
+l'itinéraire. Bonne randonnée et restez prudent.", 'itineraries': [{{'topo_id': '104', 'name':Col de l'Aiguillette, Aller Retour depuis le pont
+de l'Alpe ,'topo_link': 
+'https://skitour.fr/topos/104', 'topo_start_lat': 45.0181, 'topo_start_lon': 6.4663}}]}}
+
+{{"question": Quelles sont les conditions météo pour le massif du Mont-Blanc ?}}
+{{'message': "Les conditions météorologiques pour le massif du Mont-Blanc du 7 au 8 janvier 2025
+indiquent des températures froides avec des vents de vitesse variable et un ciel couvert, accompagnés de périodes 
+de chutes de neige. Points clés :\n\n- **Températures** : Commençant à -19,3°C ressentis -30,5°C le 7 janvier à 
+16h, elles augmentent graduellement la nuit à environ -7°C à -8,6°C ressentis -11,4°C à -14,9°C le lendemain 
+après-midi.\n\n- **Vent** : Principalement de l'ouest-nord-ouest (260° à 300°) avec des vitesses de 6 à 10 km/h et 
+rafales jusqu'à 22 km/h, plus fortes pendant les chutes de neige.\n\n- **Conditions Météo** : Alternance de petites
+éclaircies et ciel partiellement nuageux (30% de couverture), se chargeant fortement (90%) à partir de 20h avec des
+épisodes de neige durant la nuit et au petit matin suivant.\n\n- **Précipitations** : Pas de pluie ; début de 
+légères chutes de neige à 21h avec accumulation augmentant pour atteindre jusqu'à 4,7 mm/heure durant la nuit du 8 
+janvier. La neige s'atténue ensuite.\n\n- **Humidité** : De 45% le soir à 100% la nuit, restant élevée durant les 
+chutes de neige.\n\n- **Pression au niveau de la mer** : Fluctuations légères, autour de 1010-1015 hPa, indiquant 
+des conditions stables avec un abaissement de pression la nuit.\n\n- **Limite Pluie-Neige et Niveau Iso:0** : Non 
+pertinent initialement, mais augmentant à 1550m à 2550m, suggérant des masses d'air chaud en 
+altitude.\n\n\nConcernant les avalanches, le bulletin du 8 janvier 2025 met en garde contre de nombreuses plaques 
+sensibles dans la neige fraîche, avec un risque marqué. Des avalanches peuvent se produire spontanément jusqu'à la 
+mi-journée au-dessus de 1800 m, et peuvent être facilement déclenchées par les skieurs au-dessus de 2000 m. Les 
+plaques vont de petite à moyenne taille (Taille 1 à 2), bien que des plaques plus grandes puissent se former 
+au-dessus de 2600/3000 m.\n\nPour votre sécurité, tenez compte des prévisions de précipitations ventées dès 
+mercredi soir. La neige est lourde et mouillée sous 1600 m, affectée par les pluies. L'enneigement reste moyen pour
+la saison au-dessus de 2000 m.\n\nPour des mises à jour de vigilance, veuillez consulter les sites dédiés.", 
+'itineraries': None}}
+
+{{"question": Hello how are you?}}
+{{'message': "I'm doing great and you, how can I help you today?", 'itineraries': None}}
+
+"""

src/skitour_api.py

@@ -0,0 +1,140 @@
+import requests
+import json
+import os
+import datetime
+from typing import List, Dict
+
+SKITOUR_API_URL = 'https://skitour.fr/api/'
+
+def get_massifs() -> List[Dict]:
+    """
+    Fetch the list of massifs from the Skitour API.
+
+    Returns:
+        List[Dict]: List of massifs with their details.
+    """
+    url = SKITOUR_API_URL + 'massifs'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    response = requests.get(url, headers=headers, timeout=10)
+    return response.json()
+
+def get_topos(ids_massif: str) -> List[Dict]:
+    """
+    Fetch ski touring itineraries for a given massif.
+
+    Args:
+        ids_massif (str): ID of the massif.
+
+    Returns:
+        List[Dict]: List of itineraries for the specified massif.
+    """
+    url = SKITOUR_API_URL + 'topos'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    params = {'m': ids_massif}
+    response = requests.get(url, headers=headers, params=params, timeout=10) 
+    return json.loads(response.text.replace('\\\\', '\\'))
+
+def get_sommets(massif_id: str) -> List[Dict]:
+    """
+    Fetch the list of summits for a given massif.
+
+    Args:
+        massif_id (str): ID of the massif.
+
+    Returns:
+        List[Dict]: List of summits with their details.
+    """
+    url = SKITOUR_API_URL + 'sommets'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    params = {'m': massif_id}
+    response = requests.get(url, headers=headers, params=params)
+    response = response.json()
+    sommets = []
+    for _sommets in response:
+        sommets.append({
+            "name": _sommets['sommet'],
+            "lat": float(_sommets['latlon'][0]),
+            "lon": float(_sommets['latlon'][1]),
+            "range": _sommets['massif']['nom']
+        })
+    return sommets
+
+def get_refuges(massif_ids: str) -> List[Dict]:
+    """
+    Fetch the list of refuges for a given massif.
+
+    Args:
+        massif_ids (str): ID(s) of the massif(s).
+
+    Returns:
+        List[Dict]: List of refuges.
+    """
+    url = SKITOUR_API_URL + 'refuges'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    params = {'m': massif_ids}
+    response = requests.get(url, headers=headers, params=params, timeout=10)
+    return response.json()
+
+def get_details_topo(id_topo):
+    url = SKITOUR_API_URL + f'topo/{id_topo}'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    response = requests.get(url, headers=headers)
+    return response.json()
+
+def get_conditions(massif_ids: str) -> List[Dict]:
+    """
+    Fetch the list of refuges for a given massif.
+
+    Args:
+        massif_ids (str): ID(s) of the massif(s).
+
+    Returns:
+        List[Dict]: List of refuges.
+    """
+    url = SKITOUR_API_URL + 'refuges'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    params = {'m': massif_ids}
+    response = requests.get(url, headers=headers, params=params, timeout=10)
+    return response.json()
+
+def get_outing(id_outing: str) -> Dict:
+    """
+    Fetch the details of a specific outing.
+
+    Args:
+        id_outing (str): ID of the outing.
+
+    Returns:
+        Dict: Details of the outing.
+    """
+    url = SKITOUR_API_URL + f'sortie/{id_outing}'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    response = requests.get(url, headers=headers, timeout=10)
+    return response.json()
+
+def get_recent_outings(massif_id: str) -> List[Dict]:
+    """
+    Fetch the list of recent outings for a given massif.
+
+    Args:
+        massif_id (str): ID of the massif.
+
+    Returns:
+        List[Dict]: List of recent outings.
+    """
+    url = SKITOUR_API_URL + 'sorties'
+    headers = {'cle': os.getenv('SKITOUR_API_TOKEN')}
+    params = {'m': massif_id, 'j':30}
+    response = requests.get(url, headers=headers, params=params, timeout=10)
+    response = response.json()
+    if response:
+    
+        for _response in response:
+            _response['date'] = datetime.datetime.fromtimestamp(float(_response['date'])).strftime('%Y-%m-%d')
+            _response['description'] = get_outing(_response['id'])
+        return response
+    else: 
+        return []
+    
+
+   

src/tools.py

@@ -0,0 +1,206 @@
+import os
+import pandas as pd
+from smolagents import Tool
+from typing import List, Dict, Any, Union, Tuple
+from meteofrance_api import MeteoFranceClient
+from src.skitour_api import get_topos, get_refuges, get_details_topo, get_massifs, get_recent_outings
+from src.meteo_france_api import get_massif_conditions
+from src.utils import geocode_location, assign_location_to_clusters, haversine, llm_summarizer
+
+
+
+class RefugeTool(Tool):
+    name = "refuge_recherche"
+    description = "Recherche d'un refuge dans un massif donné"
+
+    inputs = {
+        "massif_id": {
+            "description": "[Optional, default: None] Id du massif souhaité ",
+            "type": "string",
+        }
+    }
+    output_type = "string"
+
+    def forward(self, massif_id) -> List[Dict]:
+        return get_refuges(massif_id)
+    
+    
+class GetRoutesTool(Tool):
+    name = "list_routes"
+    description = """
+    Looking for a list of ski touring routes in a given list of mountain ranges.
+    Returns a list containing the information of the topos found.
+    Use `topo_details` immediately after this tool to get the details of a specific topo. 
+    """
+
+    inputs = {
+        "mountain_range_ids": {
+            "description": "List of mountain range ids",
+            "type": "string",
+        }
+    }
+    output_type = "any"
+
+    def forward(self, mountain_range_ids: str) -> List[Dict]:
+
+        topos = get_topos(mountain_range_ids)
+        
+        return topos
+        
+class DescribeRouteTool(Tool):
+    name = "describe_route"
+    description = """ 
+    Searches for key information about a specific ski touring route, including weather forecasts and associated avalanche risks. 
+    Always use this tool after using the `list_routes` tool.
+    This tool returns a dictionary containing the route's information, the avalanche risk estimation bulletin, and the weather forecast for the coming days of the route.
+    """
+    inputs = {
+        "id_route": {
+            "description": "id of the route",
+            "type": "string",
+        },
+        "id_range": {
+            "description": "mountain range id of the route",
+            "type": "string"}
+    }
+    output_type = "any"
+    
+    def __init__(self, skitour2meteofrance: dict, llm_engine: Any):
+        super().__init__()
+        self.massifs_infos = skitour2meteofrance
+        self.weather_client =  MeteoFranceClient(access_token=os.getenv("METEO_FRANCE_API_KEY"))   
+        self.llm_engine = llm_engine
+
+    def forward(self, id_route: str, id_range: str) -> dict:
+
+        topo_info = get_details_topo(str(id_route))
+        avalanche_conditions = get_massif_conditions(
+            self.massifs_infos[str(id_range)]['meteofrance_id']
+            )
+        lat, lon = topo_info["depart"]["latlon"]
+        weather_forecast = self.weather_client.get_forecast(float(lat), float(lon))
+        daily_forecast = weather_forecast.forecast[:24]
+
+        for day_forecast in daily_forecast:
+            day_forecast["dt"] = weather_forecast.timestamp_to_locale_time(day_forecast["dt"]).isoformat()
+        forecast_summary = llm_summarizer(str(daily_forecast), self.llm_engine)
+        avalanche_summary = llm_summarizer(str(avalanche_conditions), self.llm_engine)
+        return {
+            "route_info": topo_info, 
+            "avalanche_conditions": avalanche_summary,
+            "daily_weather_forecast": forecast_summary,
+            "route_link": f"https://skitour.fr/topos/{id_route}"
+            }   
+    
+class RecentOutingsTool(Tool):
+    name = "recent_outings"
+    description = """ 
+    Searches for recent outings in a given mountain range.
+    Returns a list of the most recent outings in the given range.
+    """
+    inputs = {
+        "id_range": {
+            "description": "id of the mountain range",
+            "type": "string",
+        }
+    }
+    output_type = "any"
+    
+    def forward(self, id_range: str) -> List[Dict]:
+        return get_recent_outings(id_range)
+    
+class MountainRangesTool(Tool):
+    name = "list_mountain_ranges"
+    description = """ Searches for the ID(s) of the mountain ranges closest to a given location.
+    If the location is too far from known ranges, the search returns None.
+    Should return a string with the massif IDs separated by commas.
+    """
+
+    inputs = {
+        "location": {
+            "description": "Location to search for",
+            "type": "string",
+        },
+        
+        "num_ranges": {
+            "description": "[Optional, default: 3] Number of closest mountain ranges to return",
+            "type": "number",   
+        }
+    }
+    output_type = "string"
+    
+    def __init__(self, clusters: Dict[str, List[Tuple[float, float]]]):
+        super().__init__()
+        self.clusters = clusters
+
+    def forward(self, location: str, num_ranges: int) -> Union[str, None]:
+
+        coord_location = geocode_location(location)
+        if not location:
+            return None
+        
+        matched_ranges = assign_location_to_clusters(coord_location, self.clusters, k=num_ranges)
+
+        
+        list_ranges = [range[0] for range in matched_ranges if range[1] < 100]
+        if not list_ranges:
+            return ''
+      
+        massifs= get_massifs()
+       
+        massif_ids = [_massif['id'] for _massif in massifs if _massif['nom'] in list_ranges]
+        return ", ".join(massif_ids)
+    
+class ForecastTool(Tool):
+    name = "forecast"
+    description = """Searches for the weather forecast for a given location as well as the current avalanche risk estimation bulletin.  
+    Unnecessary if the user is inquiring about a route, as `describe_route` already provides this information."""
+    
+    inputs = {
+        "location": {
+            "description": "Location to search for",
+            "type": "string",
+        },
+    }
+    
+    output_type = "any"
+
+    def __init__(self, llm_engine, clusters: Dict[str, List[Tuple[float, float]]], skitour2meteofrance: dict):
+        super().__init__()
+        self.clusters = clusters
+        self.massifs_infos = skitour2meteofrance
+        self.llm_engine = llm_engine
+        
+    def forward(self, location: str) -> Union[Dict[str, Any], None]:
+
+        coord_location = geocode_location(location)
+        if not location:
+            return None
+        
+        # Get the closest mountain range to the location to get the avalanche conditions
+        matched_ranges = assign_location_to_clusters(coord_location, self.clusters, k=1)
+        
+        list_ranges = [range[0] for range in matched_ranges if range[1] < 100]
+        if not list_ranges:
+            return None
+      
+        massifs= get_massifs()
+       
+        massif_id = [_massif['id'] for _massif in massifs if _massif['nom'] in list_ranges]
+        
+        avalanche_conditions = get_massif_conditions(
+            self.massifs_infos[str(massif_id[0])]['meteofrance_id']
+            )
+        
+        weather_client =  MeteoFranceClient(access_token=os.getenv("METEO_FRANCE_API_KEY"))
+        forecast = weather_client.get_forecast(*coord_location)
+        daily_forecast = forecast.forecast[:24]
+        
+        for day_forecast in daily_forecast:
+            day_forecast["dt"] = forecast.timestamp_to_locale_time(day_forecast["dt"]).isoformat()
+        
+        forecast_summary = llm_summarizer(str(daily_forecast), self.llm_engine)
+        avalanche_summary = llm_summarizer(str(avalanche_conditions), self.llm_engine)
+            
+    
+        return {"forecast": forecast_summary, "avalanche_conditions": avalanche_summary}

src/utils.py

@@ -0,0 +1,135 @@
+
+import re
+import json
+import googlemaps
+import os
+from math import radians, sin, cos, sqrt, atan2
+import numpy as np
+from typing import Tuple, Dict, List
+from openai import OpenAI
+
+def geocode_location(query: str) -> Tuple[float, float]:
+    """
+    Geocode a location query into latitude and longitude.
+
+    Args:
+        query (str): Location query string.
+
+    Returns:
+        Tuple[float, float]: Latitude and longitude of the location.
+    """
+    gmaps = googlemaps.Client(key=os.getenv('GOOGLE_MAPS_API_KEY'))
+    geocode_result = gmaps.places(query)
+    try:
+        location = geocode_result['results'][0]['geometry']['location']
+        return location['lat'], location['lng']
+    except:
+        return None
+
+def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """
+    Calculate the great circle distance in kilometers between two points on Earth.
+
+    Args:
+        lat1 (float): Latitude of the first point.
+        lon1 (float): Longitude of the first point.
+        lat2 (float): Latitude of the second point.
+        lon2 (float): Longitude of the second point.
+
+    Returns:
+        float: Distance in kilometers.
+    """
+    R = 6371
+    lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
+    dlat = lat2 - lat1
+    dlon = lon2 - lon1
+    a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
+    c = 2 * atan2(sqrt(a), sqrt(1 - a))
+    return R * c
+
+def assign_location_to_clusters(location: Tuple[float, float], clusters: Dict[str, List[Tuple[float, float]]], k: int = 3) -> List[Tuple[str, float]]:
+    """
+    Assign a location to the closest clusters of mountain ranges based on proximity.
+
+    Args:
+        location (Tuple[float, float]): Latitude and longitude of the location.
+        clusters (Dict[str, List[Tuple[float, float]]]): Clusters of mountain ranges.
+        k (int): Number of closest clusters to return.
+
+    Returns:
+        List[Tuple[str, float]]: Closest clusters and their distances.
+    """
+    closest_summits = []
+    for range_label, points in clusters.items():
+        min_distance = float('inf')
+        for lat, lon in points:
+            dist = haversine(location[0], location[1], lat, lon)
+            if dist < min_distance:
+                min_distance = dist
+        closest_summits.append((range_label, min_distance))
+    closest_summits.sort(key=lambda x: x[1])
+    return closest_summits[:k]
+
+def build_clustered_mountain_ranges(peaks):
+    """
+    Group peaks into clusters based on their mountain range labels.
+    :param peaks: list of dicts [{'name': str, 'lat': float, 'lon': float, 'range': str}]
+    :return: dict of clusters {range_label: [(lat, lon)]}
+    """
+    clusters = {}
+    for peak in peaks:
+        range_label = peak['range']
+        if range_label not in clusters:
+            clusters[range_label] = []
+        clusters[range_label].append((peak['lat'], peak['lon']))
+    return clusters
+
+
+
+def parse_topo_blob(output: str):
+    """
+    Checks if there is any dictionary-like structure in the output string.
+
+    Args:
+        output (str): The output string potentially containing a dictionary.
+
+    Returns:
+        dict or None: Parsed dictionary if found, otherwise None.
+    """
+    # Define a regex pattern to match any dictionary-like structure
+    pattern = r"\{\s*(?:\"[^\"]+\":\s*(?:\".*?\"|\d+|true|false|null),?\s*)+\}"  # Matches any JSON-like dict
+
+    # Search for the pattern in the output
+    match = re.search(pattern, output, re.DOTALL)
+    if match:
+        blob = match.group()
+        try:
+            # Parse the JSON-like dict
+            return json.loads(blob)
+        except json.JSONDecodeError as e:
+            print(f"Error decoding JSON: {e}")
+            return None
+
+    return None
+
+def llm_summarizer(text, llm_engine):
+    
+   
+    messages=[
+           {
+            "role": "system",
+            "content": "You're an expert at summarizing data on weather forecast and avalanche conditions. Summarize the data that's been provided to you below"
+        }, 
+        {
+            "role": "user",
+            "content": text,
+        }
+    ]
+    
+    summary = llm_engine(messages)
+    return summary["content"]
+    
+    
+    
+    
+