Spaces:
Sleeping
Sleeping
| 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() |