app.py

import streamlit as st
import requests
import json
import random
from datetime import datetime
import os

# Initialize session state
if "messages" not in st.session_state:
    st.session_state.messages = [{"role": "assistant", "content": "Welcome to the Solar EV Charging Chatbot! Ask about stations, charging, or solar status."}]
if "sessions" not in st.session_state:
    st.sessions = {}
if "user_type" not in st.session_state:
    st.session_state.user_type = "user"

# API keys from Streamlit secrets
OPENCHARGEMAP_API_KEY = st.secrets.get("OPENCHARGEMAP_API_KEY", "")
OPENWEATHERMAP_API_KEY = st.secrets.get("OPENWEATHERMAP_API_KEY", "")

# Simulated API for fault diagnostics and session control
class SimulatedAPI:
    def check_faults(self):
        return {"faults": None, "status": "All systems operational"}

    def start_charging(self, station_id, user_id):
        session_id = random.randint(1000, 9999)
        return {"session_id": session_id, "status": "Charging started"}

    def stop_charging(self, session_id):
        return {"status": "Charging stopped"}

# Real-time API handlers
class RealTimeAPI:
    def get_stations(self, lat=51.5074, lon=-0.1278, max_results=5):
        """Fetch nearby charging stations using OpenChargeMap API."""
        url = "https://api.openchargemap.io/v3/poi/"
        params = {
            "key": OPENCHARGEMAP_API_KEY,
            "latitude": lat,
            "longitude": lon,
            "distance": 10,
            "distanceunit": "Miles",
            "maxresults": max_results,
            "compact": True,
            "verbose": False
        }
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            stations = response.json()
            return [
                {
                    "id": s["ID"],
                    "name": s["AddressInfo"]["Title"],
                    "distance": s["AddressInfo"]["Distance"],
                    "slots": {"total": s.get("NumberOfPoints", 1), "free": random.randint(0, s.get("NumberOfPoints", 1))},
                    "solar_power": self.estimate_solar_power(lat, lon)
                }
                for s in stations
            ]
        except requests.RequestException as e:
            return [{"error": f"Failed to fetch stations: {str(e)}"}]

    def estimate_solar_power(self, lat, lon):
        """Estimate solar power based on weather conditions using OpenWeatherMap."""
        url = f"http://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={OPENWEATHERMAP_API_KEY}"
        try:
            response = requests.get(url)
            response.raise_for_status()
            weather = response.json()
            cloud_cover = weather["clouds"]["all"]
            solar_efficiency = max(0.1, 1.0 - cloud_cover / 100.0)  # Simplified model
            return round(5.0 * solar_efficiency, 2)  # Assume 5 kW max capacity
        except requests.RequestException:
            return 5.0  # Fallback value

    def calculate_charging_cost(self, ev_model, target_percent, current_percent):
        """Calculate charging cost based on solar vs. grid power."""
        kWh_needed = (target_percent - current_percent) / 100 * 75  # Assume 75 kWh battery
        solar_cost_per_kWh = 0.10
        grid_cost_per_kWh = 0.20
        solar_share = 0.8
        cost = (kWh_needed * solar_cost_per_kWh * solar_share) + (kWh_needed * grid_cost_per_kWh * (1 - solar_share))
        return round(cost, 2), round(kWh_needed / 50 * 60, 2)  # Assume 50 kW charging speed

# Chatbot logic
class EVSolarChatbot:
    def __init__(self):
        self.real_time_api = RealTimeAPI()
        self.simulated_api = SimulatedAPI()

    def process_message(self, message):
        message = message.lower().strip()
        user_type = st.session_state.user_type

        if "find" in message and "station" in message:
            stations = self.real_time_api.get_stations()
            if "error" in stations[0]:
                return stations[0]["error"]
            response = f"Nearest station: {stations[0]['name']} ({stations[0]['distance']:.1f} miles). "
            response += f"Slots: {stations[0]['slots']['free']}/{stations[0]['slots']['total']} free. "
            response += f"Solar power: {stations[0]['solar_power']} kW. Reserve a slot?"
            return response

        elif "cost to charge" in message:
            cost, time = self.real_time_api.calculate_charging_cost("Tesla Model 3", 80, 20)
            response = f"Charging to 80% will cost ~${cost} (vs. ${cost * 1.5:.2f} on grid). "
            response += f"Time: ~{time} mins. Proceed?"
            return response

        elif "start charging" in message:
            session = self.simulated_api.start_charging(station_id=1, user_id="user123")
            st.sessions[session["session_id"]] = {"status": "active", "start_time": datetime.now()}
            return f"Charging started. Session ID: {session['session_id']}."

        elif "stop charging" in message:
            session_id = message.split("session id")[-1].strip() if "session id" in message else "1000"
            if int(session_id) in st.sessions:
                self.simulated_api.stop_charging(session_id)
                st.sessions[int(session_id)]["status"] = "stopped"
                return f"Charging stopped for Session ID: {session_id}."
            return "Invalid session ID."

        elif user_type == "operator" and "solar status" in message:
            stations = self.real_time_api.get_stations()
            if "error" in stations[0]:
                return stations[0]["error"]
            solar_power = stations[0]["solar_power"]
            response = f"Solar generation: {solar_power} kW. Battery: 65%. Grid dependency: Low."
            return response

        elif user_type == "operator" and "check faults" in message:
            faults = self.simulated_api.check_faults()
            return f"System status: {faults['status']}. Faults: {faults['faults'] or 'None'}."

        else:
            return "Sorry, I didn't understand. Try asking about stations, charging costs, or solar status."

# Streamlit app
def main():
    st.set_page_config(page_title="Solar EV Charging Chatbot", layout="wide")
    st.title("🤖 Solar EV Charging Chatbot")
    
    # Sidebar for user type and info
    with st.sidebar:
        st.header("Settings")
        st.session_state.user_type = st.selectbox("User Type", ["user", "operator"])
        st.markdown("""
        ## About
        This chatbot helps EV owners and station operators:
        - Find solar-powered charging stations
        - Manage charging sessions
        - Monitor solar energy and faults
        Powered by OpenChargeMap and OpenWeatherMap APIs.
        """)

    # Chat interface
    chatbot = EVSolarChatbot()
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.write(message["content"])

    # User input
    if prompt := st.chat_input("Type your message here..."):
        # Display user message
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.write(prompt)

        # Generate and display assistant response
        with st.chat_message("assistant"):
            with st.spinner("Thinking..."):
                response = chatbot.process_message(prompt)
                st.write(response)
                st.session_state.messages.append({"role": "assistant", "content": response})

if __name__ == "__main__":
    main()