SensorConnectのCSVをPythonで確認する｜時刻・空欄・欠損候補の品質検査

SensorConnectからCSVを書き出せても、そのままFFTや統計処理へ進む前に、時刻、列、空欄、数値、サンプル間隔を確認しておくとデータ処理の誤りを減らせます。

特に複数レートのデータでは、低レートの項目がすべての時刻行に存在しないことがあります。その空欄を自動的に「通信欠損」や「0」と判断しないことが重要です。

このページでは、Python標準ライブラリだけでCSVを読み取り専用検査するサンプルを紹介します。

このページでできること

• CSVの列名を確認する
• 時刻の重複・逆行を検出する
• NaN / inf / 数値変換できないセルを検出する
• チャンネルごとの空欄数を確認する
• 既知のサンプルレートから長い時間間隔を抽出する
• 複数レートによる正常な空欄と欠損候補を分けて考える

---

1. まずCSVの列名と時刻形式を確認する

最初にCSVをテキストエディタや表計算ソフトで開き、次を確認します。

• ヘッダー行の列名
• 時刻列の名前
• 時刻の単位：秒 / msなど
• 小数点記号
• 区切り文字
• 確認したいデータ列

このサンプルは、数値で表された時刻列を対象にします。

日時文字列を直接解析するスクリプトではありません。

元CSVは変更せず、必要なら複製したファイルで確認してください。

---

2. 複数レートでは空欄が正常な場合がある

例えば、加速度100 Hzと角速度50 Hzを同じ時間軸へ並べた概念例を考えます。

time_s,accel_x,gyro_z
0.00,1,2
0.01,1,
0.02,1,2
0.03,1,
0.04,1,2

gyro_zには空欄がありますが、有効値は0.02秒間隔なので50 Hzと整合します。

この場合、空欄があるという事実だけでは通信欠損とは判断できません。

GQ7のMultiple Rate Dataでも、異なる出力レートのFieldは共通クロック上でスケジュールされ、低レートFieldが毎パケットに含まれるわけではないことが説明されています。

---

3. 検査用スクリプト

以下のスクリプトは読み取り専用です。

行わない処理も明確にしています。

• ゼロ埋めしない
• 補間しない
• リサンプリングしない
• 校正しない
• FFTしない

まず異常候補を抽出し、その原因は元データや計測条件と照合して判断します。

# -*- coding: utf-8 -*-
"""
MicroStrain CSV Quality Check
Version: 1.1.0
変更内容:
- 複数レートCSVの空欄を通信欠損と分離して集計
- 時刻の重複・逆行・長い間隔を列ごとに確認
- NaN/inf、列不足、余分な列を検出
- 補間・ゼロ埋め・再サンプリングは行わない読み取り専用検査

標準ライブラリのみを使用します。
"""

import argparse
import csv
import json
import math
import statistics


def number(text, decimal_comma=False):
    """文字列を有限のfloatへ変換する。"""
    value = float(text.strip().replace(',', '.') if decimal_comma else text.strip())
    if not math.isfinite(value):
        raise ValueError('non-finite number')
    return value


def inspect_csv(path, time_column, value_columns, rates=None, delimiter=',',
                encoding='utf-8-sig', time_scale=1.0, decimal_comma=False,
                gap_factor=1.5):
    """数値時刻を持つCSVを読み取り専用で検査する。"""
    rates = rates or {}

    if not value_columns or len(value_columns) != len(set(value_columns)):
        raise ValueError('Value columns must be present and unique')
    if len(delimiter) != 1 or not math.isfinite(time_scale) or time_scale <= 0:
        raise ValueError('Use one delimiter character and a positive finite time scale')
    if not math.isfinite(gap_factor) or gap_factor <= 1:
        raise ValueError('Gap factor must be finite and greater than one')
    if any(key not in value_columns or not math.isfinite(rate) or rate <= 0
           for key, rate in rates.items()):
        raise ValueError('Each rate must be positive, finite and match a selected value column')

    samples = {name: [] for name in value_columns}
    counts = {
        name: {'blank_cells': 0, 'invalid_values': 0, 'examples': []}
        for name in value_columns
    }
    result = {
        'rows': 0,
        'invalid_time_rows': 0,
        'invalid_time_examples': [],
        'extra_column_rows': 0,
        'short_column_rows': 0,
        'duplicate_times': 0,
        'backward_times': 0,
        'channels': {},
    }

    previous = None
    seen_times = set()

    with open(path, encoding=encoding, newline='') as stream:
        reader = csv.DictReader(stream, delimiter=delimiter, strict=True)
        headers = reader.fieldnames or []

        if len(headers) != len(set(headers)):
            raise ValueError('Duplicate header names: rename columns before inspection')

        absent = set([time_column] + value_columns) - set(headers)
        if absent:
            raise ValueError('Missing columns: ' + ', '.join(sorted(absent)))

        for row in reader:
            result['rows'] += 1

            if None in row:
                result['extra_column_rows'] += 1
            if any(row.get(name) is None for name in headers):
                result['short_column_rows'] += 1

            try:
                timestamp = number(row.get(time_column) or '', decimal_comma) * time_scale
                if not math.isfinite(timestamp):
                    raise ValueError('non-finite scaled timestamp')
            except ValueError:
                result['invalid_time_rows'] += 1
                if len(result['invalid_time_examples']) < 5:
                    result['invalid_time_examples'].append(reader.line_num)
                continue

            result['duplicate_times'] += int(timestamp in seen_times)
            seen_times.add(timestamp)

            if previous is not None:
                result['backward_times'] += int(timestamp < previous)
            previous = timestamp

            for name in value_columns:
                text = row.get(name) or ''
                if not text.strip():
                    counts[name]['blank_cells'] += 1
                    continue

                try:
                    value = number(text, decimal_comma)
                except ValueError:
                    counts[name]['invalid_values'] += 1
                    if len(counts[name]['examples']) < 5:
                        counts[name]['examples'].append(reader.line_num)
                    continue

                samples[name].append((timestamp, value))

    for name, points in samples.items():
        steps = [(b[0] - a[0], b[0]) for a, b in zip(points, points[1:])]
        positive = [dt for dt, _ in steps if dt > 0]
        median = statistics.median(positive) if positive else None
        nominal = 1.0 / rates[name] if name in rates else median
        long_steps = [
            (dt, end_time)
            for dt, end_time in steps
            if nominal and dt > gap_factor * nominal
        ]

        result['channels'][name] = {
            **counts[name],
            'samples': len(points),
            'median_positive_interval_s': median,
            'reference_interval_s': nominal,
            'reference_source': 'specified_rate' if name in rates else 'median_estimate',
            'long_intervals': len(long_steps),
            'long_interval_examples': [
                {'end_time_s': end_time, 'interval_s': dt}
                for dt, end_time in long_steps[:5]
            ],
            'non_increasing_intervals': sum(dt <= 0 for dt, _ in steps),
            'minimum': min((value for _, value in points), default=None),
            'maximum': max((value for _, value in points), default=None),
        }

    result['notes'] = [
        'Blank cells are counted separately; they are not automatically communication loss.',
        'Channel counts include only rows with valid numeric timestamps.',
        'Long intervals require review; this program does not determine their cause.',
        'Without a specified rate, the reference interval is an estimate, not a guarantee.',
        'No resampling, interpolation, zero filling, calibration or FFT is performed.',
    ]
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('csv_path')
    parser.add_argument('--time', required=True, help='Column holding numeric time')
    parser.add_argument('--value', action='append', required=True,
                        help='Repeat for each value column')
    parser.add_argument('--rate', action='append', default=[],
                        help='Column=Hz; repeat for known rates')
    parser.add_argument('--delimiter', default=',')
    parser.add_argument('--encoding', default='utf-8-sig')
    parser.add_argument('--time-scale', type=float, default=1.0,
                        help='Multiply time by this to get seconds')
    parser.add_argument('--decimal-comma', action='store_true')
    parser.add_argument('--gap-factor', type=float, default=1.5)
    args = parser.parse_args()

    try:
        rates = {}
        for item in args.rate:
            key, rate = item.rsplit('=', 1)
            if key in rates:
                raise ValueError('Rate specified twice: ' + key)
            rates[key] = float(rate)

        result = inspect_csv(
            args.csv_path,
            args.time,
            args.value,
            rates=rates,
            delimiter=args.delimiter,
            encoding=args.encoding,
            time_scale=args.time_scale,
            decimal_comma=args.decimal_comma,
            gap_factor=args.gap_factor,
        )
    except (OSError, UnicodeError, csv.Error, ValueError) as error:
        parser.error(str(error))

    print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False))


if __name__ == '__main__':
    main()

同じコードをcode/microstrain_csv_check_v1_1_0.pyとしてワーキングパッケージにも収録しています。

---

4. 実行例

例えば、時刻列がtime_s、確認する列がaccel_xとgyro_zで、それぞれ100 Hzと50 Hzだとします。

コマンドラインでは次のように実行できます。

python microstrain_csv_check_v1_1_0.py sample.csv --time time_s --value accel_x --rate accel_x=100 --value gyro_z --rate gyro_z=50

SpyderなどのIDEで使う場合は、inspect_csv()を直接呼び出しても構いません。

result = inspect_csv(
    "sample.csv",
    time_column="time_s",
    value_columns=["accel_x", "gyro_z"],
    rates={"accel_x": 100, "gyro_z": 50},
)
print(json.dumps(result, ensure_ascii=False, indent=2))

---

5. 結果の読み方

blank_cells

その列が空欄だった行数です。

空欄数＝通信欠損数ではありません。

予定レートと実際の有効サンプル間隔を合わせて確認します。

invalid_values

数値へ変換できない値、NaN、infなどの数です。

意図したステータス文字列などを数値列として指定していないかも確認します。

duplicate_times

同じTimestampが複数行に存在する数です。

複数Fieldを表形式へ展開した結果なのか、実際の重複なのかを元データと照合します。

backward_times

Timestampが前の行より小さくなった箇所です。

ファイル結合、ソート、時刻のロールオーバーなども候補になるため、即座に通信異常とは判断しません。

median_positive_interval_s

その列に実際に数値が存在するサンプル間の代表的な時間間隔です。

50 Hzなら約0.02 s、100 Hzなら約0.01 sが目安になります。

long_intervals

指定した基準間隔より大きく空いた区間の候補です。

デフォルトでは基準間隔の1.5倍を超える箇所を抽出します。

原因は、

• 本当の欠損
• 出力設定
• データ選択範囲
• 機器状態
• CSV変換

などを別途確認します。

---

6. レートが分からない場合

--rateを指定しない場合、スクリプトは有効な正の時間差の中央値を基準間隔として使用します。

これはデータ自身から求めた推定値です。

設定したサンプルレートを保証する値ではないため、SensorConnectやMIP設定が分かる場合は明示的にレートを指定する方が判断しやすくなります。

---

7. 空欄をゼロ埋めする前に確認する

測定していない値と、測定結果が0だった値は意味が異なります。

CSVの空欄を一括で0にすると、本来存在しないサンプルを「実測値0」としてFFTやRMSへ入力してしまいます。

補間する場合も、

• どの列を補間したか
• どの方法を使ったか
• 最大何サンプルまで補間したか
• 元データを保持しているか

を記録します。

---

8. 長時間CSVでは対象区間を絞る

このサンプルは分かりやすさを優先し、選択したチャンネルのデータをメモリ上へ保持します。

非常に大きいCSVを調べる場合は、元データを残したまま、問題が疑われる時間範囲を複製して確認すると扱いやすくなります。

大規模データの常用検査へ発展させる場合は、ストリーミング集計やチャンク処理へ変更できます。

---

参考資料

• HBK SensorConnect
  https://www.hbkworld.com/en/products/software/inertial-sensor-software/sensorconnect
• GQ7 Multiple Rate Data
  https://files.microstrain.com/GQ7%20User%20Manual/dcp_content/introduction/Multiple%20Rate%20Data.htm
• Python csv
  https://docs.python.org/3/library/csv.html
• Python statistics
  https://docs.python.org/3/library/statistics.html

既存の関連ページもあわせて確認

既存の技術解説・ソフトウェア説明と相互に参照できるよう、本番反映時に現在のURLへ接続します。

• SensorConnectとは？MicroStrain Wireless・Inertial計測ソフト
• 無線計測データを複数段で保存する