Spaces:
Runtime error
Runtime error
"""Module to launch the application. | |
run: `gradio main.py` to run the application | |
""" | |
import sys | |
import gradio as gr | |
from loguru import logger | |
from app.service import ( | |
analyze_trends, | |
connect_database, | |
detect_anomalies, | |
generate_analysis_report, | |
list_available_metrics, | |
query_timeseries, | |
) | |
logger.add( | |
sys.stdout, | |
format="<green>{time}</green> <level>{message}</level>", | |
filter="my_module", | |
level="INFO", | |
colorize=True, | |
) | |
example_sensor = "temperature" | |
example_start = "2019-06-15T02:54:00" | |
example_end = "2019-06-17T02:54:00" | |
with gr.Blocks() as demo: | |
gr.Markdown("# TimescaleDB Time Series Analyzer API (Gradio)") | |
with gr.Tab("About"): | |
gr.Markdown(""" | |
# mcp-tscontext | |
**mcp-tscontext** is a time series analysis tool designed for use with TimescaleDB. | |
Its goal is to provide a textual context about time series. | |
The idea is to provide a tool to an Agent to make it possible to better | |
understand the context when a user asks a question. | |
Built with Gradio, provides a mcp server. | |
My main use Case: | |
- provide an assistant to operators in charge of supervising industrial systems | |
(like a power plant). Combine this tool with RAG on technical documentation. | |
Other use case ideas: | |
- Build a personal assistant that could be aware of health metrics provided by a | |
smartwatch. | |
## Features | |
- **Database Connection**: Connect to a TimescaleDB instance or use a mock SQLite database for testing. | |
- **List Metrics**: View all available sensor ids in the database. | |
- **Query Time Series**: Retrieve time series for a specific sensor and time range. | |
- **Anomaly Detection**: Identify anomalies in sensor data using statistical methods. | |
- **Trend Analysis**: Analyze trends and changes in sensor data over time. | |
- **Report Generation**: Generate a report with trends, anomalies, and statistics. | |
- **User-Friendly UI**: Interact with all features through a modern Gradio web interface. | |
## Usage | |
By default, the app uses a mock SQLite database (`mock.db`). | |
To connect to a real TimescaleDB instance, set the following environment variables: | |
- USE_MOCK_DB (set to false) | |
- `DB_HOST` | |
- `DB_PORT` | |
- `DB_NAME` | |
- `DB_USER` | |
- `DB_PASS` | |
""") | |
with gr.Tab("Connect DB"): | |
connect_btn = gr.Button("Connect to TimescaleDB") | |
connect_out = gr.Textbox(label="Connection Result") | |
connect_btn.click( | |
fn=connect_database, inputs=[], outputs=connect_out, show_api=False | |
) | |
with gr.Tab("List Metrics"): | |
list_btn = gr.Button("List Available Metrics") | |
list_out = gr.Textbox(label="Metrics") | |
list_btn.click( | |
fn=list_available_metrics, | |
inputs=[], | |
outputs=list_out, | |
) | |
with gr.Tab("Query Timeseries"): | |
sensor_id = gr.Textbox(label="Sensor ID", value=example_sensor) | |
start_time = gr.Textbox(label="Start Time (ISO)", value=example_start) | |
end_time = gr.Textbox(label="End Time (ISO)", value=example_end) | |
query_btn = gr.Button("Query") | |
query_out = gr.Textbox(label="Query Result") | |
query_btn.click( | |
fn=query_timeseries, | |
inputs=[sensor_id, start_time, end_time], | |
outputs=query_out, | |
) | |
with gr.Tab("Detect Anomalies"): | |
sensor_id2 = gr.Textbox(label="Sensor ID", value=example_sensor) | |
start_time2 = gr.Textbox(label="Start Time (ISO)", value=example_start) | |
end_time2 = gr.Textbox(label="End Time (ISO)", value=example_end) | |
algorithm = gr.Radio( | |
label="Algorithm", | |
choices=["zscore", "isolation_forest"], | |
value="zscore", | |
) | |
threshold = gr.Number(label="Z-Score Threshold", value=2.0) | |
contamination = gr.Number( | |
label="Isolation Forest Contamination", | |
value=0.1, | |
minimum=0.01, | |
maximum=0.5, | |
step=0.01, | |
) | |
anomaly_btn = gr.Button("Detect") | |
anomaly_out = gr.Textbox(label="Anomaly Result") | |
anomaly_btn.click( | |
fn=detect_anomalies, | |
inputs=[ | |
sensor_id2, | |
start_time2, | |
end_time2, | |
threshold, | |
algorithm, | |
contamination, | |
], | |
outputs=anomaly_out, | |
) | |
with gr.Tab("Analyze Trends"): | |
sensor_id3 = gr.Textbox(label="Sensor ID", value=example_sensor) | |
start_time3 = gr.Textbox(label="Start Time (ISO)", value=example_start) | |
end_time3 = gr.Textbox(label="End Time (ISO)", value=example_end) | |
trend_btn = gr.Button("Analyze") | |
trend_out = gr.Textbox(label="Trend Result") | |
trend_btn.click( | |
fn=analyze_trends, | |
inputs=[sensor_id3, start_time3, end_time3], | |
outputs=trend_out, | |
) | |
with gr.Tab("Generate Report"): | |
sensor_id4 = gr.Textbox(label="Sensor ID", value=example_sensor) | |
start_time4 = gr.Textbox(label="Start Time (ISO)", value=example_start) | |
end_time4 = gr.Textbox(label="End Time (ISO)", value=example_end) | |
include_anomalies = gr.Checkbox(label="Include Anomalies", value=True) | |
include_trends = gr.Checkbox(label="Include Trends", value=True) | |
user_question = gr.Textbox(label="User Question", value="") | |
anomaly_algorithm = gr.Radio( | |
label="Anomaly Detection Algorithm", | |
choices=["zscore", "isolation_forest"], | |
value="zscore", | |
) | |
anomaly_threshold = gr.Number(label="Z-Score Threshold", value=2.0) | |
anomaly_contamination = gr.Number( | |
label="Isolation Forest Contamination", | |
value=0.1, | |
minimum=0.01, | |
maximum=0.5, | |
step=0.01, | |
) | |
report_btn = gr.Button("Generate Report") | |
report_out = gr.Markdown(label="Report") | |
report_btn.click( | |
fn=generate_analysis_report, | |
inputs=[ | |
sensor_id4, | |
start_time4, | |
end_time4, | |
include_anomalies, | |
include_trends, | |
user_question, | |
anomaly_algorithm, | |
anomaly_threshold, | |
anomaly_contamination, | |
], | |
outputs=report_out, | |
) | |
if __name__ == "__main__": | |
demo.launch(mcp_server=True) | |