「2025年12月22日で太陽光発電 通知メールがなくなる件。 代替として自前で収集した情報をLINEに通知してみた」をさらに高度化を検討してみた。
過去の実装（太陽光発電システムの故障判断アルゴリズム（その１））で、故障検知を高精度（6年間で5回検知・誤検知なし。1件はパワコンブレーカーダウン、4件はデータアップロードネットワークのダウン）で実装できていましたが、さきのソーラーフロンティアのサービス終了でこの検知システムが使えなくなりました。そこで、新たな検知システムの実装を、先に公開してシステムを強化して、検討してみます。
ちなみに、公開済みの発電量収集代替実装では、従来のソーラーロンティアの発電量との誤差は±2%程度なので、３か所ともに発電量収集代替実装を導入して、既存の検知システムにデータを流し込めば高精度での判定を継続できるようになります。しかし、通常、太陽光発電システムを複数保有しているほぼないので、この方式はほぼ使えません。そこで、ここでは1システムだけであっても判定することができる仕組みを検討してみます。
故障検知実現の設計方針
最終的には、人が判断するが、できるだけ判定精度を上げたい。過去の高精度判定システム構築での知見を反映して設計する。ポイントは次の通り。
・快晴時の発電量は安定しており、きれいなカーブを描く。
→判定はこの部分のデータを使う。
・雲がある晴れの日のデータは変化が激しい。
・発電量の小さい日の発電量のばらつきが大きい。
→判定には使用せず、参考値として通知する。
・四季の変化での発電量の補正が必要
→過去実装では、実績のピークデータをもとに関数化して組み込み
想定発電量と比較し、判定するかどうかを切り分けした。
・外部情報として最も近い気象庁観測点のデータを使用する。
・太陽光パネルの様々な設置場所ごとの違いによる想定発電量を単純な計算式ではなく学習方式を採用する。
使用するデータは、以前に公開したParquet形式で保存したものとする。
実装例
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import os
import sys
import argparse
import requests
from bs4 import BeautifulSoup
# ==================== 設定エリア ====================
NAS_PATH = r"/mnt/nas/PowerData" #★
MODEL_PATH = os.path.join(NAS_PATH, "learning_model.parquet") # 学習データ保存先
IFTTT_WEBHOOK_URL0 = "https://maker.ifttt.com/trigger/{event}/with/key/{your_key}"
IFTTT_EVENT_NAME = "message_to_myline" #★
your_key = "＊＊＊＊＊＊＊＊＊＊＊＊" #★
IFTTT_WEBHOOK_URL = IFTTT_WEBHOOK_URL0.replace("{your_key}", your_key)
# 判定しきい値 (期待値の50%以下が2時間続いたら故障疑いなど)
THRESHOLD_RATIO = 0.5 #★
# ====================================================
def send_line_message(title: str, body: str, timestamp: str = "-"):
payload = {"value1": title, "value2": body, "value3": timestamp}
url = IFTTT_WEBHOOK_URL.replace("{event}", IFTTT_EVENT_NAME)
try:
response = requests.post(url, json=payload, timeout=10)
return response.status_code == 200
except Exception as e:
print(f"LINE通知失敗: {e}")
return False
def get_jma_sunshine_hourly(target_date: datetime):
"""気象庁HPから10分ごとの日照時間を取得し1時間単位(0.0-1.0)で返す"""
url = (f"https://www.data.jma.go.jp/stats/etrn/view/10min_s1.php?"
f"prec_no=＊＊&block_no=＊＊＊＊&year={target_date.year}&month={target_date.month}&day={target_date.day}&view=p2") #★
try:
res = requests.get(url, timeout=10)
res.encoding = res.apparent_encoding
soup = BeautifulSoup(res.text, 'html.parser')
rows = soup.find_all('tr', class_='mtx')
sunshine_10min = []
for row in rows:
cols = row.find_all('td')
if len(cols) > 10:
val = cols[9].text.strip() # 日照時間の列
sunshine_10min.append(float(val) if val else 0.0)
# 1時間(6個)ごとに合計(最大1.0)
return [sum(sunshine_10min[i:i+6]) for i in range(0, len(sunshine_10min), 6)]
except Exception as e:
print(f"気象庁データ取得失敗: {e}")
return [None] * 24
def update_learning_model(target_date: datetime, hourly_actual: list, hourly_sun: list):
"""日照が良い時のデータを『快晴時の正解』として学習モデルを更新"""
month = target_date.month
try:
if os.path.exists(MODEL_PATH):
model_df = pd.read_parquet(MODEL_PATH)
else:
# 初期モデル: 全月0で作成
model_df = pd.DataFrame(0.0, index=range(1, 13), columns=[f"h_{i}" for i in range(24)])
updated = False
for h in range(7, 18): # 発電時間帯のみ
# 日照率が0.9以上かつ、これまでの学習値より高い(または初データ)場合に更新
if hourly_sun[h] and hourly_sun[h] >= 0.9:
current_val = model_df.loc[month, f"h_{h}"]
if hourly_actual[h] > current_val:
model_df.loc[month, f"h_{h}"] = hourly_actual[h]
updated = True
if updated:
model_df.to_parquet(MODEL_PATH)
print("学習モデルを更新しました。")
except Exception as e:
print(f"学習モデル更新失敗: {e}")
def load_day_data(target_date: datetime) -> pd.DataFrame:
date_str = target_date.strftime("%Y%m%d")
file_path = fr"{NAS_PATH}/power_{date_str}.parquet"
if not os.path.exists(file_path):
raise FileNotFoundError(f"データが見つかりません: {file_path}")
return pd.read_parquet(file_path)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--da", type=int, default=0)
args = parser.parse_args()
base_date = datetime.now() + timedelta(days=args.da)
date_label = base_date.strftime("%Y/%m/%d")
try:
df = load_day_data(base_date)
# 1時間ごとの発電量(value1)リスト作成
df['hour'] = pd.to_datetime(df['timestamp']).dt.hour # timestamp列があると想定
hourly_actual = df.groupby('hour')['value1'].mean().reindex(range(24), fill_value=0.0).tolist()
except Exception as e:
print(f"データ読み込みエラー: {e}")
return
# 気象庁データ取得
hourly_sun = get_jma_sunshine_hourly(base_date)
# 学習モデルの更新
update_learning_model(base_date, hourly_actual, hourly_sun)
# 故障検知
is_anomaly = False
anomaly_details = []
if os.path.exists(MODEL_PATH):
model_df = pd.read_parquet(MODEL_PATH)
month = base_date.month
for h in range(8, 17): # 影の影響が少ない時間帯を重点チェック
sun = hourly_sun[h]
if sun and sun > 0.3: # ある程度の日照がある場合
expected = model_df.loc[month, f"h_{h}"] * sun
actual = hourly_actual[h]
if actual < expected * THRESHOLD_RATIO:
is_anomaly = True
anomaly_details.append(f"{h}時(実測{actual:.1f}kW/期待{expected:.1f}kW)")
# メッセージ作成
total_gen = sum(hourly_actual) * (10/60) # 10分データの場合の概算
status_msg = "✅ 正常" if not is_anomaly else "⚠️ 故障の疑い"
message = f"⚡️ {date_label} 発電実績 ({status_msg})\n" \
f"総発電量：{total_gen:.2f} kWh\n"
if is_anomaly:
message += "\n【異常検知】\n" + "\n".join(anomaly_details)
send_line_message("太陽光発電アラート", message, "ALERT")
else:
send_line_message("太陽光発電データ", message, "OK")
print(message)
if __name__ == "__main__":
main()
上の実装は、あくまでもサンプルです。 特に”★”印のある行は環境に合わせて修正してください。
このコードで検証していますが、かなり過敏で誤検知が多いです。
閾値調整で、精度を上げることになると思いますが、過去実装での故障判定ロジックを超える精度にはならなさそうです。 誤検知をゼロにするには、十分に学習させたうえで、1日中快晴だった日のデータでのみで判定する必要がありそうです。
他に改善案などあればコメントをお願いします。
関連記事
」