目次
はじめまして。 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. クラウドプロジェクトを作成します。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image.png)
Data CenterはWestern America Data Centerを選択してください。
3. 作成したプロジェクトのAccess ID
とAccess Secret
をどこかにコピーしておきましょう。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-5-1024x501.png)
4. Smart Life のアカウントを紐づけます。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-1-1024x418.png)
5. すると、Smart Lifeで登録しているデバイスが追加されるのでDevice ID
をコピーしておき、Debug Device
を選択します。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-4-1024x498.png)
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を使っていきます。 このような簡単な帳票を作成してみました。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-6.png)
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
で実行することで、コマンドプロンプトが表示されません。
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-8.png)
動作確認
タスクスケジューラにセットした2日後の朝です。帳票を見てみましょう。
作成された帳票を見てみる
・11月28日
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-2-1024x823.png)
・11月29日
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-3-1024x823.png)
・11月30日
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-7-1024x823.png)
現在の状態を確認ボタン
「現在の状態を確認」ボタンを押下
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-9-1024x823.png)
→電源はONになっています。
電源ONボタン
「電源OFF」ボタンを押下
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-10-1024x823.png)
・再度「現在の状態を確認」ボタンを押下
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-11-1024x823.png)
→電源がOFFになりました。
電源OFFボタン
・「電源ON」ボタンを押下
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-12-1024x823.png)
・再度「現在の状態を確認」ボタンを押下
![](https://i-reporter.jp/wp-content/uploads/2023/12/image-13-1024x823.png)
→電源がONになりました。
最後に
こういったスマートデバイスを使うことで、i-Reporterの知識さえあれば、データの測定&取得、機器の操作が簡単にできます。
今回紹介したGosundのスマートプラグ以外にも、Tuyaに対応した製品はたくさんある ようですし、 Tuya対応でなくともAPIが使用できるスマートデバイスを使えば何でもOKです!
ネクストビジョンでは、このようなi-Reporterの機能をフル活用したシステムの開発を行っています。
「i-Reporterだったら実現できそう!」なシステムは是非ネクストビジョンにお任せください。全力でサポートいたします!
この記事をもとに作られた仕組みで不利益を被っても一切責任は持てません。
執筆者:株式会社ネクストビジョン i-Reporterチーム
![](https://i-reporter.jp/wp-content/uploads/2024/01/Nextvision_Logo_小.png)
i-Reporterの導入支援を行うシステム開発会社 株式会社ネクストビジョンのi-Reporterチームです。
当チームには2013年からお取引させていただいて以降、i-Reporterの機能に魅了され、
探求し続ける技術者が多数在籍しております。
本記事を通じてi-Reporterの機能と外部システム連携のアイデアや具体的実行方法など共有させて
いただけますと幸いです。導入支援も行っておりますのでお気軽にご相談くださいませ。
https://www.nextvision.co.jp/product/ireporter/introduction_support/
![](https://i-reporter.jp/wp-content/uploads/2023/05/アイレポちゃん1.png)
現場帳票研究所の編集部です!
当ブログは現場帳票電子化ソリューション「i-Reporter」の開発・販売を行う株式会社シムトップスが運営しております。
現場DXの推進に奮闘する皆様のお役に立てるよう、業界情報を定期的に配信致しますので、ぜひ御覧ください!