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