OMRON製2JCIE-BL01からデータを取り出す

やっと涼しくなりました～
「暑さ寒さも彼岸まで」とはいいますが今年の異常な暑さもやっと終わりですかねぇ。

↑　https://ja.weatherspark.com/　より9月の気温の推移のグラフを拝借しました。これを見ればわかるように最低、最高気温ともに大幅に平年値を上回っていることがわかります。22日頃からやっと最低、最高気温が平年並みになってきました。

↑　自宅北側ベランダに設置している簡易気象観測器「WxBeacon2=OMRON製(2JCIE-BL01)」で観測した気温の推移です。

WxBeacon2 はオムロンの「2JCIE-BL01」同等品です

2JCIE-BL 環境センサ/特長 | オムロン制御機器

環境センサ身の回りの様々な環境情報をセンシング

残念ながら「2JCIE-BL01」は2023年4月に製造中止になっています。代替品もないようです。

WxBeacon2 | Weathernews Inc.

WxBeacon2 MISSION 気温・湿度・紫外線など6要素を観測するスマホ連携の気象センサー WxBea

WxBeacon2 のスペックでは気温の誤差は ±2℃ ということです。実際、わが家の WxBeacon2 で観測する気温のデータは気象庁の発表する神戸市のデータよりやや高い目になっています。これが立地条件によるものなのか誤差の範囲なのかは残念ながら校正したわけではないのでわかりません。

WxBeacon2 は Bluetooth で通信をしてるのですが、weathernews 製の純正スマホアプリでは上図のように最高で2週間しかデータを取り込むことができません。しかも、データはアプリ上でしか見ることができず、なんといってもデータの取得ができないというないというのは致命的で過去データを比較、蓄積するには不便な面がありました。

【気象観測】Windows＆PythonでWxBeacon2からデータ取得 - Qiita

概要数年前にウェザーニュースのリポート特典で，WxBeacon2という簡易気象観測機器をもらいました．観測データは，ウェザーニュースアプリの画面上で閲覧することができるようになっています．しかし観測データは14日前までしか見ることがで...

↑　そんな時に見つけたのが上記の記事です。WxBeacon2の生データを取り込んでCSVファイルにしてしまう、というものでまさに今回の目的にうってつけでした。

スクリプトは「Python」で書かれているのですが、私が普段使っている Jupyter Notebook で走らせると観測器を検出できないとか何点か細かいエラーが出てしまい、その修正にやや手こずりましたが、ChatGPTに助けを求めながらなんとか完成しました。(ちなみに Jupyter Notebook はノートパソコンではなくて Python の対話型開発環境のことです。)

完成したコードです

#Python
#WXBeacon2(2JCIE-BL01)から生データを取り出しCSVファイルにするスクリプト

import os
import datetime
import asyncio
import struct
from bleak import BleakScanner, BleakClient
from loguru import logger

UUID = {
    'latestData': '3001',
    'latestPage': '3002',
    'requestPage': '3003',
    'responseFlag': '3004',
    'responseData': '3005',
    'eventFlag': '3006',
    'measurementInterval': '3011'
}

def unix_to_jst(unix_time, return_str=False):
    '''Convert UNIX time to JST time'''
    jst_time = datetime.datetime.fromtimestamp(unix_time)
    if return_str:
        return jst_time.strftime("%Y/%m/%d %H:%M:%S")
    else:
        return jst_time

async def connect(device_identifier, datefrom, output):
    """device_identifier は BLEアドレスまたは名前を指定可能"""
    datefrom = datetime.datetime.strptime(datefrom, '%Y/%m/%d %H:%M:%S')
    break_flag = False

    # BLEデバイス探索
    devices = await BleakScanner.discover()
    device = None
    for d in devices:
        logger.debug(f'Found device: Address={d.address}, Name={d.name}')
        if d.address == device_identifier or d.name == device_identifier:
            device = d
            break

    if not device:
        raise Exception(f'Device "{device_identifier}" not found. Make sure it is powered on and advertising.')

    # デバイスに接続
    async with BleakClient(device) as client:
        logger.success('Connected!')

        # 最新ページ情報取得
        data = await client.read_gatt_char(f'0c4c{UUID["latestPage"]}-7700-46f4-aa96-d5e974e32a54')
        (latest_page_time, interval, latest_page, latest_row) = struct.unpack('<IHHB', data)
        latest_page_time = unix_to_jst(latest_page_time + interval * latest_row)
        logger.info(f'Latest page: {latest_page}, Latest row: {latest_row}, Latest page time: {latest_page_time}')

        # データ取得ループ
        for page in reversed(range(latest_page + 1)):
            # ページリクエスト
            set_value = struct.pack('<HB', page, latest_row if page == latest_page else 12)
            await client.write_gatt_char(f'0c4c{UUID["requestPage"]}-7700-46f4-aa96-d5e974e32a54', set_value)

            # フラグ確認
            flag, retry = 0, 0
            while flag != 1:
                response_flag = await client.read_gatt_char(f'0c4c{UUID["responseFlag"]}-7700-46f4-aa96-d5e974e32a54')
                (flag, start_time) = struct.unpack('<BI', response_flag)
                retry += 1
                if retry > 1:
                    logger.warning(f'Retrying request page (Page:{page}, Status:{flag})')

            # ページデータ取得
            for row in reversed(range(latest_row+1 if page == latest_page else 12+1)):
                if unix_to_jst(start_time + interval * row) < datefrom:
                    break_flag = True
                    break
                page_value = await client.read_gatt_char(f'0c4c{UUID["responseData"]}-7700-46f4-aa96-d5e974e32a54')
                (row, temp, hum, light, uv, press, noise, discom, heat, batt) = struct.unpack('<BhhhhhhhhH', page_value)
                time_str = unix_to_jst(start_time + interval * row, return_str=True)
                temp /= 100
                hum /= 100
                uv /= 100
                press /= 10
                noise /= 100
                discom /= 100
                heat /= 100
                batt /= 1000

                # CSVに書き込み
                mode = 'a' if os.path.exists(output) else 'w'
                with open(output, mode) as f:
                    f.write(f'{time_str},{temp},{hum},{light},{uv},{press},{noise},{discom},{heat},{batt}\n')

            if break_flag:
                break


# ========== Notebook用の設定 ==========
device_address = "E1:2E:A8:4A:BC:77"   # 確認したセンサーのBLEアドレス(環境に依存します)
datefrom = "2025/09/01 00:00:00"       # 取得開始日時　
output = "sensor_data.csv"             # 出力CSV

if os.path.exists(output):
    print(f'WARNING: "{output}" already exists. Data will be appended.')

# Notebookで実行
await connect(device_address, datefrom, output)