ここ数年「デジタルツイン」というワードを聞く機会が増えましたが、自宅で試してみたい！という方も多いのではないでしょうか。
デジタルツインとは、現実世界のモノやコトをデジタル空間に再現し、リアルタイムに同期させる技術です。例えば工場の生産ラインの状態を監視したり、ビルの電力使用量を可視化したりといった使い方があります。
Autodesk Tandemは、そんなデジタルツインを簡単に実現できるクラウドサービスです。CADソフトで有名なAutodesk社が開発しており、BIMモデル（建物の3Dモデル）にセンサーデータを紐づけリアルタイムに可視化できます。
本記事では、Autodesk TandemとSwitchBot温湿度計を使ってお手軽にデジタルツインに入門してみます。
必要なもの
- SwitchBot温湿度計：1980円
- Autodesk Tandem：無料
- Autodesk Revit（体験版or学生版）：無料
- PC or ラズパイ
- 基礎的なPythonの知識
1. モデルの作成
まずは、Autodesk社のBIMソフト「Revit」でBIMモデルを作成します。Revitの操作方法は公式ドキュメントに詳しいので省略しますが、とりあえず床、壁、部屋、温湿度計さえあれば大丈夫です。温湿度計は分かりやすいように大きなピンの形でモデリングしています。
筆者の自宅は狭いので🥲代わりに研究室のモデルを使用することにします。
2. モデルのアップロード
TandemでFacilityを新規作成し、モデルをアップロードします。
"SwitchBot温湿度計1"Connectionを作成して、温湿度計オブジェクトに紐づけます。"🔗"ボタンをクリックしてStream URLを取得します。
以下のように、Stream URLにはAPI SECRETとMODEL URNとSTREAM IDが含まれています。
Stream URLのフォーマット
https://:{API_SECRET}@tandem.autodesk.com/api/v1/timeseries/models/{MODEL_URN}/streams/{STREAM_ID}
3. 温湿度データの取得
SwitchBot温湿度計からBLE通信で温湿度データを取得します。SwitchBotはスマートフォンアプリと通信するために、2秒毎にアドバタイズパケットを送信しています。このパケットのマニュファクチャーデータに温度と湿度の情報が格納されています。
まずはAnacondaなどでPythonの実行環境を用意します。
conda create -n tandem-ble python=3.9
conda activate tandem-ble
pip install bleak requests
次に以下のコードを実行して、温湿度データを取得します。
Tandem.py
from bleak import BleakScanner
from datetime import datetime
import signal
import sys
import requests
import base64
import time
# BLE設定
SWITCHBOT_COMPANY_ID = 0x0969 # SwitchBotのCompanyID
TARGET_MAC_ADDRESS = "XX:XX:XX:XX:XX:XX" # ← ここに温湿度計のMACアドレスを入力
# グローバル変数
is_running = True
def parse_temperature(manufacturer_data):
"""温度データの解析"""
is_negative = (manufacturer_data[9] & 0x80) == 0
integer_part = manufacturer_data[9] & 0x7F
decimal_part = manufacturer_data[8] & 0x0F
temperature = integer_part + (decimal_part / 10)
return -temperature if is_negative else temperature
def parse_humidity(manufacturer_data):
"""湿度データの解析"""
return manufacturer_data[10] & 0x7F
def handle_sigint(signum, frame):
"""Ctrl+C のハンドラ"""
global is_running
print("\nStopping scanner...")
is_running = False
def detection_callback(device, advertisement_data):
"""デバイス検出時のコールバック関数"""
if device.address == TARGET_MAC_ADDRESS:
if advertisement_data.manufacturer_data:
for company_id, data in advertisement_data.manufacturer_data.items():
if company_id == SWITCHBOT_COMPANY_ID:
data_bytes = bytes(data)
try:
temperature = parse_temperature(data_bytes)
humidity = parse_humidity(data_bytes)
current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"\n[{current_datetime}] Device detected:")
print(f"Temperature: {temperature:.1f}°C")
print(f"Humidity: {humidity}%")
except Exception as e:
print(f"Error parsing data: {e}")
async def scan_devices():
print("Starting continuous scan for SwitchBot temperature/humidity meter...")
print(f"Target MAC Address: {TARGET_MAC_ADDRESS}")
print("Press Ctrl+C to stop scanning")
try:
scanner = BleakScanner(detection_callback)
await scanner.start()
while is_running:
await asyncio.sleep(1.0)
await scanner.stop()
except Exception as e:
print(f"Error during scanning: {str(e)}")
finally:
print("\nScanner stopped.")
def main():
signal.signal(signal.SIGINT, handle_sigint)
print(f"Scan started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
try:
asyncio.run(scan_devices())
except KeyboardInterrupt:
pass
print(f"\nScan completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if __name__ == "__main__":
main()
コードを実行すると温湿度データが表示されます。
(tandem-ble) C:\Python-Tandem-Tutorial>python Tandem.py
Scan started at: 2025-02-21 18:27:41
Starting continuous scan for SwitchBot Meter...
Target MAC Address: XX:XX:XX:XX:XX:XX
Press Ctrl+C to stop scanning
[2025-02-21 18:27:42] Device detected:
Temperature: 20.5°C
Humidity: 25%
[2025-02-21 18:27:44] Device detected:
Temperature: 20.5°C
Humidity: 25%
4. Tandemへのデータ送信
続いてTandemへのデータ送信機能を追加し、1分間隔で温湿度を送信するようにします。
Tandem.py
def handle_sigint(signum, frame):
"""Ctrl+C のハンドラ"""
global is_running
print("\nStopping scanner...")
is_running = False
#===ここまでは変更なし===
+ # Tandem設定
+ MODEL_ID = 'your_model_urn' # ← ここにMODEL URNを入力
+ STREAM_ID = 'your_stream_id' # ← ここにSTREAM IDを入力
+ API_SECRET = 'your_api_secret' # ← ここにAPI SECRETを入力
# グローバル変数
is_running = True
+ last_sent_time = 0 # 最後にデータを送信した時刻
+ SEND_INTERVAL = 60 # 送信間隔（秒）
+ def send_to_tandem(temperature: float, humidity: float):
+ """Tandemにデータを送信する"""
+ try:
+ send_data = {
+ "temperature": temperature,
+ "humidity": humidity,
+ "timestamp": int(time.time())
+ }
+ url = f"https://tandem.autodesk.com/api/v1/timeseries/models/{MODEL_ID}/streams/{STREAM_ID}"
+ auth_string = f":{API_SECRET}"
+ auth_bytes = auth_string.encode('ascii')
+ base64_auth = base64.b64encode(auth_bytes).decode('ascii')
+ headers = {
+ 'Content-Type': 'application/json',
+ 'Authorization': f'Basic {base64_auth}'
+ }
+ response = requests.post(url, json=send_data, headers=headers)
+ response.raise_for_status()
+ print(f"Data sent successfully to Tandem at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+ print(f"Temperature: {temperature}°C, Humidity: {humidity}%")
+ return True
+ except requests.exceptions.RequestException as e:
+ print(f"Error sending data to Tandem: {e}")
+ if hasattr(e, 'response'):
+ print(f"Response status: {e.response.status_code}")
+ print(f"Response body: {e.response.text}")
+ return False
def detection_callback(device, advertisement_data):
"""デバイス検出時のコールバック関数"""
+ global last_sent_time # 最後の送信時刻を参照
+ current_time = time.time() # 現在時刻を取得
if device.address == TARGET_MAC_ADDRESS:
if advertisement_data.manufacturer_data:
for company_id, data in advertisement_data.manufacturer_data.items():
if company_id == SWITCHBOT_COMPANY_ID:
data_bytes = bytes(data)
try:
temperature = parse_temperature(data_bytes)
humidity = parse_humidity(data_bytes)
current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"\n[{current_datetime}] Device detected:")
print(f"Temperature: {temperature:.1f}°C")
print(f"Humidity: {humidity}%")
+ if current_time - last_sent_time >= SEND_INTERVAL:
+ if send_to_tandem(temperature, humidity):
+ last_sent_time = current_time
except Exception as e:
print(f"Error parsing data: {e}")
async def scan_devices():
print("Starting continuous scan for SwitchBot temperature/humidity meter...")
print(f"Target MAC Address: {TARGET_MAC_ADDRESS}")
print("Press Ctrl+C to stop scanning")
+ print(f"Data will be sent to Tandem every {SEND_INTERVAL} seconds")
try:
scanner = BleakScanner(detection_callback)
await scanner.start()
while is_running:
await asyncio.sleep(1.0)
await scanner.stop()
except Exception as e:
print(f"Error during scanning: {str(e)}")
finally:
print("\nScanner stopped.")
#===以下は変更なし===
コードを実行すると温湿度データがTandemへ送信されます。以下のように404エラーが表示されますが、現段階では問題ありません。
(tandem-ble) C:\Python-Tandem-Tutorial>python Tandem.py
Scan started at: 2025-02-22 01:15:07
Starting continuous scan for SwitchBot temperature/humidity meter...
Target MAC Address: D0:32:34:35:43:13
Press Ctrl+C to stop scanning
Data will be sent to Tandem every 60 seconds
[2025-02-22 01:15:15] Device detected:
Temperature: 22.2°C
Humidity: 24%
Error sending data to Tandem: 404 Client Error: Not Found for url: https://tandem.autodesk.com/api/v1/timeseries/models/urn:adsk.dtm:e8RsORY6SGGJ4BMtsYgRyw/streams/AQAAADYT4jmhnUV3jzkLMsiCpY0AAAAA
Response status: 404
Response body: {
"title":"Not Found",
"detail": "Stream Payload Received - Proceed to map/configure Parameters to initialize event storage"
}
Tandemに送信されたJSONの"temperature"と"humidity"をそれぞれ"温度"と"湿度"に割り当てましょう。そしてもう一度コードを実行すると、今度は404エラーが消えているはずです。
6. データ可視化
Tandemに蓄積されたデータを可視化してみましょう。
Tandemでは、「Stream」からグラフを表示することができます。「Combine XX with other stream」から複数のグラフを重ねて表示することも可能です。
また、「Select heatmap」からモデル上でヒートマップ表示したり、過去～現在のアニメーションを再生することもできます。
※ ヒートマップ表示を試す際は複数の温湿度計を使うと分かりやすいです。
さらに、Tandemにはカッコいいダッシュボードを作成する機能もあります。
参考記事
SwitchBotハブを使って長期的にデータ送信したい方は、こちらのkeiwatanabeさんの記事がオススメです。
また、Tandemについてより詳しく知りたい方はAPIドキュメントやDeveloper Community Blogをご確認ください。