Create app.py

app.py

@@ -0,0 +1,209 @@
+import gradio as gr
+import joblib
+import pandas as pd
+import numpy as np
+import re
+import requests
+from datetime import datetime
+from geopy.geocoders import Nominatim
+from geopy.distance import geodesic
+
+# Cargar modelo y dataset
+model = joblib.load("modelo_docks.pkl")
+df_stations = pd.read_csv("Informacio_Estacions_Bicing_2025.csv")
+geolocator = Nominatim(user_agent="bicing-agent")
+
+# LLM: llama-3.3-70b-versatile
+def preguntar_al_usuario(pregunta):
+    response = client.chat.completions.create(
+        messages=[
+            {"role": "system", "content": "Eres un asistente de Bicing. Tu tarea es hacer una preguntal usuario y esperar su respuesta. No saludes a menos que te lo pida"},
+            {"role": "user", "content": f"Pregunta al usuario lo siguiente, puedes modificar el tono para hacerlo mas amigable y añadir ayudas adicionales sobre como introducir los datos: '{pregunta}'"}
+        ],
+        model="llama-3.3-70b-versatile",
+        temperature=0.5,
+        max_completion_tokens=512,
+        top_p=1,
+        stream=False,
+    )
+    return response.choices[0].message.content.strip()
+
+# Estaciones más cercanas
+def get_nearest_stations(ubicacion, top_n=5):
+    loc = geolocator.geocode(f"{ubicacion}, Barcelona, Spain")
+    if not loc:
+        return pd.DataFrame()
+
+    user_coord = (loc.latitude, loc.longitude)
+    df_stations["distancia"] = df_stations.apply(
+        lambda row: geodesic(user_coord, (row["lat"], row["lon"])).meters,
+        axis=1
+    )
+    return df_stations.nsmallest(top_n, "distancia")[["station_id", "address", "lat", "lon"]]
+
+# Tiempo (Open-Meteo)
+def get_weather_forecast(lat, lon, year, month, day, hour):
+    fecha = f"{year}-{month:02d}-{day:02d}"
+    hora_str = f"{hour:02d}:00"
+
+    url = (
+        f"https://api.open-meteo.com/v1/forecast?"
+        f"latitude={lat}&longitude={lon}&hourly=temperature_2m,precipitation&timezone=Europe%2FMadrid"
+        f"&start_date={fecha}&end_date={fecha}"
+    )
+
+    r = requests.get(url)
+    if r.status_code != 200:
+        return None, None
+
+    data = r.json()
+    horas = data["hourly"]["time"]
+    temperaturas = data["hourly"]["temperature_2m"]
+    precipitaciones = data["hourly"]["precipitation"]
+
+    for i, h in enumerate(horas):
+        if h.endswith(hora_str):
+            return temperaturas[i], precipitaciones[i]
+
+    return None, None
+
+# Predicción con el modelo
+def predict_disponibilidad(context):
+    estaciones_cercanas = get_nearest_stations(context["ubicacion"])
+    if estaciones_cercanas.empty:
+        return {"error": "No se encontraron estaciones cercanas."}
+
+    resultados = []
+
+    for _, row in estaciones_cercanas.iterrows():
+        temp, precip = get_weather_forecast(
+            row["lat"], row["lon"], 2025,
+            context["month"], context["day"], context["hour"]
+        )
+        if temp is None:
+            continue
+
+        X = np.array([[
+            row["station_id"],
+            context["month"],
+            context["day"],
+            context["hour"],
+            0, 0, 0, 0,  # ctx históricos por defecto
+            temp,
+            1.0 if precip > 0 else 0.0
+        ]])
+        pred = model.predict(X)[0]
+
+        resultados.append({
+            "station_id": row["station_id"],
+            "address": row["address"],
+            "pred_pct": float(pred),
+            "temperature": round(temp, 1),
+            "precip": round(precip, 1)
+        })
+
+    if not resultados:
+        return {"error": "No se pudieron calcular predicciones meteorológicas."}
+
+    resultados_ordenados = sorted(resultados, key=lambda x: x["pred_pct"], reverse=True)
+    return {
+        "target_pct": context["target_pct"],
+        "candidatas": resultados_ordenados
+    }
+
+# Preguntas al usuario
+preguntas = [
+    ("ubicacion", "INTRODUCE SALUDO y la pregunta ¿Dónde te gustaría coger la bici? (zona o dirección en Barcelona)"),
+    ("month", "¿En qué mes planeas cogerla? (número 1-12)"),
+    ("day", "¿Qué día del mes?"),
+    ("hour", "¿A qué hora la necesitas? (0-23)?"),
+    ("target_pct", "¿Qué porcentaje mínimo de bicicletas esperas encontrar disponibles? (0 a 100%)")
+]
+
+# Flujo de conversación
+def chat(user_input, chat_history, current_step, user_context):
+    key, _ = preguntas[current_step]
+
+    if key in ["month", "day", "hour", "target_pct"]:
+        match = re.search(r"\d+(\.\d+)?", user_input)
+        if match:
+            value = float(match.group())
+            user_context[key] = value / 100 if key == "target_pct" else int(value)
+        else:
+            chat_history.append(("user", user_input))
+            chat_history.append(("assistant", "Introduce un número válido."))
+            return chat_history, current_step, user_context
+    else:
+        user_context[key] = user_input.strip()
+
+    chat_history.append(("user", user_input))
+    current_step += 1
+
+    if current_step < len(preguntas):
+        siguiente_pregunta = preguntar_al_usuario(preguntas[current_step][1])
+        chat_history.append(("assistant", siguiente_pregunta))
+    else:
+        resultado = predict_disponibilidad(user_context)
+        if "error" in resultado:
+          chat_history.append(("assistant", resultado["error"] + " Reiniciando conversación..."))
+          # Reiniciar el estado
+          user_context = {
+              "ubicacion": None,
+              "month": None,
+              "day": None,
+              "hour": None,
+              "target_pct": None,
+              "temperature": None,
+              "lluvia": None
+          }
+          current_step = 0
+          chat_history.append(("assistant", preguntar_al_usuario(preguntas[0][1])))
+          return chat_history, current_step, user_context
+
+        else:
+            clima = resultado["candidatas"][0]
+            msg = (
+                f"📅 Predicción meteorológica para la hora solicitada:\n"
+                f"🌡️ Temperatura aprox.: {clima['temperature']}°C\n"
+                f"☔ Precipitación aprox.: {clima['precip']} mm\n\n"
+                f"🚲 Estado de las estaciones de bicing en ese momento {round(resultado['target_pct']*100)}% de bicis:\n\n"
+            )
+            for r in resultado["candidatas"]:
+                emoji = "✅" if r["pred_pct"] >= resultado["target_pct"] else "⚠️"
+                msg += (
+                    f"{emoji} Estación '{r['address']}' (ID {r['station_id']}): "
+                    f"{round(r['pred_pct']*100)}% disponibilidad\n"
+                )
+            chat_history.append(("assistant", msg.strip()))
+
+    return chat_history, current_step, user_context
+
+# Interfaz Gradio
+with gr.Blocks() as demo:
+    chatbot = gr.Chatbot()
+    txt = gr.Textbox(placeholder="Escribe tu respuesta...", label="Tu mensaje")
+
+    state_chat = gr.State([])
+    state_step = gr.State(0)
+    state_context = gr.State({
+        "ubicacion": None,
+        "month": None,
+        "day": None,
+        "hour": None,
+        "target_pct": None,
+        "temperature": None,
+        "lluvia": None
+    })
+
+    def user_submit(message, chat_history, current_step, user_context):
+        return chat(message, chat_history, current_step, user_context)
+
+    txt.submit(user_submit, inputs=[txt, state_chat, state_step, state_context],
+               outputs=[chatbot, state_step, state_context])
+
+    # Primer mensaje
+    primer_pregunta = preguntar_al_usuario(preguntas[0][1])
+    state_chat.value = [("assistant", primer_pregunta)]
+    chatbot.value = state_chat.value
+
+demo.launch()