Refactor check_thresholds and run_troubleshooting functions: Enhance type annotations, improve alert generation logic, and update return messages for better clarity and consistency.

app.py

@@ -89,23 +89,34 @@ import pandas as pd
 import supabase
 import datetime # Import datetime here as it's used in run_troubleshooting
 import pytz # Import pytz for timezone conversion
-
-
-# Assuming the data loading and check_thresholds function from the previous cell are available
-
-# トラブルシューティング実行関数の定義
-# ① check_thresholds の戻り DataFrameは列を先に固定して作る
-def check_thresholds(sensor_df_filtered, threshold_df):
-    alerts = []
+from typing import List, Dict, Union
+
+# 閾値チェック関数
+def check_thresholds(sensor_df_filtered: pd.DataFrame, threshold_df: pd.DataFrame) -> pd.DataFrame:
+    """
+    センサーデータに対して閾値チェックを行い、下限値未満や上限値超過を検出する。
+
+    Args:
+        sensor_df_filtered (pd.DataFrame): 対象期間で抽出したセンサーデータ。
+            - 必須列: "datetime"（時刻情報）, センサー値列（指標名と一致する列）
+        threshold_df (pd.DataFrame): 閾値情報のデータフレーム。
+            - 必須列: "指標名", "下限", "上限", "No."
+
+    Returns:
+        pd.DataFrame: 異常が検出された場合の結果データフレーム。
+            - 列: ["timestamp", "metric", "value", "status", "data no."]
+            - 検出されなければ空の DataFrame（ただし列は固定）。
+    """
+    alerts: List[Dict[str, Union[str, float, datetime.datetime]]] = []
 
     threshold_df['下限'] = pd.to_numeric(threshold_df['下限'], errors='coerce')
     threshold_df['上限'] = pd.to_numeric(threshold_df['上限'], errors='coerce')
 
     for _, row in threshold_df.iterrows():
-        metric = row["指標名"]
-        min_val = row["下限"]
-        max_val = row["上限"]
-        data_no = row["No."]
+        metric: str = row["指標名"]
+        min_val: float = row["下限"]
+        max_val: float = row["上限"]
+        data_no: int = row["No."]
 
         if metric not in sensor_df_filtered.columns:
             continue
@@ -116,8 +127,7 @@ def check_thresholds(sensor_df_filtered, threshold_df):
             if index not in sensor_df_filtered.index:
                 continue
 
-            # 常に 'datetime' 列を優先し、無ければ index を使う
-            timestamp = (
+            timestamp: Union[pd.Timestamp, int] = (
                 sensor_df_filtered.loc[index, "datetime"]
                 if "datetime" in sensor_df_filtered.columns else index
             )
@@ -126,7 +136,7 @@ def check_thresholds(sensor_df_filtered, threshold_df):
                 alerts.append({
                     "timestamp": timestamp,
                     "metric": metric,
-                    "value": value,
+                    "value": float(value),
                     "status": f"下限値 {min_val} 未満",
                     "data no.": data_no
                 })
@@ -135,33 +145,40 @@ def check_thresholds(sensor_df_filtered, threshold_df):
                 alerts.append({
                     "timestamp": timestamp,
                     "metric": metric,
-                    "value": value,
+                    "value": float(value),
                     "status": f"上限値 {max_val} 超過",
                     "data no.": data_no
                 })
 
-    # ← ここで列を固定。空でも 'timestamp' などの列が存在するようにする
     return pd.DataFrame(alerts, columns=["timestamp", "metric", "value", "status", "data no."])
 
 
-# ② run_troubleshooting 内の空チェックとタイムゾーン担保
 # トラブルシューティング実行関数
-def run_troubleshooting(hours: int = 24):   # ← デフォルト24時間
+def run_troubleshooting(hours: int = 24) -> str:
+    """
+    指定時間内のセンサーデータを対象に閾値チェックを行い、
+    異常が同時に複数指標で発生した場合に対応策を返す。
+
+    Args:
+        hours (int, optional): 過去何時間分のデータをチェックするか。デフォルトは24。
+
+    Returns:
+        str: トラブルシューティング情報のテキスト。
+            - 異常がない場合: 「過去◯時間 異常ありません」
+            - 閾値超過がある場合: タイムスタンプと状況・解決策の一覧
+            - エラー時: エラーメッセージ
+    """
     try:
         current_time_utc = datetime.datetime.now(datetime.timezone.utc)
-
-        # ユーザー指定の時間分さかのぼる
         time_start_utc = current_time_utc - datetime.timedelta(hours=hours)
 
         global sensor_df, threshold_df, troubleshooting_df
 
-        # 指定範囲のセンサーデータを抽出
         recent_sensor_df = sensor_df[
             (sensor_df['datetime'] >= time_start_utc) &
             (sensor_df['datetime'] <= current_time_utc)
         ].copy()
 
-        # 閾値チェック実行
         alerts_df = check_thresholds(recent_sensor_df, threshold_df)
 
         if alerts_df.empty:
@@ -171,28 +188,24 @@ def run_troubleshooting(hours: int = 24):   # ← デフォルト24時間
         multiple_data_nos_timestamps = grouped_alerts[grouped_alerts > 1].index.tolist()
 
         filtered_alerts_df = alerts_df[alerts_df['timestamp'].isin(multiple_data_nos_timestamps)]
-
-        # ここで空になるケースにも対応
         if filtered_alerts_df.empty:
-            return "過去24時間 異常ありません（複数指標の同時異常なし）"
+            return f"過去{hours}時間 異常ありません（複数指標の同時異常なし）"
 
         data_nos_by_timestamp = filtered_alerts_df.groupby('timestamp')['data no.'].unique().apply(list)
 
-        result_list = []
+        result_list: List[Dict[str, Union[str, datetime.datetime]]] = []
         for timestamp, data_nos in data_nos_by_timestamp.items():
             data_nos_str = ', '.join(map(str, data_nos))
             result_list.append({"timestamp": timestamp, "data_nos": data_nos_str})
 
         result_df = pd.DataFrame(result_list, columns=["timestamp", "data_nos"])
 
-        # JST に変換（常に tz-aware 前提）
         JST = pytz.timezone('Asia/Tokyo')
         result_df['timestamp'] = result_df['timestamp'].dt.tz_convert(JST)
 
         if result_df.empty:
-            return "過去24時間 異常ありません"
+            return f"過去{hours}時間 異常ありません"
 
-        # 以下、トラブルシューティング照合
         if '指標No.' not in troubleshooting_df.columns:
             return "設定テーブルに『指標No.』列が見つかりません。"
 
@@ -203,12 +216,11 @@ def run_troubleshooting(hours: int = 24):   # ← デフォルト24時間
             lambda x: [int(i) for i in x if i.strip().isdigit()]
         )
 
-        output_text = ""
+        output_text: str = ""
         for i, result_nos in enumerate(result_data_nos_lists):
             result_timestamp = result_df.loc[i, 'timestamp']
             for j, troubleshooting_nos in enumerate(troubleshooting_indicator_lists):
                 if set(troubleshooting_nos).issubset(set(result_nos)):
-                    # 列の存在チェックも加える
                     if ('シチュエーション\n（対応が必要な状況）' in troubleshooting_df.columns and
                         'sub goal到達のために必要な行動\n（解決策）' in troubleshooting_df.columns):
                         troubleshooting_situation = troubleshooting_df.loc[j, 'シチュエーション\n（対応が必要な状況）']
@@ -227,7 +239,6 @@ def run_troubleshooting(hours: int = 24):   # ← デフォルト24時間
     except Exception as e:
         return f"エラーが発生しました: {type(e).__name__} - {e}"
 
-
 # Gradioインターフェースの設定
 iface = gr.Interface(
     fn=run_troubleshooting,