intdashの計測データ分析ボードを作成してみた

aptpod Advent Calendar 2024 12月13日の記事を担当します、ソリューションアーキテクトの門脇です。

今回はintdash を活用し、取得した時系列データの分析にフォーカスします。簡易的な例として計測データの傾向を可視化するための分析ボード構築手法を共有します。そして計測から分析可視化までのプロセスを自動化します。

今回の要件
概要構成
- intdashとは？
- intdash Motionとは？
分析ボードの構築プロセス
結果とメリット
おわりに

今回の要件

1週間の歩行イベントにおける、5名の歩行距離を可視化したい。
計測>分析>可視化のプロセスを、出来る限りシンプルな構成で自動化したい。

概要構成

概要構成を示します。計測はiPhoneにインストールしたindash Motion V2で行います。分析したデータの可視化ツール（BIツール）にはLookerStudioを使用します。

intdashとは？

intdashは、様々なIoTデータを低遅延に伝送するハイパフォーマンスなデータ伝送ミドルウェアです。産業用バスなどの秒間数百〜数千点にもなる大量データにも対応。独自開発のプロトコルにより、帯域の細いモバイル回線でも確実かつリアルタイムにデータ伝送が可能です。

intdash Motionとは？

intdash Motionは、本格的なIoTデバイス導入前のPoCやトライアル用途でご利用いただける、簡易データ収集用スマートフォンアプリです。スマートフォンに搭載された加速度センサーやジャイロスコープ、GPS、カメラやマイクといった各種センサーを利用して、簡易的な実験を行うことができます。

分析ボードの構築プロセス

今回のプロジェクトは以下の流れで進めます。

intdash APIを使ったデータ取得
データ分析
データアップロード
分析ボード作成

intdash APIを使ったデータ取得

分析には、intdash Motion V2アプリから送信されるGNSSデータを利用します。

intdash APIで計測データを取得します。

# 計測の取得
def get_measurements(edge_uuid: str, start: pd.Timestamp, end: pd.Timestamp) -> list[dict]:
    measurements = []
    page = 1
    while True:
        resp = get(
            path=f"/v1/projects/{PROJECT_UUID}/measurements",
            params={
                "edge_uuid": edge_uuid,
                "start": start.tz_convert('utc').strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                "end": end.tz_convert('utc').strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                "page": page,
            },
        )

        for meas in resp["items"]:
            measurements.append(meas)

        if resp["page"]["next"]:
            page += 1
        else:
            break

    return measurements

# 時系列データの取得
def get_data_points(meas_uuid: str, idq: list[str]) -> list[dict]:
    resp = requests.get(
        url=INTDASH_URL + f"/api/v1/projects/{PROJECT_UUID}/data",
        headers=HEADERS,
        params={
            "name": meas_uuid,
            "idq": idq,
            "time_format": "ns",
        },
        stream=True,
    )
    resp.raise_for_status()

    data_points = []
    received = ""
    for chunk in resp.iter_content(chunk_size=None, decode_unicode=True):
        received += chunk
        for line in received.split("\n"):
            try:
                decoded = json.loads(line)
                data_points.append(decoded)
            except ValueError:
                received = line
                break

    return data_points

# DataFrameに変換
def convert(data_point: dict) -> list[dict]:
    match data_point["data_name"]:
        case "0/gnss_speed":
            bin = base64.b64decode(data_point["data"]["d"])
            value = struct.unpack(">d", bin)[0]
            return [{
                "time": data_point["time"],
                "name": "GNSS_Speed",
                "value": value,
            }]
        case _:
            raise ValueError(f"Unsupported data_type: {data_point['data_type']}")

データ分析

次に、各計測ごとにデータ分析を行います。速度の分析値と、速度を基にした移動距離を計算しています。

def analyze_gnss_speed(df: pd.DataFrame) -> dict:
    # タイムスタンプをdatetime型に変換
    df.index = pd.to_datetime(df.index)

    # 開始時刻と終了時刻
    start_time = df.index.min()
    end_time = df.index.max()

    # 時間間隔を計算（秒単位から時間単位に変換）
    time_deltas = df.index.to_series().diff().dt.total_seconds()
    time_deltas_in_hours = time_deltas / 3600  # 時間単位に変換

    # 移動距離の計算
    distances = df["GNSS_Speed"] * time_deltas_in_hours  # km/h × 時間 = km
    total_distance = distances.sum()

    # スピードの分析値
    speed_stats = {
        "mean_speed": df["GNSS_Speed"].mean(),
        "median_speed": df["GNSS_Speed"].median(),
        "max_speed": df["GNSS_Speed"].max(),
        "min_speed": df["GNSS_Speed"].min(),
        "range_speed": df["GNSS_Speed"].max() - df["GNSS_Speed"].min(),
        "std_speed": df["GNSS_Speed"].std(),
        "time_of_peak_speed": df["GNSS_Speed"].idxmax(),
    }


    # 分析結果のまとめ
    analysis = {
        "start_time": start_time,
        "end_time": end_time,
        "total_distance": total_distance,
        **speed_stats,
    }
    return analysis