消費電力でキングスライムの色が変わる！楽しい電力ビジュアライズ

スマートメーター
    ↓ (ECHONET Lite / Bルーター接続)
Nature Remo E lite
    ↓ (Nature Remo クラウドAPI)
[電力記録プログラム] ← crontab 1分ごとに実行
    ↓ (InfluxDB v2 に書き込み)
InfluxDB
    ↑ (15秒ごとに最新値を取得)
[スライムライト制御プログラム] ← systemd サービスとして常駐
    ↓ (SwitchBot API v1.1)
SwitchBot Hub
    ↓
SwitchBot カラー電球 → 色が変わる！

{
    "switchbot": {
        "api": "<your-switchbot-token>",
        "secret": "<your-switchbot-secret>",
        "api_url_v11": "https://api.switch-bot.com/v1.1/",
        "device_id": {
            "slime_light_bulb": "<カラー電球のデバイスID>",
            "living_aircon_sensor": "<照度検知用コンタクトセンサーのデバイスID>"
        }
    },
    "remo": {
        "api": "<your-nature-remo-token>",
        "api_url_appliances": "https://api.nature.global/1/appliances",
        "remo_e_no": 0
    }
}

# -*- coding: utf-8 -*-
import sys
import os
import requests
import datetime
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS

# InfluxDB 接続設定
INFLUXDB_URL    = "http://localhost:8086"
INFLUXDB_TOKEN  = "<your-influxdb-token>"
INFLUXDB_ORG    = "<your-org>"
INFLUXDB_BUCKET = "houseinfo"

# Nature Remo API 設定
REMO_API_TOKEN        = "<your-nature-remo-token>"
REMO_API_URL          = "https://api.nature.global/1/appliances"
REMO_E_INDEX          = 0  # Remo E lite が appliances の何番目か

def fetch_smart_meter_data():
    """Nature Remo E lite API からスマートメーター情報を取得する"""
    headers = {
        'Authorization': 'Bearer ' + REMO_API_TOKEN,
        'accept': 'application/json'
    }
    appliances = requests.get(REMO_API_URL, headers=headers).json()
    props = appliances[REMO_E_INDEX]['smart_meter']['echonetlite_properties']
    return {item['name']: item['val'] for item in props}

def write_to_influxdb(data):
    """取得した電力データを InfluxDB に書き込む"""
    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
    write_api = client.write_api(write_options=SYNCHRONOUS)

    point = (
        Point("remo_e")
        .tag("deviceType", "Remo E lite")
        .field("measured_instantaneous",
               int(data['measured_instantaneous']))
        .field("normal_direction_cumulative_electric_energy",
               int(data['normal_direction_cumulative_electric_energy']))
        .time(datetime.datetime.utcnow(), WritePrecision.SECONDS)
    )
    write_api.write(bucket=INFLUXDB_BUCKET, org=INFLUXDB_ORG, record=point)
    client.close()

if __name__ == '__main__':
    try:
        data = fetch_smart_meter_data()
        write_to_influxdb(data)
        print("電力データの記録が完了しました: " + str(data['measured_instantaneous']) + "W")
    except Exception as e:
        print("エラー: " + str(e))
        sys.exit(1)

# crontabを編集
crontab -e

# 以下を追記（1分ごとに実行）
* * * * * /usr/bin/python3 /home/<username>/projects/smarthome/core/record/energy/home_energy_status_record.py >> /home/<username>/projects/smarthome/logs/crontab/home_energy_status.log 2>&1

# -*- coding: utf-8 -*-
import sys
import os
import time
import json
import requests
import hashlib
import hmac
import base64
import uuid
import datetime
from influxdb_client import InfluxDBClient
from influxdb_client.client.query_api import QueryApi

# ---- InfluxDB 接続設定 ----
INFLUXDB_URL    = "http://localhost:8086"
INFLUXDB_TOKEN  = "<your-influxdb-token>"
INFLUXDB_ORG    = "<your-org>"
INFLUXDB_BUCKET = "houseinfo"

# ---- SwitchBot API 設定 ----
SWITCHBOT_TOKEN           = "<your-switchbot-token>"
SWITCHBOT_SECRET          = "<your-switchbot-secret>"
SWITCHBOT_API_URL         = "https://api.switch-bot.com/v1.1/"
SLIME_BULB_DEVICE_ID      = "<カラー電球のデバイスID>"
LIVING_SENSOR_DEVICE_ID   = "<照度センサーのデバイスID>"  # コンタクトセンサー

# ---- スライムのカラー定義（ドラクエモチーフ） ----
SLIME_COLOR = {
    'white':  '255:255:255',  # メタルキング    (~300W未満)
    'blue':   '0:118:255',    # キングスライム  (~500W未満)
    'green':  '36:255:0',     # スライムベホマズン (~1000W未満)
    'yellow': '255:174:0',    # (~1500W未満)
    'orange': '255:57:17',    # (~2000W未満)
    'purple': '255:0:255',    # (~2500W未満)
    'red':    '255:0:0'        # (2500W以上)
}

def get_switchbot_header():
    """SwitchBot API v1.1 用の認証ヘッダーを生成する"""
    nonce  = str(uuid.uuid4())
    t      = int(round(time.time() * 1000))
    string = '{}{}{}'.format(SWITCHBOT_TOKEN, t, nonce)
    sign   = base64.b64encode(
        hmac.new(SWITCHBOT_SECRET.encode('utf-8'),
                 msg=string.encode('utf-8'),
                 digestmod=hashlib.sha256).digest()
    )
    return {
        'Authorization': SWITCHBOT_TOKEN,
        'sign':          str(sign, 'utf-8'),
        't':             str(t),
        'nonce':         nonce,
        'Content-Type':  'application/json; charset: utf8'
    }

def switchbot_command(device_id, command, parameter='default'):
    """SwitchBot デバイスにコマンドを送信する"""
    url  = SWITCHBOT_API_URL + 'devices/' + device_id + '/commands'
    body = {'command': command, 'parameter': parameter, 'commandType': 'command'}
    requests.post(url, headers=get_switchbot_header(), json=body)

def get_latest_power_w():
    """InfluxDB から最新の瞬間消費電力(W)を取得する"""
    client    = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
    query_api = client.query_api()
    query = '''
        from(bucket: "houseinfo")
          |> range(start: -5m)
          |> filter(fn: (r) => r._measurement == "remo_e")
          |> filter(fn: (r) => r._field == "measured_instantaneous")
          |> last()
    '''
    tables = query_api.query(query, org=INFLUXDB_ORG)
    client.close()
    for table in tables:
        for record in table.records:
            return int(record.get_value())
    return 0

def get_latest_brightness():
    """InfluxDB から照度センサーの最新状態を取得する"""
    client    = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
    query_api = client.query_api()
    query = '''
        from(bucket: "houseinfo")
          |> range(start: -5m)
          |> filter(fn: (r) => r._measurement == "contact_sensor")
          |> filter(fn: (r) => r._field == "brightness")
          |> filter(fn: (r) => r.deviceId == "''' + LIVING_SENSOR_DEVICE_ID + '''")
          |> last()
    '''
    tables = query_api.query(query, org=INFLUXDB_ORG)
    client.close()
    for table in tables:
        for record in table.records:
            return record.get_value()
    return 'unknown'

def get_slime_color(watts):
    """消費電力(W)に応じてスライムの色(RGB文字列)を返す"""
    if watts < 300:
        return SLIME_COLOR['white']
    elif watts < 500:
        return SLIME_COLOR['blue']
    elif watts < 1000:
        return SLIME_COLOR['green']
    elif watts < 1500:
        return SLIME_COLOR['yellow']
    elif watts < 2000:
        return SLIME_COLOR['orange']
    elif watts < 2500:
        return SLIME_COLOR['purple']
    else:
        return SLIME_COLOR['red']

def main():
    color_before         = None
    is_over_3000w_before = None
    brightness_before    = None
    is_lighting          = None

    print("スライムライト制御を開始します。")

    while True:
        try:
            # InfluxDB から最新データを取得
            watts      = get_latest_power_w()
            brightness = get_latest_brightness()

            print(datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S')
                  + " | 消費電力: " + str(watts) + "W | 照度: " + str(brightness))

            # ---- 照度の変化に応じて電球を自動 ON/OFF ----
            if brightness != brightness_before:
                if brightness == 'bright':
                    switchbot_command(SLIME_BULB_DEVICE_ID, 'turnOn')
                    is_lighting = True
                    print("電球を点灯しました。")
                else:
                    switchbot_command(SLIME_BULB_DEVICE_ID, 'turnOff')
                    is_lighting = False
                    print("電球を消灯しました。")

            # ---- 消費電力に応じてスライムの色を決定 ----
            is_over_3000w = (watts >= 3000)
            color = get_slime_color(watts)

            # 電球が点灯中かつ前回と状態が変わった場合のみ API を叩く
            color_changed    = (color != color_before)
            warning_changed  = (is_over_3000w != is_over_3000w_before)

            if is_lighting and (color_changed or warning_changed):
                # 電球の色を変更
                switchbot_command(SLIME_BULB_DEVICE_ID, 'setColor', color)
                print("電球色を変更しました: " + color + " (" + str(watts) + "W)")

                # 3000W 超えの場合は輝度を最大にして警告演出
                if warning_changed and is_over_3000w_before is not None:
                    if is_over_3000w:
                        switchbot_command(SLIME_BULB_DEVICE_ID, 'setBrightness', '100')
                        print("警告！3000W超え → 輝度を最大にしました。")
                    else:
                        switchbot_command(SLIME_BULB_DEVICE_ID, 'setBrightness', '1')
                        print("3000W以下に戻りました → 輝度を最小にしました。")

            # 状態を保存（次回比較用）
            color_before         = color
            is_over_3000w_before = is_over_3000w
            brightness_before    = brightness

        except Exception as e:
            print("エラー: " + str(e))

        time.sleep(15)

if __name__ == '__main__':
    main()

[Unit]
Description=Smarthome Slime Light Control
After=network.target influxdb.service

[Service]
Type=simple
User=<username>
ExecStart=/usr/bin/python3 /home/<username>/projects/smarthome/core/display/energy_saving/slime_light_control.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

# サービスを登録して自動起動を有効化
sudo systemctl daemon-reload
sudo systemctl enable smh-slime-light-control.service

# サービスを起動
sudo systemctl start smh-slime-light-control.service

# 起動状態を確認
sudo systemctl status smh-slime-light-control.service

SLIME_COLOR = {
    'white':  '255:255:255',  # メタルキング
    'blue':   '0:118:255',    # キングスライム
    'green':  '36:255:0',     # スライムベホマズン
    'yellow': '255:174:0',
    'orange': '255:57:17',
    'purple': '255:0:255',
    'red':    '255:0:0'
}

def get_slime_color(watts):
    if watts < 300:   return SLIME_COLOR['white']   # 待機電力レベル
    elif watts < 500:  return SLIME_COLOR['blue']    # 通常使用
    elif watts < 1000: return SLIME_COLOR['green']   # やや多め
    elif watts < 1500: return SLIME_COLOR['yellow']  # エアコン1台程度
    elif watts < 2000: return SLIME_COLOR['orange']  # エアコン+α
    elif watts < 2500: return SLIME_COLOR['purple']  # 使いすぎ注意
    else:              return SLIME_COLOR['red']      # 危険域

# 前回と変化がない場合は API を呼ばない
color_changed   = (color != color_before)
warning_changed = (is_over_3000w != is_over_3000w_before)

if is_lighting and (color_changed or warning_changed):
    switchbot_command(SLIME_BULB_DEVICE_ID, 'setColor', color)

def get_switchbot_header():
    nonce  = str(uuid.uuid4())
    t      = int(round(time.time() * 1000))
    string = '{}{}{}'.format(SWITCHBOT_TOKEN, t, nonce)
    sign   = base64.b64encode(
        hmac.new(SWITCHBOT_SECRET.encode('utf-8'),
                 msg=string.encode('utf-8'),
                 digestmod=hashlib.sha256).digest()
    )
    return {
        'Authorization': SWITCHBOT_TOKEN,
        'sign': str(sign, 'utf-8'),
        't': str(t),
        'nonce': nonce,
    }

python3 /home/<username>/projects/smarthome/core/display/energy_saving/slime_light_control.py

スライムライト制御を開始します。
2026/05/23 10:00:00 | 消費電力: 430W | 照度: bright
電球を点灯しました。
電球色を変更しました: 0:118:255 (430W)
2026/05/23 10:00:15 | 消費電力: 435W | 照度: bright
2026/05/23 10:00:30 | 消費電力: 1250W | 照度: bright
電球色を変更しました: 255:174:0 (1250W)