Мобильная версияНастройка включает в себя:
API Google Gemini: используется для генерации ответов для чат-бота. .
Приложение Streamlit: в этом приложении есть панель управления для отображения данных датчиков и страница чат-бота.
Firebase: хранит данные датчиков в реальном времени (например, температуру, влажность).
Код: Выделить всё
import streamlit as st
import google.generativeai as genai
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, db
import os
import pandas as pd
from datetime import datetime, timedelta
load_dotenv()
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
if not firebase_admin._apps:
cred = credentials.Certificate("path/to/firebase-adminsdk.json")
firebase_admin.initialize_app(cred, {
'databaseURL': 'https://.firebaseio.com/'
})
sensor_data_ref = db.reference('sensorData')
def get_sensor_data():
sensor_data = sensor_data_ref.get()
if sensor_data:
data_list = [
{**values, 'id': key} for key, values in sensor_data.items() if isinstance(values, dict)
]
df = pd.DataFrame(data_list)
df['timestamp'] = pd.to_datetime(df['timestamp'])
return df
return pd.DataFrame()
def is_project_related(prompt):
project_keywords = [
"temperature", "pressure", "humidity", "current", "average", "today",
"sensor", "dashboard", "maintenance", "real-time"
]
return any(keyword in prompt.lower() for keyword in project_keywords)
def generate_response(prompt):
if not is_project_related(prompt):
return "I'm sorry, but I can only answer questions related to the Predictive Maintenance System project."
df = get_sensor_data()
if df.empty:
return "I'm unable to retrieve data from the sensors right now. Please check if the data is being pushed correctly to Firebase."
if "current temperature" in prompt.lower() and not df.empty:
latest_temp = df.sort_values(by='timestamp', ascending=False).iloc[0]['temperature']
return f"The current temperature is {latest_temp}°C."
elif "average temperature" in prompt.lower() and not df.empty:
last_day = datetime.now() - timedelta(days=1)
recent_data = df[df['timestamp'] > last_day]
if not recent_data.empty:
avg_temp = recent_data['temperature'].mean()
return f"The average temperature over the last 24 hours is {avg_temp:.2f}°C."
else:
return "No data available for the last 24 hours."
context = """
You are an AI assistant for a Predictive Maintenance System project. The system uses sensors
(temperature, pressure, humidity, and current) connected to an ESP32 microcontroller to monitor
industrial equipment. It collects real-time data, uses machine learning for failure prediction,
and provides a dashboard for monitoring. Only respond to queries related to this system.
"""
full_prompt = f"{context}\n\nUser: {prompt}\nAssistant:"
model = genai.GenerativeModel('gemini-pro')
response = model.generate_content(full_prompt)
return response.text
def main():
st.set_page_config(page_title="Predictive Maintenance System", layout="wide")
st.markdown("""
.navbar { display: flex; justify-content: space-evenly; font-size: 1.2em; }
.navbar-item { cursor: pointer; padding: 0.5em 1em; color: #007ACC; text-decoration: none; }
.navbar-item:hover { background-color: #f1f1f1; border-radius: 5px; }
.navbar-container { display: flex; justify-content: center; padding: 1em 0; background-color: #F0F2F6; }
""", unsafe_allow_html=True)
navbar_items = st.columns(2)
with navbar_items[0]:
if st.button("Dashboard"):
st.session_state.page = "Dashboard"
with navbar_items[1]:
if st.button("Chatbot"):
st.session_state.page = "Chatbot"
if "page" not in st.session_state:
st.session_state.page = "Dashboard"
if st.session_state.page == "Dashboard":
st.header("Real-Time Sensor Dashboard")
df = get_sensor_data()
if not df.empty:
st.write("### Latest Sensor Readings")
latest_data = df.sort_values(by='timestamp', ascending=False).iloc[0]
st.write(latest_data)
st.write("### Sensor Data Over Time")
st.line_chart(df.set_index("timestamp")[["temperature", "pressure", "humidity", "current"]])
else:
st.write("No sensor data available.")
elif st.session_state.page == "Chatbot":
st.header("Maintenance Assistant Chatbot")
if "messages" not in st.session_state:
st.session_state.messages = []
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
if prompt := st.chat_input("Ask about the Predictive Maintenance System"):
st.chat_message("user").markdown(prompt)
st.session_state.messages.append({"role": "user", "content": prompt})
response = generate_response(prompt)
with st.chat_message("assistant"):
st.markdown(response)
st.session_state.messages.append({"role": "assistant", "content": response})
if __name__ == "__main__":
main()
Получение данных Firebase: функция get_sensor_data успешно извлекает данные в реальном времени из Firebase на странице панели управления, но чат-бот не может напрямую получить доступ к данным. По умолчанию используются ответы Gemini или возвращается сообщение о том, что данные недоступны.
Логика ответа чат-бота: для вопросов типа «Какова текущая температура?» чат-бот не использует Данные Firebase соответствуют ожиданиям. Мы хотели, чтобы он извлекал последние добавленные данные из скрипта Python (в последней демонстрации мы будем использовать реальную настройку IOT) и считал их текущим значением этой конкретной системы. Мы попробовали несколько условий в Generate_response, чтобы проверить, присутствуют ли в подсказке такие ключевые слова, как «текущая температура» или «средняя температура», но это все равно не работает должным образом.
Подробнее здесь: https://stackoverflow.com/questions/791 ... base-and-r
