スマートプラグで測定した電⼒値をAPIでi-Reporterと連携してデータ蓄積してみた

はじめまして。 i-Reporterの導入支援などシステム開発を行っている、株式会社ネクストビジョンの友木です。

ネクストビジョンのi-Reporterチームの一員としてi-Reporterの機能に魅了されているひとりです。いつもお世話になっております。

今回のテーマではIoT機器とi-Reporterを連携させる方法を、誰でも購入できるスマートデバイスを使うことで簡単に実現してみました。 IoT機器は機器自体にプログラムを組み込んだり、ネットワークの設定をしたりと難しく感じますが、機能的な製品を上手く組み合わせることで、簡単にi-Reporterと連携できます!ということをお伝えできればと思います。

概要

今回使用するスマートプラグは、Gosundというものです。スマートプラグは、専用アプリを使用して、機器の電源を操作したり、電力値などの情報を見ることができます。 Gosundはアプリから電力値を見ることはもちろんできますが、Tuyaというものと連携して APIで電力値を取得できることが判明したのでi-Reporter と連携してみました。

1時間ごとにAPIを使用してGosundから取得したデータをi-Reporter帳票に蓄積してみようと思います。

Smart Life アプリの設定

1. Smart Life アプリをスマホにインストールします。

2. お手持ちのGosundデバイスをアプリに登録します。

今回は、仕事部屋にあるオイルヒーターのコンセントをGosundにつなぎました。

Tuya IoT Platform の設定

1. Tuya IoT Platformのアカウントを作成します。

2. クラウドプロジェクトを作成します。

Data CenterはWestern America Data Centerを選択してください。

3. 作成したプロジェクトのAccess IDとAccess Secretをどこかにコピーしておきましょう。

4. Smart Life のアカウントを紐づけます。

5. すると、Smart Lifeで登録しているデバイスが追加されるのでDevice IDをコピーしておき、Debug Deviceを選択します。

6. ここでいろいろ情報が見れればOKです。

このあたりで躓いた場合は、こちらの記事を参考にすると分かりやすいと思います。

Pythonでデータ取得するための準備

ライブラリtinytuyaをインストールします。

pip install tinytuya

2. 以下のプログラムを実行します。

 import tinytuya

   c = tinytuya.Cloud(
       apiRegion="us", 
       apiKey="コピーしておいた Access ID", 
       apiSecret="コピーしておいた Access Secret")

   # デバイスID
   device_id = "コピーしておいた Device ID"

   # プロパティ
   result = c.getproperties(device_id)
   print(result)

   # 測定値
   result = c.getstatus(device_id)
   print(result)

今回は会社のPCから自宅のGosundプラグにアクセスするため、CloudのAPIを使用しています。
別の方法でLAN上のIPアドレスを直接指定して、Cloudを介さずに行うこともできます。

3. 実行結果は以下の通りです。

・プロパティ

{'result': {'category': 'cz', 'functions': [{'code': 'switch_1', 'desc': '{}', 'name': '开关1', 'type': 'Boolean', 'values': '{}'}, {'code': 'countdown_1', 'desc': '{"unit":"s","min":0,"max":86400,"scale":0,"step":1}', 'name': '开关1倒计时', 'type': 'Integer', 'values': '{"unit":"s","min":0,"max":86400,"scale":0,"step":1}'}], 'status': [{'code': 'switch_1', 'name': '开关1', 'type': 'Boolean', 'values': '{}'}, {'code': 'countdown_1', 'name': '开关1倒计时', 'type': 'Integer', 'values': '{"unit":"s","min":0,"max":86400,"scale":0,"step":1}'}, {'code': 'add_ele', 'name': '增加电量', 'type': 'Integer', 'values': '{"unit":"","min":0,"max":50000,"scale":3,"step":100}'}, {'code': 'cur_current', 'name': '当前电流', 'type': 'Integer', 'values': '{"unit":"mA","min":0,"max":30000,"scale":0,"step":1}'}, {'code': 'cur_power', 'name': '当前功率', 'type': 'Integer', 'values': '{"unit":"W","min":0,"max":50000,"scale":1,"step":1}'}, {'code': 'cur_voltage', 'name': '当前电压', 'type': 'Integer', 'values': '{"unit":"V","min":0,"max":5000,"scale":1,"step":1}'}]}, 'success': True, 't': 1701134330547, 'tid': '165952118d8c11eebeb4027cbd1575e0'}

Tuya IoT Platform の設定の手順6の画面で確認できる内容と等しいことが分かります。

・測定値

{'result': [{'code': 'switch_1', 'value': True}, {'code': 'countdown_1', 'value': 0}, {'code': 'add_ele', 'value': 1}, {'code': 'cur_current', 'value': 3908}, {'code': 'cur_power', 'value': 3973}, {'code': 'cur_voltage', 'value': 1014}], 'success': True, 't': 1701134331433, 'tid': '16e378d78d8c11eebeb4027cbd1575e0'}

今回必要になる電力値はcur_powerの値です。3973となっていますが、プロパティのscaleが1 になっているので、小数点以下1桁を含んでいるんだと思います。なので397.3Wということですね。

帳票を作成

以降の手順ではi-Reporterを使っていきます。このような簡単な帳票を作成してみました。

1時間ずつ電力が自動で記録される想定で、自動で1日1帳票が作成されるようにしていきます。なお、右上に以下の3つのボタンを配置しています。

・現在の状態を確認

デバイスの電源の状態と、その時の電力値をダイアログで見ることができます。

Gateway連携URL

http://xxx.xxx.xxx.xxx:3000/api/v1/getvalue/gosund_linkage?action=get_info

・電源ON

デバイスの電源をONにします。

Gateway連携URL

http://xxx.xxx.xxx.xxx:3000/api/v1/getvalue/gosund_linkage?action=turn_on

・電源OFF

デバイスの電源をOFFにします。

Gateway連携URL

http://xxx.xxx.xxx.xxx:3000/api/v1/getvalue/gosund_linkage?action=turn_off

Gateway連携プログラムの作成

以下のようなPythonプログラムを作成してみました。これをConMas Gatewayに配置します。

import sys
import json
import tinytuya
import traceback

class Tuya:
    """
    Tuya Cloud にアクセスするクラス
    """
    # Access ID
    access_id = "xxxxxxxxxxxxxxxxxxxx"
    # Access Secret
    access_secret = "xxxxxxxxxxxxxxxxxxxx"
    # Device ID
    device_id = "xxxxxxxxxxxxxxxxxxxx"
    cloud = None
    status = None

    @classmethod
    def connect(cls) -> None:
        cls.cloud = tinytuya.Cloud(
                     apiRegion="us", 
                     apiKey=cls.access_id, 
                     apiSecret=cls.access_secret)
    
    @classmethod
    def update_status(cls) -> dict:
        """
        情報取得
        """
        cls.status = cls.cloud.getstatus(cls.device_id)
    
    @classmethod
    def get_value(cls,key:str):
        return next((r["value"] for r in cls.status["result"] if key == r["code"]), None)

    @classmethod
    def turn_on(cls) -> bool:
        command = {
            "commands": [
                {"code": "switch_1", "value": True}
            ]
        }
        result = cls.cloud.sendcommand(cls.device_id,command)
        return result["result"]

    @classmethod
    def turn_off(cls) -> None:
        command = {
            "commands": [
                {"code": "switch_1", "value": False}
            ]
        }
        result = cls.cloud.sendcommand(cls.device_id,command)
        return result["result"]
    
    @classmethod
    def disconnect(cls) -> None:
        del cls.cloud

# パラメータ
param = json.loads(sys.stdin.readline())

action = param["data"][0]
"""
アクション

get_info : 情報取得
turn_on : 電源ON
turn_off : 電源OFF
"""

def out_message(msg:str) -> None:
    """
    メッセージ出力
    """
    print(json.dumps({"error":msg}))

def get_info() -> None:
    """
    情報取得
    """
    Tuya.update_status()
    power_status = Tuya.get_value("switch_1")
    watt = Tuya.get_value("cur_power")
    msg = "電源状態：{}\n".format("ON" if True == power_status else "OFF") +\
          "電力値　：{}W".format(watt * 0.1)
    out_message(msg)

def turn_on() -> None:
    """
    電源ON
    """
    if True == Tuya.turn_on():
        out_message("電源をONにしました。")
    else:
        out_message("電源操作に失敗しました。")

def turn_off() -> None:
    """
    電源OFF
    """
    if True == Tuya.turn_off():
        out_message("電源をOFFにしました。")
    else:
        out_message("電源操作に失敗しました。")

def main() -> None:
    """
    メイン処理
    """
    try:
        Tuya.connect()
        if "get_info" == action:
            get_info()
        elif "turn_on" == action:
            turn_on()
        elif "turn_off" == action:
            turn_off()
        else:
            out_message("アクションが異なります。")

    except Exception as e:
        msg = f"意図しない例外が発生しました。\n" +\
              f"{str(e)}\n" +\
              f"{traceback.format_exc()}"
        out_message(msg)

    finally:
        Tuya.disconnect()

main()

バッチプログラムの作成

1時間に1回実行されるバッチプログラムを作成します。

import datetime
import psycopg2
import psycopg2.sql as sql
from psycopg2.extras import DictCursor
import xml.etree.ElementTree as ET
import requests
import tinytuya
import traceback

# DB接続文字列
DB_CONNECTION_STRING = "host= port= dbname= user= password="
# 元定義ID
DEF_TOP_ORG = xxx
# ConMas WebAPI URL
CONMAS_WEBAPI_URL = "http://xxx.xxx.xxx.xxx/ConMasAPI/Rests/APIExecute.aspx"
# ConMas WebAPI ユーザー
CONMAS_WEBAPI_USER = "xxxxxxxxxx"
# ConMas WebAPI パスワード
CONMAS_WEBAPI_PASSWORD = "xxxxxxxxxx"

class MyException(Exception):
    """
    自作例外
    """
    pass

class PostgreSQL:
    """
    DB処理クラス
    """
    connection = None
    cursor = None

    @classmethod
    def open(cls) -> None:
        cls.connection = psycopg2.connect(DB_CONNECTION_STRING)
        cls.cursor = cls.connection.cursor(cursor_factory=DictCursor)

    @classmethod
    def get_rep_top_id(cls) -> list:
        """
        帳票ID取得
        """
        query = sql.SQL(r"""
            SELECT
                rep_top_id 
            FROM
                view_rep_cluster 
            WHERE
                rep_top_name ~ '^Gosundデータ連携_'
                AND rep_sheet_no = 1
                AND cluster_id = 0
                AND input_value = {input_value}
                AND public_status = 2 
                AND deleted = 0
        """).format(
            input_value = sql.Literal(datetime.datetime.now().strftime("%Y/%m/%d"))
        )
        cls.cursor.execute(query)
        return [dict(result) for result in cls.cursor.fetchall()]

    @classmethod
    def get_def_top_id(cls) -> list:
        """
        定義ID取得
        """
        query = sql.SQL(r"""
            SELECT
                def_top_id
            FROM
                view_def_current
            WHERE
                def_top_org = {def_top_org}
        """).format(
            def_top_org = sql.Literal(DEF_TOP_ORG)
        )
        cls.cursor.execute(query)
        return [dict(result) for result in cls.cursor.fetchall()]
    
    @classmethod
    def get_output_cluster(cls,rep_top_id:str) -> list:
        """
        クラスター取得
        """
        query = sql.SQL(r"""
            SELECT
                rep_sheet_no
                , cluster_id
                , cluster_name 
            FROM
                view_rep_cluster 
            WHERE
                rep_top_id = {rep_top_id} 
                AND cluster_name ~( 
                    SELECT
                        CONCAT( 
                            '_'
                            , MIN( 
                                CAST( 
                                    (REGEXP_MATCH(cluster_name, '\d+$')) [1] AS integer
                                )
                            ) 
                            , '$'
                        ) 
                    FROM
                        view_rep_cluster 
                    WHERE
                        rep_top_id = {rep_top_id} 
                        AND cluster_name ~ '^時刻_\d+$' 
                        AND input_value = ''
                )
        """).format(
            rep_top_id = sql.Literal(rep_top_id)
        )
        cls.cursor.execute(query)
        return [dict(result) for result in cls.cursor.fetchall()]

    @classmethod
    def close(cls) -> None:
        cls.cursor.close()
        cls.connection.close()

class ConMasWebApi:
    """
    ConMas WebAPI を実行するクラス
    """
    session = None

    @classmethod
    def login(cls) -> None:
        """
        ログイン
        """
        cls.session = requests.session()
        response = cls.session.post(CONMAS_WEBAPI_URL, params={
            "command": "Login",
            "user": CONMAS_WEBAPI_USER,
            "password": CONMAS_WEBAPI_PASSWORD
        })
        if "0" != ET.fromstring(response.text).findtext("./loginResult/code"):
            raise MyException("ConMas WebAPI のログインに失敗しました。")
    
    @classmethod
    def create_report(cls,def_top_id:str) -> str:
        """
        帳票作成
        """
        xml = f"""
            <conmas>
                <top>
                    <defTopId>{def_top_id}</defTopId>
                    <sheets>
                        <sheet>
                            <sheetNo>1</sheetNo>
                            <clusters>
                                <cluster>
                                    <sheetNo>1</sheetNo>
                                    <clusterId>0</clusterId>
                                    <value>{datetime.datetime.now().strftime("%Y/%m/%d")}</value>
                                </cluster>
                            </clusters>
                        </sheet>
                    </sheets>
                </top>
            </conmas>
        """
        response = cls.session.post(CONMAS_WEBAPI_URL, params={
            "command": "AutoGenerate",
            "type": "xml"
        }, files={"data": xml})
        rep_top_id = ET.fromstring(response.text).findtext("./results/result/topId")
        if None == rep_top_id:
            raise MyException("帳票作成に失敗しました。")
        return rep_top_id
    
    @classmethod
    def update_report(cls,rep_top_id:str,date_cluster:dict,watt_cluster:dict,watt:str) -> None:
        """
        帳票更新
        """
        xml = f"""
            <conmas>
                <top>
                    <repTopId>{rep_top_id}</repTopId>
                    <sheets>
                        <sheet>
                            <sheetNo>{date_cluster["rep_sheet_no"]}</sheetNo>
                            <clusters>
                                <cluster>
                                    <sheetNo>{date_cluster["rep_sheet_no"]}</sheetNo>
                                    <clusterId>{date_cluster["cluster_id"]}</clusterId>
                                    <value>{datetime.datetime.now().strftime("%H:%M")}</value>
                                </cluster>
                                <cluster>
                                    <sheetNo>{watt_cluster["rep_sheet_no"]}</sheetNo>
                                    <clusterId>{watt_cluster["cluster_id"]}</clusterId>
                                    <value>{watt}</value>
                                </cluster>
                            </clusters>
                        </sheet>
                    </sheets>
                </top>
            </conmas>
        """
        response = cls.session.post(CONMAS_WEBAPI_URL, params={
            "command": "UpdateReport",
            "type": "xml",
            "mode": "1"
        }, files={"data": xml})
        if "0" != ET.fromstring(response.text).findtext("./error/code"):
            raise MyException("帳票更新に失敗しました。")
        return
    
    @classmethod
    def logout(cls) -> None:
        """
        ログアウト
        """
        cls.session.post(CONMAS_WEBAPI_URL, params={
            "command": "Logout"
        })
        cls.session.close()

class Tuya:
    """
    Tuya Cloud にアクセスするクラス
    """
    # Access ID
    access_id = "xxxxxxxxxxxxxxxxxxxx"
    # Access Secret
    access_secret = "xxxxxxxxxxxxxxxxxxxx"
    # Device ID
    device_id = "xxxxxxxxxxxxxxxxxxxx"
    cloud = None
    status = None

    @classmethod
    def connect(cls) -> None:
        cls.cloud = tinytuya.Cloud(
                     apiRegion="us", 
                     apiKey=cls.access_id, 
                     apiSecret=cls.access_secret)
    
    @classmethod
    def update_status(cls) -> dict:
        """
        情報取得
        """
        cls.status = cls.cloud.getstatus(cls.device_id)
    
    @classmethod
    def get_value(cls,key:str):
        return next((r["value"] for r in cls.status["result"] if key == r["code"]), None)
    
    @classmethod
    def disconnect(cls) -> None:
        del cls.cloud

def get_rep_top_id() -> str:
    """
    今日の帳票IDを取得する
    """
    # DBから今日の帳票IDを取得する
    results = PostgreSQL.get_rep_top_id()

    if 0 == len(results):
        results = PostgreSQL.get_def_top_id()
        if 0 == len(results):
            raise MyException("帳票情報を取得できませんでした。")
        else:
            rep_top_id = ConMasWebApi.create_report(results[0]["def_top_id"])
    else:
        rep_top_id = results[0]["rep_top_id"]

    return rep_top_id

def get_output_cluster(rep_top_id:str) -> tuple:
    """
    値を入力するクラスターを取得
    """
    # DBから値を入力するクラスターを取得する
    results = PostgreSQL.get_output_cluster(rep_top_id)

    if 0 == len(results):
        raise MyException("クラスター情報を取得できませんでした。")
    else:
        date_cluster = next((r for r in results if "時刻" in r["cluster_name"]), None)
        watt_cluster = next((r for r in results if "電力" in r["cluster_name"]), None)

    return date_cluster, watt_cluster

def main() -> None:
    """
    メイン処理
    """
    try:
        PostgreSQL.open()
        ConMasWebApi.login()
        Tuya.connect()

        # 今日の帳票IDを取得する
        rep_top_id = get_rep_top_id()

        # 値を連携する帳票とクラスターの情報を取得
        date_cluster, watt_cluster = get_output_cluster(rep_top_id)

        # 電力値を取得
        Tuya.update_status()
        watt = Tuya.get_value("cur_power") * 0.1

        # 帳票更新
        ConMasWebApi.update_report(rep_top_id, date_cluster, watt_cluster, watt)

    except MyException as e:
        print(str(e))
    
    except Exception as e:
        print(str(e))
        print(traceback.format_exc())

    finally:
        PostgreSQL.close()
        ConMasWebApi.logout()
        Tuya.disconnect()

main()

タスクスケジューラにセット

詳しい説明は省きますが、以下のようにタスクスケジューラをセットします。

バッチプログラムが1時間毎に実行されるように設定を行いました。

ちなみに、python.exeではなくpythonw.exeで実行することで、コマンドプロンプトが表示されません。

動作確認

タスクスケジューラにセットした2日後の朝です。帳票を見てみましょう。

作成された帳票を見てみる

・11月28日

・11月29日

・11月30日

現在の状態を確認ボタン

「現在の状態を確認」ボタンを押下

→電源はONになっています。

電源ONボタン

「電源OFF」ボタンを押下

・再度「現在の状態を確認」ボタンを押下

→電源がOFFになりました。

電源OFFボタン

・「電源ON」ボタンを押下

・再度「現在の状態を確認」ボタンを押下

→電源がONになりました。

最後に

こういったスマートデバイスを使うことで、i-Reporterの知識さえあれば、データの測定&取得、機器の操作が簡単にできます。
今回紹介したGosundのスマートプラグ以外にも、Tuyaに対応した製品はたくさんあるようですし、 Tuya対応でなくともAPIが使用できるスマートデバイスを使えば何でもOKです！

ネクストビジョンでは、このようなi-Reporterの機能をフル活用したシステムの開発を行っています。
「i-Reporterだったら実現できそう！」なシステムは是非ネクストビジョンにお任せください。全力でサポートいたします！

注意事項
この記事をもとに作られた仕組みで不利益を被っても一切責任は持てません。

執筆者：株式会社ネクストビジョン　i-Reporterチーム

i-Reporterの導入支援を行うシステム開発会社株式会社ネクストビジョンのi-Reporterチームです。
当チームには2013年からお取引させていただいて以降、i-Reporterの機能に魅了され、
探求し続ける技術者が多数在籍しております。
本記事を通じてi-Reporterの機能と外部システム連携のアイデアや具体的実行方法など共有させて
いただけますと幸いです。導入支援も行っておりますのでお気軽にご相談くださいませ。
https://www.nextvision.co.jp/product/ireporter/introduction_support/

現場帳票研究所編集部

現場帳票研究所の編集部です！
当ブログは現場帳票電子化ソリューション「i-Reporter」の開発・販売を行う株式会社シムトップスが運営しております。
現場DXの推進に奮闘する皆様のお役に立てるよう、業界情報を定期的に配信致しますので、ぜひ御覧ください！

概要

Smart Life アプリの設定

Tuya IoT Platform の設定

Pythonでデータ取得するための準備

帳票を作成

Gateway連携プログラムの作成

バッチプログラムの作成

タスクスケジューラにセット

動作確認

最後に

執筆者：株式会社ネクストビジョン　i-Reporterチーム

カテゴリー

導入社数3,500社以上！
ペーパレスアプリでの
シェアNo.1

使い慣れたエクセル帳票から
そのまま移行できる
現場帳票の電子化システム

概要

Smart Life アプリの設定

Tuya IoT Platform の設定

Pythonでデータ取得するための準備

帳票を作成

Gateway連携プログラムの作成

バッチプログラムの作成

タスクスケジューラにセット

動作確認

最後に

執筆者：株式会社ネクストビジョン i-Reporterチーム

カテゴリー

人気記事

導入社数3,500社以上！ペーパレスアプリでのシェアNo.1

使い慣れたエクセル帳票からそのまま移行できる 現場帳票の電子化システム

執筆者：株式会社ネクストビジョン　i-Reporterチーム

導入社数3,500社以上！
ペーパレスアプリでの
シェアNo.1

使い慣れたエクセル帳票から
そのまま移行できる
現場帳票の電子化システム