app.py

from __future__ import annotations

import time
from dataclasses import dataclass
from pathlib import Path
from typing import Generator, Optional

import gradio as gr
import joblib
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import tensorflow as tf
from sklearn.preprocessing import StandardScaler


@dataclass(frozen=True)
class DemoConfig:
    model_path: Path
    data_path: Path
    scaler_path: Path
    lookback: int = 30
    horizon: int = 60
    threshold: float = 0.5
    refresh_delay: float = 0.08


class DataRepository:
    def __init__(self, data_path: Path) -> None:
        self.data_path = data_path
        self._data: Optional[pd.DataFrame] = None

    def get_data(self) -> pd.DataFrame:
        if self._data is None:
            df = pd.read_csv(self.data_path, parse_dates=["datetime"])
            df = df.sort_values(["user_id", "datetime"]).reset_index(drop=True)
            self._data = df
        return self._data.copy()


class SequenceBuilder:
    def __init__(self, lookback: int) -> None:
        self.lookback = lookback

    def build_window(self, user_frame: pd.DataFrame, feature_cols: list[str], end_index: int) -> np.ndarray:
        start_index = end_index - self.lookback + 1
        window = user_frame.iloc[start_index:end_index + 1][feature_cols].to_numpy(dtype=np.float32)

        if len(window) != self.lookback:
            raise ValueError("Недостаточно данных для построения окна наблюдений.")

        return window


class PredictionService:
    def __init__(self, config: DemoConfig) -> None:
        self.config = config
        self._model: Optional[tf.keras.Model] = None
        self._scaler: Optional[StandardScaler] = None

    @staticmethod
    def _make_weighted_bce(pos_weight: float = 1.0):
        pos_weight_tensor = tf.constant(pos_weight, dtype=tf.float32)

        def weighted_bce(y_true, y_pred):
            y_true_cast = tf.cast(y_true, tf.float32)
            y_pred_clip = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7)

            loss_pos = -y_true_cast * tf.math.log(y_pred_clip) * pos_weight_tensor
            loss_neg = -(1.0 - y_true_cast) * tf.math.log(1.0 - y_pred_clip)

            return tf.reduce_mean(loss_pos + loss_neg)

        return weighted_bce

    def _load_model(self) -> tf.keras.Model:
        if self._model is None:
            self._model = tf.keras.models.load_model(
                self.config.model_path,
                custom_objects={"weighted_bce": self._make_weighted_bce(1.0)},
                compile=False,
            )
        return self._model

    def _load_scaler(self) -> StandardScaler:
        if self._scaler is None:
            if self.config.scaler_path.exists():
                self._scaler = joblib.load(self.config.scaler_path)
            else:
                self._scaler = StandardScaler()
        return self._scaler

    def predict(self, window: np.ndarray) -> np.ndarray:
        model = self._load_model()
        scaler = self._load_scaler()

        if hasattr(scaler, "mean_"):
            window_2d = window.reshape(-1, window.shape[-1])
            scaled_window = scaler.transform(window_2d).reshape(1, window.shape[0], window.shape[1])
        else:
            scaled_window = window.reshape(1, window.shape[0], window.shape[1])

        prediction = model.predict(scaled_window, verbose=0)[0]
        return np.asarray(prediction, dtype=np.float32)


class DemoFormatter:
    @staticmethod
    def risk_band(score: float) -> str:
        if score >= 0.8:
            return "Критический"
        if score >= 0.6:
            return "Высокий"
        if score >= 0.4:
            return "Средний"
        return "Низкий"

    @staticmethod
    def summary_html(max_risk_so_far: float, mean_risk_so_far: float, current_probability: float, horizon_minutes: int) -> str:
        level = DemoFormatter.risk_band(max_risk_so_far)
        color = "#d7263d" if current_probability >= 0.5 else "#16a34a"
        glow = "rgba(215,38,61,0.22)" if current_probability >= 0.5 else "rgba(22,163,74,0.18)"

        return f"""
<div style="border-radius:24px;padding:24px 28px;background:linear-gradient(135deg,#0f172a,#1e293b);color:#fff;box-shadow:0 14px 40px {glow};max-width:1200px;margin:0 auto;">
  <div style="font-size:34px;font-weight:700;margin-bottom:14px;text-align:center;">{level} риск отказа</div>
  <div style="display:flex;gap:16px;flex-wrap:wrap;justify-content:center;">
    <div style="background:rgba(255,255,255,.08);padding:14px 14px;border-radius:16px;min-width:220px;text-align:center;">
      <div style="font-size:13px;opacity:.8;">Текущий шаг</div>
      <div style="font-size:32px;color:{color};font-weight:700;">{current_probability:.1%}</div>
    </div>
    <div style="background:rgba(255,255,255,.08);padding:14px 14px;border-radius:16px;min-width:220px;text-align:center;">
      <div style="font-size:13px;opacity:.8;">Пиковая вероятность</div>
      <div style="font-size:32px;font-weight:700;">{max_risk_so_far:.1%}</div>
    </div>
    <div style="background:rgba(255,255,255,.08);padding:14px 14px;border-radius:16px;min-width:220px;text-align:center;">
      <div style="font-size:13px;opacity:.8;">Средняя вероятность</div>
      <div style="font-size:32px;font-weight:700;">{mean_risk_so_far:.1%}</div>
    </div>
    <div style="background:rgba(255,255,255,.08);padding:14px 14px;border-radius:16px;min-width:220px;text-align:center;">
      <div style="font-size:13px;opacity:.8;">Окно прогноза</div>
      <div style="font-size:32px;font-weight:700;">{horizon_minutes} мин</div>
    </div>
  </div>
</div>
""".strip()


class FailureForecastDemoApp:
    def __init__(self, config: DemoConfig) -> None:
        self.config = config
        self.repository = DataRepository(config.data_path)
        self.sequence_builder = SequenceBuilder(config.lookback)
        self.prediction_service = PredictionService(config)

        self.df = self.repository.get_data()
        self.feature_cols = [col for col in self.df.columns if col not in ["user_id", "datetime"]]
        self.valid_users = self._collect_valid_users()

        self.theme = gr.themes.Soft(primary_hue="blue", secondary_hue="slate", neutral_hue="slate")
        self.custom_css = """
        .gradio-container {
            max-width: 1880px !important;
            margin: 0 auto !important;
        }
        .hero-card {
            background: linear-gradient(135deg, #0f172a 0%, #1e3a8a 45%, #2563eb 100%);
            border-radius: 24px;
            padding: 34px;
            color: white;
            box-shadow: 0 18px 60px rgba(37,99,235,.22);
            margin: 0 auto 22px auto;
            text-align: center;
            max-width: 1320px;
        }
        .hero-card h1 {
            margin: 0 0 10px 0;
            font-size: 42px;
        }
        .hero-card p {
            margin: 0;
            font-size: 20px;
            opacity: .92;
        }
        #controls-wrap, #status-wrap, #plots-wrap, #table-wrap {
            max-width: 1700px;
            margin: 0 auto;
        }
        #plots-wrap .gr-plot {
            min-height: 720px !important;
        }
        .forecast-table {
            margin: 0 auto;
            max-width: 1400px;
        }
        """

    def _collect_valid_users(self) -> list[int]:
        counts = self.df.groupby("user_id").size()
        valid = counts[counts >= self.config.lookback + self.config.horizon].index.tolist()
        return [int(user_id) for user_id in valid]

    def get_user_choices(self) -> list[int]:
        return self.valid_users

    def get_user_index_bounds(self, user_id: int) -> tuple[int, int, int]:
        frame = self.df[self.df["user_id"] == user_id].reset_index(drop=True)
        min_index = self.config.lookback - 1
        max_index = len(frame) - self.config.horizon - 1
        default_index = max_index
        return min_index, max_index, default_index

    def _build_history_plot(
        self,
        user_frame: pd.DataFrame,
        end_index: int,
        metric_name: str,
        prediction_df: pd.DataFrame,
        current_step: int,
    ):
        fig, ax1 = plt.subplots(figsize=(18, 8.5))

        history = user_frame.iloc[max(0, end_index - self.config.lookback + 1): end_index + 1]
        future = prediction_df.iloc[:current_step].copy()

        ax1.plot(
            history["datetime"],
            history[metric_name],
            linewidth=3.0,
            label=f"История: {metric_name}",
        )
        ax1.set_xlabel("Время", fontsize=13)
        ax1.set_ylabel(metric_name, fontsize=13)
        ax1.grid(alpha=0.25)

        ax2 = ax1.twinx()
        ax2.plot(
            future["datetime"],
            future["failure_probability"],
            linestyle="--",
            linewidth=2.8,
            label="Вероятность отказа",
        )

        if not future.empty:
            current_point = future.iloc[-1]
            ax2.scatter(
                [current_point["datetime"]],
                [current_point["failure_probability"]],
                s=120,
                label="Текущий шаг",
            )

        ax2.set_ylabel("P(failure)", fontsize=13)
        ax2.set_ylim(0, 1.05)

        lines1, labels1 = ax1.get_legend_handles_labels()
        lines2, labels2 = ax2.get_legend_handles_labels()
        ax1.legend(lines1 + lines2, labels1 + labels2, loc="upper left", fontsize=12)

        title = f"История и текущий прогноз по метрике '{metric_name}'"
        if current_step > 0:
            current_point = future.iloc[-1]
            title += f" • +{int(current_point['step'])} мин • риск {current_point['failure_probability']:.1%}"

        ax1.set_title(title, fontsize=16)
        ax1.tick_params(axis="both", labelsize=11)
        ax2.tick_params(axis="y", labelsize=11)

        fig.autofmt_xdate()
        plt.tight_layout()
        return fig

    def _build_probability_plot(self, prediction_df: pd.DataFrame, current_step: int):
        fig, ax = plt.subplots(figsize=(18, 7.8))

        shown = prediction_df.iloc[:current_step].copy()
        remaining = prediction_df.iloc[current_step:].copy()

        if not shown.empty:
            ax.plot(
                shown["step"],
                shown["failure_probability"],
                linewidth=3.0,
                label="Раскрытый участок",
            )

            peak_idx = shown["failure_probability"].idxmax()
            peak_row = shown.loc[peak_idx]
            ax.scatter(
                [peak_row["step"]],
                [peak_row["failure_probability"]],
                s=120,
                label="Пик на текущий момент",
            )

            current_row = shown.iloc[-1]
            ax.scatter(
                [current_row["step"]],
                [current_row["failure_probability"]],
                s=120,
                label="Текущий шаг",
            )

        if not remaining.empty:
            ax.plot(
                remaining["step"],
                remaining["failure_probability"],
                linestyle=":",
                linewidth=2.3,
                alpha=0.45,
                label="Оставшийся горизонт",
            )

        ax.set_ylim(0, 1.05)
        ax.set_xlabel("Шаг прогноза, мин", fontsize=13)
        ax.set_ylabel("Вероятность отказа", fontsize=13)
        ax.set_title("Профиль риска на горизонте 60 минут", fontsize=16)
        ax.grid(alpha=0.25)
        ax.legend(loc="upper left", fontsize=12)
        ax.tick_params(axis="both", labelsize=11)

        plt.tight_layout()
        return fig

    def _build_prediction_table(self, forecast_start: pd.Timestamp, prediction: np.ndarray, current_step: int) -> pd.DataFrame:
        current_index = current_step - 1
        current_probability = float(prediction[current_index])
        current_datetime = (forecast_start + pd.Timedelta(minutes=current_index)).strftime("%Y-%m-%d %H:%M:%S")
        current_state = "ALERT" if current_probability >= self.config.threshold else "OK"

        return pd.DataFrame(
            [
                {
                    "step": current_step,
                    "datetime": current_datetime,
                    "failure_probability": f"{current_probability:.2%}",
                    "state": current_state,
                    "progress": "← текущий",
                }
            ]
        )

    def simulate(self, user_id: int, metric_name: str, end_index: int) -> Generator:
        user_frame = self.df[self.df["user_id"] == user_id].reset_index(drop=True)
        min_index, max_index, _ = self.get_user_index_bounds(user_id)
        end_index = int(np.clip(end_index, min_index, max_index))

        history_end_dt = user_frame.loc[end_index, "datetime"]
        forecast_start_dt = history_end_dt + pd.Timedelta(minutes=1)

        window = self.sequence_builder.build_window(user_frame, self.feature_cols, end_index)
        prediction = self.prediction_service.predict(window)

        prediction_df = pd.DataFrame(
            {
                "step": np.arange(1, len(prediction) + 1),
                "datetime": pd.date_range(forecast_start_dt, periods=len(prediction), freq="min"),
                "failure_probability": prediction,
            }
        )

        for step in range(1, self.config.horizon + 1):
            shown_prediction = prediction[:step]
            current_probability = float(shown_prediction[-1])
            max_risk_so_far = float(shown_prediction.max())
            mean_risk_so_far = float(shown_prediction.mean())

            summary_html = DemoFormatter.summary_html(
                max_risk_so_far=max_risk_so_far,
                mean_risk_so_far=mean_risk_so_far,
                current_probability=current_probability,
                horizon_minutes=self.config.horizon,
            )
            history_plot = self._build_history_plot(user_frame, end_index, metric_name, prediction_df, step)
            risk_plot = self._build_probability_plot(prediction_df, step)
            table = self._build_prediction_table(forecast_start_dt, prediction, step)

            yield summary_html, history_plot, risk_plot, table
            time.sleep(self.config.refresh_delay)

    @staticmethod
    def refresh_slider(user_id: int, app: "FailureForecastDemoApp"):
        minimum, maximum, value = app.get_user_index_bounds(user_id)
        return gr.Slider(minimum=minimum, maximum=maximum, value=value, step=1)

    def build(self) -> gr.Blocks:
        metric_choices = [
            metric
            for metric in ["cpus", "memory", "assigned_memory", "page_cache_memory", "failed"]
            if metric in self.df.columns
        ]

        default_user = self.valid_users[0]
        slider_min, slider_max, slider_value = self.get_user_index_bounds(default_user)

        with gr.Blocks(title="Демонстрация прогнозирования отказов") as demo:
            gr.HTML(
                """
                <div class="hero-card">
                    <h1>Предвестник отказов</h1>
                    <p>Интерактивная демонстрация многократного прогноза отказов на горизонте 60 минут. Нажми кнопку запуска — и интерфейс поэтапно покажет, как меняется риск во времени.</p>
                </div>
                """
            )

            with gr.Column(elem_id="controls-wrap"):
                with gr.Row(equal_height=True):
                    user_input = gr.Dropdown(
                        choices=self.get_user_choices(),
                        value=default_user,
                        label="Объект мониторинга (user_id)",
                        info="Доступны только ряды, где хватает истории и горизонта прогноза.",
                        scale=1,
                    )
                    metric_input = gr.Dropdown(
                        choices=metric_choices,
                        value=metric_choices[0],
                        label="Ключевая метрика для визуализации",
                        scale=1,
                    )

                end_index_input = gr.Slider(
                    minimum=slider_min,
                    maximum=slider_max,
                    value=slider_value,
                    step=1,
                    label="Позиция конца окна истории",
                    info="Чем правее, тем ближе к последним данным выбранного объекта.",
                )

                launch_button = gr.Button("Запустить симуляцию", variant="primary", size="lg")

            with gr.Column(elem_id="status-wrap"):
                summary_output = gr.HTML(label="Сводка")

            with gr.Column(elem_id="plots-wrap"):
                with gr.Row(equal_height=True):
                    history_plot_output = gr.Plot(label="История + текущий прогноз", scale=1)
                    risk_plot_output = gr.Plot(label="Кривая риска", scale=1)

            with gr.Column(elem_id="table-wrap"):
                table_output = gr.Dataframe(
                    headers=["step", "datetime", "failure_probability", "state", "progress"],
                    datatype=["number", "str", "str", "str", "str"],
                    interactive=False,
                    label="Поточечный прогноз",
                    wrap=True,
                    row_count=1,
                    column_count=5,
                    max_height=120,
                    elem_classes="forecast-table",
                )

            user_input.change(
                fn=lambda selected_user: self.refresh_slider(selected_user, self),
                inputs=user_input,
                outputs=end_index_input,
            )

            launch_button.click(
                fn=self.simulate,
                inputs=[user_input, metric_input, end_index_input],
                outputs=[summary_output, history_plot_output, risk_plot_output, table_output],
            )

        return demo


def resolve_project_paths() -> DemoConfig:
    base_dir = Path(__file__).resolve().parent
    model_path = (base_dir / "models/gru_direct_multistep/final_model.keras").resolve()
    data_path = (base_dir / "data/processed/data_processed.csv").resolve()
    scaler_path = (base_dir / "models/gru_direct_multistep/seq_scaler.pkl").resolve()

    return DemoConfig(
        model_path=model_path,
        data_path=data_path,
        scaler_path=scaler_path,
    )


def validate_paths(config: DemoConfig) -> None:
    missing_paths = [str(path) for path in [config.model_path, config.data_path] if not path.exists()]
    if missing_paths:
        joined = "\n - ".join(missing_paths)
        raise FileNotFoundError(f"Не найдены обязательные файлы:\n - {joined}")


if __name__ == "__main__":
    tf.keras.utils.set_random_seed(42)

    config = resolve_project_paths()
    validate_paths(config)

    app = FailureForecastDemoApp(config)
    demo = app.build()

    demo.queue().launch(
        server_name="0.0.0.0",
        server_port=7860,
        theme=app.theme,
        css=app.custom_css,
    )