太陽光発電の発電量をロギングしてみた。(7)発電異常をLINEで通知

【太陽光発電の発電量をロギングしてみた。目次】

太陽光パネル不調

Amazonで購入した折り畳み式の120W太陽光パネルが、はずれだった。
発電量が100Wにもいかないのに、昼前になるとオーバーヒートし、発電量がほぼゼロになってしまう。
太陽光パネルとバッテリーとの接続を解除し、パネルに日光が当たらないようにして15分くらい冷却してから再接続すると、再び復活する。
メーカーに問い合わせ、太陽光パネルを交換してもらった。
しかし、数回使用するとまた同じ現象が発生するようになった。
折り畳み式だとパネル間の接合部に負荷がかかり、内部的にどこか抵抗値が増加し、電流の流れがアンバランスになっているのかもしれないが、詳細は不明である。

折り畳み式は収納性と携帯性は高いが、純粋にたくさん発電させたければ固定式のパネルが良いようだ。

太陽光パネルのオーバーヒートを通知させる

太陽光パネルがオーバーヒートするたびに冷却する必要があるが、発電量を常に監視していなければ、オーバーヒートしているのかよくわからない。
そこで、発電量が急激に低下したタイミングでスマホに通知を送ることを考えた。

はじめはメール通知を考えたが、最近DMが多く、メールが届いてもすぐ見ないので気がつかないことが多い。
そこで、LINE通知させることにした。

LINE Notify

LINE通知させるには、LINE Notifyを使用する。

LINE Notify

LINE Notifyを利用するには、LINEのアカウント設定でメールアドレスとパスワードを登録し、上記サイトからログインする必要がある。
詳細は、下記のサイトを参考にさせてもらった。

・【使ってみた】LINE Notifyを使ってトークルームにメッセージ送信
・PythonでLINE Notifyへ通知を送る

ソースへの組み込み

ここでは、発電異常の判定に、3回分の電流測定値を使用する。
1分間隔でサンプリングするが、各サンプリングごとに電流測定値を保持し、前回値と前々回値まで残しておく。
異常とみなすのは次のようなケースである。

N回目のサンプリング値　　 5.5A
N＋1回目のサンプリング値　0.4A
N＋2回目のサンプリング値　0.3A

電流測定値はサンプリングミス等で異常値になることも考えられるので、2回連続異常値となったときにLINE通知を行うようにする。

閾値は以下を設定
正常値：2.0A以上
異常値：0.5A以下

日照量の低下時は、ゆっくりと電流値が下がっていくので、このくらいの閾値であれば誤通知することはまずない。

以下、ソースである

import time
import subprocess
import os
import glob
from datetime import date, datetime
import board
import adafruit_ina260
import requests

import serial
import struct
import datetime as dt
import json
from simple_salesforce import Salesforce
     
USERNAME = "Salesforceユーザー名"
PASSWORD = 'Salesforceパスワード'
SECURITY_TOKEN = "Salesforceセキュリティトークン"

# Salesforceログイン関数
def Login():
    sf = Salesforce(username=USERNAME, password=PASSWORD,
                    security_token=SECURITY_TOKEN)
    return sf

# LINE通知関数
def send_line_notify(notification_message):
    line_notify_token = "LINE Notifyセキュリティトークン"
    line_notify_api = 'https://notify-api.line.me/api/notify'
    #headers = {'Authorization': f'Bearer {line_notify_token}'} # f''表記はPython3.6以降しか使用できない。
    #data = {'message': f'message: {notification_message}'}
    headers = {'Authorization': 'Bearer ' + line_notify_token}
    data = {'message': notification_message}
    requests.post(line_notify_api, headers = headers, data = data)
    
try:
    # 電流測定のための設定
    i2c=board.I2C()
    ina260=adafruit_ina260.INA260(i2c,0x42)
    
    dirs = os.listdir("/sys/bus/w1/devices")
    counter = 0

    # Salesforceにログイン
    sf = Login()
except Exception as ex:
    print(ex)
    
try:
    histCurrent = []  # 電流測定値の履歴保持用配列
    while True:
        # DS18B20温度測定値取得
        for dir in dirs:
            if "28-" in dir:
                counter += 1
                tfiles = subprocess.check_output(['cat','/sys/bus/w1/devices/' + dir + '/w1_slave'])
                stfiles = str(tfiles)
                if "YES" in stfiles :
                    idx = stfiles.find("t=",50,80)
                    temp = int(stfiles[idx +2:-3])/1000.0
                    current = ina260.current/1000.0
                    power = ina260.power/1000.0
                    print("Current: %.2f Voltage: %.2f Power: %.2f Temp: %.2f" % (current, ina260.voltage, power, temp))
                    st = "{0},{1},{2},{3},{4}\n".format(datetime.now(), current, ina260.voltage, power, temp)
                    # ログファイルに測定値出力
                    with open('/home/pi/Documents/current_{}.log'.format(date.today()), mode='a') as f:
                        f.write(st)
                    
                    datas = st.split(",")
                    if len(datas)<3:
                        continue
                    t1 = datas[0][0:19]   # 日時を文字列に
                    t2 = t1.replace(" ","T") + "Z"  # 日時をdatetime型として挿入するために整形
                    #print("{0},{1}".format(t1,t2))
                    try:
                        for i in range(1,len(datas)):
                            datas[i] = datas[i].replace("b'","").replace("\\r\\n'","").strip()   # データから改行や空白を除去
                            
                        if len(datas[1])==0 or len(datas[2])==0 or len(datas[1])>10 or len(datas[2])>10 :
                            continue    # 明らか異常値はアップロードしない
                        if datas[1][:1].isdigit()==False or datas[2][:1].isdigit()==False:
                            continue

                        # Salesforceにアップロード
                        res = sf.GeneratorLogStack__b.create({'No__c ': t1,
                                                              'Date__c': t1,
                                                                'Current__c': float(datas[1]),
                                                                'Voltage__c': float(datas[2]),
                                                                'PowerMeas__c': float(datas[3]),
                                                                'Temperature__c': float(datas[4])})
                        if res.get('error'):
                            # ログインセッションが切れた場合は例外処理へ
                            raise Exception(response.get('error_description'))
                        print(json.dumps(res, indent=6))
                        
                    
                    except Exception as e:
                            # セッション切れを想定し、再ログインと再アップロード
                            sf = Login()
                            res = sf.GeneratorLogStack__b.create({'No__c': t1,
                                                                'Date__c': t1,
                                                                'Current__c': float(datas[1]),
                                                                'Voltage__c': float(datas[2]),
                                                                'PowerMeas__c': float(datas[3]),
                                                                'Temperature__c': float(datas[4])})
                            print(json.dumps(res, indent=6))
        
        # 電流測定値を履歴として保持                
        histCurrent.append(float(datas[1]))
        if(len(histCurrent)>=3):
            # 履歴データによる異常判定
            if(histCurrent[0]>=2.0 and histCurrent[1]<=0.5 and histCurrent[2]<=0.5):
                print("line start")
                # LINE通知
                send_line_notify("Current value down !")
                print("line notify----")
            # 判定後、2回前の履歴を削除
            del histCurrent[0]
        
        time.sleep(60)

except Exception as ex:
    with open('/home/pi/log.txt',mode='a') as f:
        f.write(ex)
    print(ex)

これで、確かにスマホに下記のようなLINE通知が届くようになった。

スマホのLINEの通知設定をONにしておけばすぐに異常に気がつくようにでき、
とても便利である。

判定条件や閾値をカスタマイズし、いろいろ検出したい事象を通知するのもよいかもしれない。