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()