app.py

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