"""
Streamlit Frontend - PM Data Chatbot
"""
import streamlit as st
import requests
import pandas as pd
import plotly.express as px
from datetime import datetime
from config import API_BASE_URL
# Page configuration
st.set_page_config(
page_title="PM Data Chatbot",
page_icon="📊",
layout="wide"
)
# Initialize session state
if "messages" not in st.session_state:
st.session_state.messages = []
if "query_history" not in st.session_state:
st.session_state.query_history = []
def query_api(query: str) -> dict:
"""Query the API with natural language"""
try:
response = requests.get(f"{API_BASE_URL}/api/query", params={"q": query}, timeout=30)
response.raise_for_status()
return response.json()
except Exception as e:
return {"success": False, "error": str(e), "results": []}
def get_interfaces() -> list:
"""Get list of interfaces"""
try:
response = requests.get(f"{API_BASE_URL}/api/interfaces", timeout=10)
response.raise_for_status()
data = response.json()
return data.get("interfaces", [])
except:
return []
def get_counters() -> dict:
"""Get list of counters"""
try:
response = requests.get(f"{API_BASE_URL}/api/counters", timeout=10)
response.raise_for_status()
data = response.json()
return data.get("counters", {})
except:
return {}
def get_fetch_interval() -> int:
"""Get current fetch interval"""
try:
response = requests.get(f"{API_BASE_URL}/api/config/fetch-interval", timeout=10)
response.raise_for_status()
data = response.json()
return data.get("interval_minutes", 5)
except:
return 5
def update_fetch_interval(minutes: int) -> bool:
"""Update fetch interval"""
try:
response = requests.post(
f"{API_BASE_URL}/api/config/fetch-interval",
json={"minutes": minutes},
timeout=10
)
response.raise_for_status()
return True
except:
return False
def display_results(results: list):
"""Display query results in a formatted way"""
if not results:
st.info("No results found")
return
# Convert to DataFrame for better display
df_data = []
for result in results:
row = {}
for key, value in result.items():
if isinstance(value, (int, float)):
row[key] = value
elif isinstance(value, datetime):
row[key] = value.isoformat()
else:
row[key] = str(value)
df_data.append(row)
df = pd.DataFrame(df_data)
# Display table
st.dataframe(df, use_container_width=True)
# If there are numeric values, show chart
numeric_cols = df.select_dtypes(include=['float64', 'int64']).columns.tolist()
if 'value' in numeric_cols and len(df) > 1:
try:
# Group by interface if available
if 'interface_name' in df.columns:
fig = px.bar(
df,
x='interface_name',
y='value',
title="Counter Values by Interface",
labels={'value': 'Value', 'interface_name': 'Interface'}
)
else:
fig = px.bar(
df,
y='value',
title="Counter Values",
labels={'value': 'Value'}
)
st.plotly_chart(fig, use_container_width=True)
except Exception as e:
st.caption(f"Could not generate chart: {e}")
# Sidebar for configuration
with st.sidebar:
st.header("⚙️ Configuration")
# Fetch interval configuration
st.subheader("Fetch Interval")
current_interval = get_fetch_interval()
new_interval = st.number_input(
"Fetch Interval (minutes)",
min_value=1,
max_value=1440,
value=current_interval,
step=1
)
if st.button("Update Interval"):
if update_fetch_interval(new_interval):
st.success(f"Fetch interval updated to {new_interval} minutes")
else:
st.error("Failed to update fetch interval")
st.divider()
# Available interfaces
st.subheader("Available Interfaces")
interfaces = get_interfaces()
if interfaces:
for iface in interfaces[:10]: # Show first 10
st.text(f"• {iface.get('name', 'N/A')}")
if len(interfaces) > 10:
st.caption(f"... and {len(interfaces) - 10} more")
else:
st.caption("No interfaces found")
st.divider()
# Available counters
st.subheader("Available Counters")
counters = get_counters()
if counters:
for category, counter_list in counters.items():
if counter_list:
with st.expander(f"{category.title()} ({len(counter_list)})"):
for counter in counter_list[:5]:
st.text(f"• {counter}")
if len(counter_list) > 5:
st.caption(f"... and {len(counter_list) - 5} more")
else:
st.caption("No counters found")
# Main chat interface
st.title("📊 PM Data Chatbot")
st.caption("Ask questions about performance monitoring data in natural language")
# Display chat history
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
if "results" in message:
display_results(message["results"])
# Chat input
if prompt := st.chat_input("Ask a question about PM data..."):
# Add user message to chat
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# Query API
with st.chat_message("assistant"):
with st.spinner("Thinking..."):
result = query_api(prompt)
if result.get("success") and result.get("results"):
st.markdown("Here are the results:")
display_results(result["results"])
# Add to session state
st.session_state.messages.append({
"role": "assistant",
"content": "Query executed successfully",
"results": result["results"]
})
st.session_state.query_history.append({
"query": prompt,
"timestamp": datetime.now().isoformat(),
"results_count": len(result["results"])
})
else:
error_msg = result.get("error", "No results found")
# Show error with helpful message
st.warning(f"⚠️ {error_msg}")
# Also show parsed query info if available
if "parsed" in result:
with st.expander("Query Details"):
st.json(result["parsed"])
st.session_state.messages.append({
"role": "assistant",
"content": error_msg
})
# Query history section
with st.expander("📜 Query History"):
if st.session_state.query_history:
history_df = pd.DataFrame(st.session_state.query_history)
st.dataframe(history_df, use_container_width=True)
if st.button("Clear History"):
st.session_state.query_history = []
st.rerun()
else:
st.caption("No query history yet")
