Nature Remo E でスマートメーターの電力データを自動収集 ─ 瞬時電力・積算電力量を InfluxDBに蓄積

[
    {"name": "measured_instantaneous",                            "val": "842"},
    {"name": "normal_direction_cumulative_electric_energy",       "val": "123456"},
    {"name": "reverse_direction_cumulative_electric_energy",      "val": "0"},
    {"name": "coefficient",                                       "val": "1"},
    {"name": "cumulative_electric_energy_unit",                   "val": "1"},
    {"name": "cumulative_electric_energy_effective_digits",       "val": "6"}
]

# 1分ごとに起動し、15秒間隔で4回実行（0秒・15秒・30秒・45秒）
* * * * * for i in `seq 0 15 59`; do (sleep ${i}; python /path/to/home_energy_status_record.py >> /path/to/home_energy_status.log 2>&1) & done;

{
    "remo": {
        "api": "<Nature Remo アクセストークン>",
        "api_url_appliances": "https://api.nature.global/1/appliances",
        "remo_e_no": 2
    }
}

pip install requests

# -*- coding: utf-8 -*-
import sys
import requests
import datetime
from influx.v2 import client as influx_client

# 設定
REMO_API_TOKEN  = '<Nature Remo アクセストークン>'
REMO_API_URL    = 'https://api.nature.global/1/appliances'
REMO_E_INDEX    = 2    # appliances レスポンス中の Remo E のインデックス（0始まり）
INFLUX_MEASUREMENT = 'remo_e'

def smame_get() -> dict:
    """Nature Remo E API からスマートメーターの ECHONET Lite プロパティを取得して dict に変換する。"""
    headers = {
        'Authorization': 'Bearer ' + REMO_API_TOKEN,
        'accept': 'application/json'
    }
    appliances = requests.get(REMO_API_URL, headers=headers).json()

    # 指定インデックスの Remo E からプロパティリストを取得
    prop_list = appliances[REMO_E_INDEX]['smart_meter']['echonetlite_properties']

    # リスト → 辞書に変換 {name: val}
    result = {}
    for item in prop_list:
        result[item['name']] = item['val']

    return result

def insert_energy_data(dict_elc: dict):
    """電力データを InfluxDB に書き込む。増分フィールドも合わせて記録する。"""

    # InfluxDB から直近の積算電力量を取得（増分計算に使用）
    res = influx_client.query_dict_by_influxql(
        INFLUX_MEASUREMENT,
        {'normal_direction_cumulative_electric_energy'},
        query_mode=influx_client.QueryMode.LAST
    )
    if res:
        prev_cumulative = int(res[0].get('normal_direction_cumulative_electric_energy'))
    else:
        prev_cumulative = 0

    current_cumulative = int(dict_elc['normal_direction_cumulative_electric_energy'])

    # InfluxDB への書き込みデータを構築
    json_point = {
        'measurement': INFLUX_MEASUREMENT,
        'tags': {
            'deviceType':  'Remo E lite',
            'deviceModel': 'Nature Remo E lite',
        },
        'fields': {
            # 瞬時電力（W）
            'measured_instantaneous':
                int(dict_elc['measured_instantaneous']),
            # 正方向積算電力量（電力会社から買った電力の累計）
            'normal_direction_cumulative_electric_energy':
                current_cumulative,
            # 逆方向積算電力量（太陽光などで売電した電力の累計）
            'reverse_direction_cumulative_electric_energy':
                int(dict_elc['reverse_direction_cumulative_electric_energy']),
            # 係数・単位・有効桁数
            'coefficient':
                int(dict_elc['coefficient']),
            'cumulative_electric_energy_unit':
                int(dict_elc['cumulative_electric_energy_unit']),
            'cumulative_electric_energy_effective_digits':
                int(dict_elc['cumulative_electric_energy_effective_digits']),
            # 前回からの増分（今日の消費電力量計算に使用）
            'normal_direction_cumulative_electric_energy_increment':
                current_cumulative - prev_cumulative,
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
    }

    influx_client.write(json_point)

if __name__ == '__main__':
    try:
        dict_elc = smame_get()
        insert_energy_data(dict_elc)
        print('正常に完了しました。')
    except Exception as e:
        print(f'エラー: {e}')
        sys.exit(1)

crontab -e

# 1分ごとに起動し、15秒間隔で4回実行
* * * * * for i in `seq 0 15 59`; do (sleep ${i}; /home/<username>/Projects/venv/bin/python /home/<username>/Projects/smarthome/core/record/energy/home_energy_status_record.py >> /home/<username>/Projects/smarthome/logs/crontab/home_energy_status.log 2>&1) & done;

curl -X GET "https://api.nature.global/1/appliances" \
  -H "Authorization: Bearer <アクセストークン>" \
  -H "accept: application/json" | python3 -m json.tool | grep -E '"type"|"smart_meter"' | head -20

python home_energy_status_record.py

tail -f /home/<username>/Projects/smarthome/logs/crontab/home_energy_status.log