SwitchBot温湿度データをPythonで1時間おきに自動取得してCSV保存【API活用】

この記事では、SwitchBotの温湿度計からクラウド経由で温度・湿度を取得し、Pythonを使って1時間ごとにCSVへ自動記録するスクリプトとの備忘録です。

📌 この方法の特徴

• SwitchBotのAPIを活用
• 複数デバイスを並列処理
• Power BIなどへの可視化にも対応
• 重複記録を防ぐロジックあり

💡 使用環境と準備

• SwitchBot温湿度計 + Hub
• SwitchBot APIトークン
• Python 3.8以上（Windows/Linux）
• Device IDとCSV保存フォルダ

💻 スクリプト全文（ダミー値含む）

以下のコードを使用して、1時間ごとに温湿度を取得し、重複排除付きのCSV出力を実現しています。

import csv
import time
import os
import shutil
from datetime import datetime
import multiprocessing

API_TOKEN = os.getenv("SWITCHBOT_API_TOKEN")

DEVICE_INFO = {
'AAAAAAAAAA': 'aaaa.csv',
'BBBBBBBBBB': 'bbbb.csv',
'CCCCCCCCCC': 'cccc.csv',
'DDDDDDDDDD': 'dddd.csv',
}

URL_TEMPLATE = 'https://api.switch-bot.com/v1.0/devices/{}/status'

HEADERS = {
'Authorization': API_TOKEN,
'Content-Type': 'application/json; charset=utf8'
}

SOURCE_FOLDER = './data'
DESTINATION_FOLDER = './data'

def get_temperature_humidity(device_id):
url = URL_TEMPLATE.format(device_id)
response = requests.get(url, headers=HEADERS)
data = response.json()
temperature = data['body']['temperature']
humidity = data['body']['humidity']
return temperature, humidity

def read_last_two_timestamps(filename):
last_two = []
if os.path.exists(filename):
with open(filename, 'r', newline='') as file:
reader = csv.reader(file)
next(reader)
for row in reader:
last_two.append(row[0])
return last_two[-2:]

def save_to_csv(device_id, temperature, humidity):
now = datetime.now().replace(minute=0, second=0, microsecond=0)
timestamp = now.strftime('%Y-%m-%d %H:%M:%S')
filename = f'temperature_humidity_data_{device_id}.csv'
file_path = os.path.join(SOURCE_FOLDER, filename)
file_exists = os.path.isfile(file_path)

with open(file_path, mode='a', newline='') as file:
writer = csv.writer(file)
if not file_exists:
writer.writerow(['Timestamp', 'Temperature (°C)', 'Humidity (%)'])
writer.writerow([timestamp, temperature, humidity])

copy_csv(filename, DEVICE_INFO[device_id])

last_two = read_last_two_timestamps(file_path)
if len(last_two) == 2 and last_two[0] == last_two[1]:
remove_last_row(file_path)

copied_file_path = os.path.join(DESTINATION_FOLDER, DEVICE_INFO[device_id])
last_two_copied = read_last_two_timestamps(copied_file_path)
if len(last_two_copied) == 2 and last_two_copied[0] == last_two_copied[1]:
remove_last_row(copied_file_path)

def copy_csv(source_filename, destination_filename):
if not os.path.exists(DESTINATION_FOLDER):
os.makedirs(DESTINATION_FOLDER)
shutil.copy(
os.path.join(SOURCE_FOLDER, source_filename),
os.path.join(DESTINATION_FOLDER, destination_filename)
)

def remove_last_row(filename):
with open(filename, 'r+', newline='') as file:
lines = file.readlines()
file.seek(0)
file.writelines(lines[:-1])
file.truncate()

def process_device(device_id):
while True:
try:
temperature, humidity = get_temperature_humidity(device_id)
save_to_csv(device_id, temperature, humidity)
print(f'{device_id}：{temperature}°C, {humidity}％ saved at {datetime.now()}')
except Exception as e:
print(f'エラー（{device_id}）：{e}')
time.sleep(3600)

def main():
processes = []
for device_id in DEVICE_INFO.keys():
p = multiprocessing.Process(target=process_device, args=(device_id,))
processes.append(p)
p.start()

for p in processes:
p.join()

if __name__ == '__main__':
main()

🔐 セキュリティ＆個人情報

下記の部分は、ご自身の環境に合わせて置き換えてください。

プレースホルダー	説明
'os.getenv("SWITCHBOT_API_TOKEN")'	SwitchBotアプリで取得したAPIトークン
'AAAAAAAAAA' など	各デバイスのDevice ID
'aaaa.csv'など	出力するCSVファイル名

📊 可視化や応用アイデア

• Power BI DesktopでCSVを読み込んでグラフ化
• Google Sheetsと連携してクラウド上で監視
• 家庭内の複数部屋の温湿度を一括ログ化
• ラズパイなどに組み込んで24時間稼働