はじめに
本記事では、ラズパイ対応センサシリーズの加速度センサ（LSM6DSR搭載）について、接続台数ごとに連続計測可能な最大サンプリングレートを検証します。
ラズパイ対応センサシリーズは1台のセンサ用4chインターフェース基板に最大4台のセンサを接続できます。各センサとはI2Cマルチプレクサを介して通信し、I2C通信はFast-mode（最大400kHz）に対応しています。
加速度センサ（LSM6DSR搭載）自体は最大6667Hzでの計測が可能ですが、実際に連続計測できるサンプリングレートは、センサ単体の性能だけでなく、I2C通信を含む通信経路全体の影響を受けます。そこで、センサの接続台数を1～4台に変更し、それぞれの構成で連続計測可能な最大サンプリングレートを確認しました。
なお、本製品では加速度に加えて角速度も計測できますが、本記事では設備の異常監視等を想定して加速度のみを計測対象とし、検証を行いました。
連続計測とI2C通信速度
加速度センサ（LSM6DSR搭載）からデータを連続的に取得するため、LSM6DSRに搭載されたFIFO（First In, First Out）を使用します。計測データをFIFOに一時的に蓄積し、複数サンプルをまとめて読み出すことで、I2C通信のオーバーヘッドを抑えます。
FIFOに格納される1サンプルは、TAG（1byte）とX、Y、Z各軸の加速度データ（各2byte）の計7byteで構成されます。I2Cでは1byteの転送に9クロック必要なため、1サンプルの読み出しには少なくとも63クロック必要です。
そのため、例えば6667Hzで1台のみ接続した場合でも、データの読み出しには単純計算で約420kHz（6667Hz × 7byte × 9クロック）の通信速度が必要となり、FIFOデータそのものの転送だけでも、センサ用4chインターフェース基板が対応する最大400kHzを上回ります。
検証方法
LSM6DSRのFIFOには最大512サンプルのデータを格納できます。FIFOへのデータの蓄積に対して読み出しが十分に速ければ、FIFO蓄積量は一定の範囲内で増減します。一方、読み出しが追いつかない場合はFIFO蓄積量が徐々に増加し、FIFOの容量を超えるとオーバーランが発生します。
今回の検証では、接続する加速度センサの台数とサンプリングレートを変更し、それぞれの条件で60秒間の連続計測を行いました。LSM6DSRでは、加速度のサンプリングレートを12.5Hz、26Hz、52Hz、104Hz、208Hz、416Hz、833Hz、1667Hz、3333Hz、6667Hzから設定できます。
計測中は各加速度センサのFIFO蓄積量を一定間隔で記録し、その推移を確認しました。また、FIFOに128サンプル以上蓄積された時点で128サンプルずつ読み出しました。そのため、グラフのFIFO蓄積量はデータの蓄積と読み出しに伴って増減します。
なお、本検証では、60秒間の計測中にFIFOのオーバーランが発生せず、継続してデータを読み出せた場合を「連続計測可能」と判断しました。
使うもの
- Raspberry Pi（今回はRaspberry Pi 4 Model Bを使用）
- センサ用4chインターフェース基板
- 加速度センサ（LSM6DSR搭載）
- slee-Pi 3（必須ではありませんが、ラズパイおよび本センサーへの安定した電源供給のために併用しました。）
- ACアダプタ
- 開発用のPC、ラズパイへのログイン用のUSB-シリアル変換ケーブルなど
検証結果
1台接続時
まず、1台を6667Hzで動作させました。
FIFO蓄積量が上限の512サンプルに達した状態が継続したため、FIFOのオーバーランが発生し、連続計測できないことがわかりました。
次に、3333Hzで計測しました。
FIFO蓄積量は上限に達することなく一定の範囲内で増減しており、連続計測が可能であることを確認しました。
2台接続時
3333HzではFIFO蓄積量が上限に達しましたが、1667Hzでは上限に達することなく推移し、連続計測が可能であることを確認しました。
3台接続時
3333HzではFIFO蓄積量が上限に達しましたが、1667Hzでは上限に達することなく推移し、連続計測が可能であることを確認しました。
4台接続時
1667HzではFIFO蓄積量が上限に達しましたが、833Hzでは上限に達することなく推移し、連続計測が可能であることを確認しました。
まとめ
今回の検証では、加速度センサの接続数を1～4台に変更し、それぞれの構成で連続計測可能な最大サンプリングレートを確認しました。
その結果、1台接続時では3333Hz、2台および3台接続時では1667Hz、4台接続時では833Hzまで連続計測できることを確認しました。
| 接続数 | 連続計測可能な最大サンプリングレート |
|---|---|
| 1台 | 3333Hz |
| 2台 | 1667Hz |
| 3台 | 1667Hz |
| 4台 | 833Hz |
LSM6DSR自体は最大6667Hzのサンプリングレートに対応していますが、ラズパイ対応センサシリーズで連続計測できるサンプリングレートは、I2C通信を含めた通信経路全体の影響を受けます。そのため、接続するセンサ数や計測対象に応じてサンプリングレートを設定する必要があります。
一方、FIFOがオーバーランする前に計測を終了し、FIFO内に残ったデータを計測後に読み出すことで、今回確認した値よりも高いサンプリングレートで計測できる可能性があります。ただし、計測可能な時間は接続台数やサンプリングレートによって異なります。
なお、加速度センサ（LSM6DSR搭載）では加速度に加えて角速度も計測できます。加速度と角速度を同時に計測する場合は通信データ量が増加するため、連続計測可能なサンプリングレートは今回の検証結果とは異なります。
付録
サンプルコードを表示
import csv,time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from smbus2 import SMBus, i2c_msg
CHANNELS,ODR_HZ,DURATION_SEC,CHUNK_WORDS=[1,2,3,4],833,60.0,128
CHANNEL_BUSES = {
1: 22,
2: 23,
3: 24,
4: 25,
}
I2C_ADDRESS,WHO_AM_I,WHO_AM_I_VALUE=0x6A,0x0F,0x6B
CTRL1_XL,CTRL3_C=0x10,0x12
FIFO_CTRL1,FIFO_CTRL2,FIFO_CTRL3,FIFO_CTRL4=0x07,0x08,0x09,0x0A
FIFO_STATUS1,FIFO_DATA_OUT_TAG=0x3A,0x78
ACCEL_TAG,FIFO_MODE_BYPASS,FIFO_MODE_CONTINUOUS=0x02,0x00,0x06
FIFO_OVERRUN_FLAGS = (1 << 6) | (1 << 3)
FIFO_DEPTH_WORDS = 512
ODR_BITS={12.5:1,26:2,52:3,104:4,208:5,416:6,833:7,1667:8,3333:9,6667:10}
@dataclass
class Sensor:
channel: int; bus_no: int; bus: SMBus
samples: int=0; other_tags: int=0; max_fifo_level: int=0
overrun_observations: int=0; start_ns: int=0; stop_ns: int=0
def read_regs(self, register: int, length: int) -> bytes:
write = i2c_msg.write(I2C_ADDRESS, [register])
read = i2c_msg.read(I2C_ADDRESS, length)
self.bus.i2c_rdwr(write, read)
return bytes(read)
def write_reg(self, register: int, value: int) -> None:
self.bus.write_byte_data(I2C_ADDRESS, register, value)
def configure(self) -> None:
if self.bus.read_byte_data(I2C_ADDRESS, WHO_AM_I) != WHO_AM_I_VALUE:
raise RuntimeError(f"CH{self.channel}: WHO_AM_I mismatch")
self.write_reg(FIFO_CTRL4, FIFO_MODE_BYPASS)
time.sleep(0.002)
self.write_reg(CTRL3_C, 0x44)
self.write_reg(FIFO_CTRL1, 0x00)
self.write_reg(FIFO_CTRL2, 0x00)
odr = ODR_BITS[ODR_HZ]
self.write_reg(CTRL1_XL, odr << 4)
self.write_reg(FIFO_CTRL3, odr)
def start(self) -> None:
self.write_reg(FIFO_CTRL4, FIFO_MODE_CONTINUOUS)
self.start_ns = time.perf_counter_ns()
def stop(self) -> None:
self.write_reg(CTRL1_XL, 0x00)
self.stop_ns = time.perf_counter_ns()
def fifo_status(self) -> tuple[int, int]:
status1, status2 = self.read_regs(FIFO_STATUS1, 2)
level = ((status2 & 0x03) << 8) | status1
self.max_fifo_level = max(self.max_fifo_level, level)
if status2 & FIFO_OVERRUN_FLAGS:
self.overrun_observations += 1
return level, status2
def read_fifo(self, words: int) -> None:
raw = self.read_regs(FIFO_DATA_OUT_TAG, words * 7)
for offset in range(0, len(raw), 7):
tag = (raw[offset] >> 3) & 0x1F
if tag == ACCEL_TAG:
self.samples += 1
else:
self.other_tags += 1
def close(self) -> None:
self.bus.close()
def validate_settings() -> None:
if not CHANNELS or len(CHANNELS) > 4 or len(CHANNELS) != len(set(CHANNELS)):
raise SystemExit("CHANNELS must contain 1-4 unique channel numbers")
if any(channel not in CHANNEL_BUSES for channel in CHANNELS):
raise SystemExit("Every selected channel needs an entry in CHANNEL_BUSES")
if ODR_HZ not in ODR_BITS:
raise SystemExit(f"Unsupported ODR_HZ: {ODR_HZ}")
if not 1 <= CHUNK_WORDS <= FIFO_DEPTH_WORDS:
raise SystemExit(f"CHUNK_WORDS must be 1..{FIFO_DEPTH_WORDS}")
if DURATION_SEC <= 0:
raise SystemExit("DURATION_SEC must be greater than zero")
def open_sensors() -> list[Sensor]:
sensors: list[Sensor] = []
try:
for channel in CHANNELS:
bus_no = CHANNEL_BUSES[channel]
bus = SMBus(bus_no)
sensor = Sensor(channel, bus_no, bus)
sensors.append(sensor)
sensor.configure()
return sensors
except Exception:
for sensor in sensors:
sensor.close()
raise
def drain_fifos(sensors: list[Sensor]) -> None:
for sensor in sensors:
level, _ = sensor.fifo_status()
while level:
sensor.read_fifo(min(level, CHUNK_WORDS))
level, _ = sensor.fifo_status()
def save_results(
result_dir: Path,
sensors: list[Sensor],
snapshots: list[list],
) -> bool:
result_dir.mkdir(parents=True)
all_pass = all(
sensor.overrun_observations == 0 and sensor.other_tags == 0
for sensor in sensors
)
with (result_dir / "summary.csv").open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["channel","bus","odr_hz","active_time_s","accel_samples",
"effective_samples_per_s","max_fifo_level_words",
"overrun_observations","other_tag_words","result"])
for sensor in sensors:
active_s = (sensor.stop_ns - sensor.start_ns) / 1_000_000_000
passed = sensor.overrun_observations == 0 and sensor.other_tags == 0
writer.writerow([sensor.channel,sensor.bus_no,ODR_HZ,f"{active_s:.6f}",
sensor.samples,f"{sensor.samples/active_s:.3f}",
sensor.max_fifo_level,sensor.overrun_observations,
sensor.other_tags,"PASS" if passed else "FAIL"])
with (result_dir / "fifo_levels.csv").open(
"w", newline="", encoding="utf-8"
) as f:
writer = csv.writer(f)
writer.writerow(["elapsed_s"]+[f"ch{s.channel}_fifo_words" for s in sensors])
for snapshot in snapshots:
writer.writerow([f"{snapshot[0]:.6f}"]+snapshot[1:])
return all_pass
def main() -> int:
validate_settings()
sensors = open_sensors()
snapshots: list[list] = []
started = False
try:
for sensor in sensors:
sensor.start()
started = True
benchmark_start_ns = time.perf_counter_ns()
end_ns = benchmark_start_ns + int(DURATION_SEC * 1_000_000_000)
next_snapshot_ns = benchmark_start_ns
snapshot_interval_ns = 100_000_000
while time.perf_counter_ns() < end_ns:
did_read = False
now_ns = time.perf_counter_ns()
take_snapshot = now_ns >= next_snapshot_ns
snapshot = [(now_ns-benchmark_start_ns)/1_000_000_000] if take_snapshot else None
for sensor in sensors:
level, _ = sensor.fifo_status()
if take_snapshot:
snapshot.append(level)
if level >= CHUNK_WORDS:
sensor.read_fifo(CHUNK_WORDS)
did_read = True
if take_snapshot:
snapshots.append(snapshot)
while next_snapshot_ns <= now_ns:
next_snapshot_ns += snapshot_interval_ns
if not did_read:
time.sleep(0.0005)
for sensor in sensors:
sensor.stop()
started = False
drain_fifos(sensors)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
result_dir = Path("results") / (
f"{timestamp}_{len(sensors)}sensor_{ODR_HZ:g}Hz_chunk{CHUNK_WORDS}"
)
all_pass = save_results(result_dir, sensors, snapshots)
return 0 if all_pass else 2
finally:
for sensor in sensors:
try:
if started:
sensor.stop()
sensor.write_reg(FIFO_CTRL4, FIFO_MODE_BYPASS)
except OSError:
pass
sensor.close()
if __name__ == "__main__":
raise SystemExit(main())