私は自動車業界で働くソフトウェアエンジニアとして、自動運転車、データ測定技術、分析手法に強い関心を持っています。この投稿では、カスタム構築の測定システムについて説明し、そのプロセスを一から詳しく説明し、いくつかの実験結果を示します。私のデータロガーは、Raspberry Pi 3、BNO055 センサー、ELM327 OBD-II アダプターで構成されており、それぞれ計算、加速度と回転速度データの収集、実際の車両からのエンジン情報の取得を担当します。
1.背景
2.アーキテクチャ
3.センサー情報
4.座標系情報
5.データ比較
6.コアロジック
7.結論
8.参考
近年、IoT、人工知能、エッジ コンピューティングの台頭により、自動車分野を含むさまざまな業界に大きな進歩がもたらされました。現代の車両には多数のセンサーが装備されており、それらが連携して高度なシステムを構築し、強化された安全機能から自動運転機能まであらゆるものを可能にしています。
通常、車両の状態を監視するには、正確で信頼性の高い測定を実現するために、高度で高価なセンサーを使用する必要があります。ただし、車両のダイナミクスを一般的に理解するためにサーキット上のデータを取得するなど、特定の状況では、これらのハイエンド センサーが必ずしも必要ではない場合があります。これを認識し、私はより手頃な価格のコンポーネントを使用して車両の物理的状態を測定できる費用対効果の高いシステムを作成することにしました。
私は乗り心地よりも車両運動に関するデータの測定に重点を置くことにしました。これは、乗り心地を解析するには、より多くの高周波信号を適切な精度で捕捉する必要があり、測定時のサンプリング周波数を高く設定する必要があるためです。
具体的には、操舵角に対する横加速度の周波数応答特性において、車速にもよるが、一般に5Hz以下に共振周波数が現れる。一方、乗り心地解析では数十Hzの範囲にまで議論が及ぶことが多い
。したがって、測定システム開発の主な目標は、車両ダイナミクス領域でのデータ分析を容易にする低コストのデータロギングシステムを作成することです。
測定システムは次のハードウェアで構成されています:
Raspberry Pi はメインコンピューターとして機能し、BNO055 と ELM327 からデータを収集します。 Raspberry Pi は、I2C 経由で BNO055 と通信し、Bluetooth 経由で ELM327 と通信します (図 1 を参照)。
BNO055 センサーは、さまざまな種類のモーションとデータを測定できます:★Adafruit BNO055 概要[1]
絶対方位 (オイラーベクトル、100Hz): 360° 球面に基づいた 3 軸方位データを提供します。
絶対方向 (クォータニオン、100Hz): より正確なデータ操作のための 4 ポイント クォータニオン出力を提供します。
角速度ベクトル (100Hz): 3 つの軸にわたる回転速度をラジアン/秒 (rad/s) で測定します。
加速度ベクトル (100Hz): 3 つの軸にわたる重力と直線運動を含む加速度をメートル/秒の 2 乗 (m/s²) でキャプチャします。
磁場強度ベクトル (20Hz): 3 軸にわたる磁場強度をマイクロテスラ (μT) 単位で感知します。
線形加速度ベクトル (100Hz): 3 軸にわたる線形加速度データ (重力を除く) をメートル/秒 2 乗 (m/s²) で記録します。
重力ベクトル (100Hz): 3 軸にわたる重力加速度 (動きを除く) をメートル/秒 2 乗 (m/s²) で測定します。
温度 (1Hz): 周囲温度を摂氏で提供します。
BNO055 はコンパクトで、Adafruit_CircuitPython と呼ばれる Adafruit の Python ライブラリでサポートされています。さらに、Adafruit から C 言語ライブラリを入手できます。
★Adafruit_CircuitPython_BNO055[2]
★Adafruit_BNO055[3]
デモについては、ここでビデオをご覧ください:
Adafruit BNO055 デモ
★Adafruit BNO055 デモ[4]
ELM327 是一种广泛使用的 OBD-II(车载诊断)适配器,允许通过标准接口访问车辆发动机数据。它充当车辆 ECU(电子控制单元)和计算机或智能手机等外部设备之间的桥梁,能够检索诊断和性能数据。 ELM327 的主要特性和功能包括:★ELM327 信息[5]
OBD-II 协议支持:ELM327 支持各种 OBD-II 协议,包括 ISO 9141、ISO 14230 (KWP2000)、ISO 15765 (CAN) 等,使其与广泛的车辆兼容。
诊断故障代码(DTC):它可以读取和清除车辆 ECU 中的诊断故障代码,帮助识别和排除问题。
实时数据流:提供来自车辆传感器的实时数据,例如发动机转速、车速、冷却液温度和燃油油位。
冻结帧数据:在检测到故障时捕获并存储数据,这有助于诊断间歇性问题。
车辆信息:检索有关车辆的详细信息,包括 VIN(车辆识别码)、校准 ID 等。
兼容性:提供多种形式,包括蓝牙、USB、Wi-Fi版本,可以兼容不同的设备和平台。
通常,ELM327 用于诊断车辆状态,特别是发动机状态。然而,它也可以用来从车辆收集广泛的数据。 ELM327 可在亚马逊等平台购买,价格约为 20 美元到 80 美元不等。
使用 python-OBD 库,您可以创建查询以通过 ELM327 适配器收集数据,从而实现可定制和详细的数据收集
由于该系统与多个传感器和实际车辆交互,因此它必须了解每个坐标系。
下图显示了我的数据测量系统的 BNO055 坐标系。我的系统还使用四元数计算旋转角度。传感器坐标系需要与车辆坐标系相匹配。
测量系统基于车辆坐标系运行。
我使用了一款 iPhone 应用程序,它可以测量多个物理数据点,包括加速度、陀螺仪等,用于验证任务。使用iPhone应用程序的目的是通过该应用程序中的数据来确认我的数据测量系统的有效性。图4引自苹果开发者网站。
在本节中,我将展示一些实验结果。第一步,我进行了正弦波任务,以确认我的测量系统输出的读数是否没有错误,例如单位错误。这部分很基础,但对于确保以正确的方式收集数据至关重要。
接下来,我在实车上驾驶时测量了一些物理值。
然后,我检查了ELM327的性能和每个传感器的采样精度。
我进行了正弦波测试,在本节中,我将展示结果。这些测试的目的是确认以下内容:
为了实现这一目标,我使用了“phyphox”★phyphox[8],这是一款可以测量各种物理值的智能手机应用程序。我将 BNO055 传感器安装在 iPhone 上,如下所示 [图 5]。
图6显示了采样频率为10 Hz的正弦波测试结果。图例“VDDM”表示由我开发的测量系统收集的数据,而“iPhone”表示使用“phyphox”收集的数据。
来自 VDDM 和 iPhone 的数据在所有轴上都显示出相似的趋势,表明这两个应用程序正在捕获相似的运动模式。
VDDM 数据似乎稍有变化,特别是在加速度图中,这可能表明与 iPhone 相比具有更高的灵敏度或不同的过滤方法。
两个数据大体吻合,但幅度差异表明,根据精度要求,可能需要进一步校准或调整。
VDDM 可能会捕获额外的噪声或更高频率的分量,尤其是在加速度图中,而 iPhone 数据中不存在这些分量。这可能有利于详细分析或需要进一步过滤。
图 7 显示了针对实际车辆中的 BNO055 数据进行的实验结果。一个名为“phyphox”的 iPhone 应用程序被用作参考。采样频率为50Hz。我观察到,我在车辆上开发的数据记录系统测量的加速度和旋转信号包含噪声、异常值和 NaN 值。因此,我应用了后处理功能,包括 IQR 方法和巴特沃斯低通滤波器。每个图表的图例中的术语“fc”表示截止频率。例如,fc_19.5表示截止频率为19.5Hz。请注意,本实验仅使用 BNO055 传感器进行,并且 ELM327 功能被禁用。
来自 VDDM(原始数据)和 iPhone 的数据在所有轴上都显示出相似的趋势。这表明两个系统正在捕捉相同的运动模式。
与 iPhone 数据相比,VDDM 的数据似乎有更大的变化。这在加速图中尤其明显,可能表明 VDDM 以更高的灵敏度捕获数据或使用不同的过滤方法。
来自 VDDM 的过滤数据(fc 19.5Hz 和 10Hz)与原始数据相比变化较小,并且更接近 iPhone 数据。这表明 VDDM(原始数据)可能包含更多噪声或高频分量。
来自 VDDM 的原始数据包含异常值和缺失值,通过应用过滤器可以减轻这些异常值和缺失值。这在滚动率和俯仰率图表中尤其明显。
与 iPhone 相比,VDDM 似乎包含更多高频成分和噪声。虽然这有利于详细分析,但如果降噪至关重要,则可能需要额外的滤波或硬件级调整。
图 8 显示了有关发动机性能的数据。这些数据是使用 ELM327 扫描仪以 4 Hz 的采样频率收集的。 BNO055 功能在数据收集期间被禁用。
在使用 BNO055 和 ELM327 进行实验期间,我观察到采样延迟。为了找出问题,我通过调整采样频率以及激活和停用 ELM327 和 BNO055 来实施测量。
首先,我给出与采样频率相对应的时间差。此测试涉及使用 BNO055 和 ELM327 传感器
图9显示,当采样频率设置为10Hz时,采样时间的精度明显下降。
ELM327 无法实现与 BNO055 一样高的采样频率,这可能会对整体系统性能产生负面影响。尽管如此,为了更好地了解每个传感器的性能,我进行了单独的测量。
图10和图11说明了采样频率和采样定时精度之间的关系。采样时序精度表示为:
根据图 10,BNO055 可以响应高达 50 Hz 的采样频率。然而,超过 60 Hz 后,采样计时精度急剧下降。
图 11 显示 ELM327 的响应度低于 BNO055。 ELM327 在最高 5 Hz 时仍能保持良好的采样定时精度。这是因为 ELM327 通过 OBD 与车辆通信,而车辆的 ECU 可能不支持对 OBD 请求的高频响应,从而导致通信频率较低。
此外,ELM327 是可通过 OBD-II 从车辆中的多个 ECU 检索数据的传感器中最便宜且质量最低的选项。
这篇文章★挑战你汽车的行车电脑燃油效率数据[9]说
Elm327 需要考虑的一件重要事情是响应时间不容忽视:设备可能需要 0.1 到 0.3 秒的时间来响应消息请求,这意味着快速添加太多信号最终会减少采样率达到无用的程度
更多详细建议,请访问此页面:★选择OBDII适配器[10]
对于性能低下的原因,可以考虑以下原因:
我一直在开发的测量软件假设多个传感器需要的用例。此外,用户可以更改 yaml 文件上的测量设置。用户可以在简单的 GUI 上操作测量应用程序。您可以在这里查看源代码:★VDDM[11]
测量系统的结构如下:
VDDM/ │ ├── fusion/ │ ├── __init__.py │ ├── sensor_fusion.py # Handles data fusion, collects data from sensors │ └── sensors/ │ ├── __init__.py │ ├── bno055_measurement.py # Script for BNO055 sensor data collection │ └── elm327_measurement.py # Script for ELM327 OBD2 data collection │ ├── signalprocessing/ │ ├── __init__.py │ └── filter.py # Contains filtering algorithms (e.g., Butterworth filter) │ ├── config/ │ ├── config_manager.py # Loads configuration from YAML │ └── measurement_system_config.yaml # Configuration file for sensors and system settings │ ├── utils/ │ ├── tools.py # Utility functions (e.g., wait functions) │ └── visualize_data.py # Functions to format and display sensor data │ ├── measurement/ │ ├── __init__.py │ └── measurement_control.py # Controls the measurement process │ ├── gui/ │ ├── __init__.py │ └── main_gui.py # GUI setup for starting/stopping measurement │ ├── main.py # Entry point for starting the system └── requirements.txt # Project dependencies
下图表示 VDDM 的流程。
main.py └── main_gui.py ├── Starts GUI interface and buttons └── measurement_control.py ├── Initializes sensor measurements ├── Controls start/stop of measurements └── sensor_fusion.py ├── Manages sensor data collection and fusion ├── Uses: │ ├── bno055_measurement.py (BNO055 sensor data collection) │ └── elm327_measurement.py (ELM327 OBD2 data collection) └── Processes data with: ├── filter.py (Applies Butterworth low-pass filter) └── visualize_data.py (Formats and visualizes sensor data)
VDDM 的主要组件是 VDDM/fusion/sensor_fusion.py、VDDM/fusion/sensors/bno055_measurement.py 和 VDDM/fusion/sensors/elm327_measurement.py。
# sensor_fusion.py import os import sys import importlib from time import perf_counter from collections import defaultdict import pandas as pd import datetime import asyncio import numpy as np parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.append(parent_dir) from config import config_manager from utils.tools import wait_process from utils.visualize_data import format_sensor_fusion_data from signalprocessing.filter import butterlowpass config_path = os.path.join(parent_dir, 'config', 'measurement_system_config.yaml') class SensorFactory: @staticmethod def create_sensor(sensor_type, config): try: # Import the appropriate sensor module based on sensor_type module = importlib.import_module(f"fusion.sensors.{sensor_type}_measurement") # Get a sensor class sensor_class = getattr(module, f"{sensor_type.upper()}") # Create a sensor instance and return return sensor_class(config) except (ImportError, AttributeError) as e: print(f"Error creating sensor {sensor_type}: {e}") return None class Sensors: def __init__(self, config): self.config = config_manager.load_config(config_path) self.sensor_list = tuple(self.config.sensors.keys()) self.sensor_instances = {} self.is_running = False self.SAMPLING_FREQUENCY_HZ = config.sampling_frequency_hz self.SAMPLING_TIME = 1 / self.SAMPLING_FREQUENCY_HZ self.SAVE_DATA_DIR = config.save_data_dir self.SAVE_BUF_CSVDATA_PATH = self.SAVE_DATA_DIR + "/" + "measurement_raw_data.csv" self.SEQUENCE_LENGTH = config.sequence_length # Windows size [s] # Buffer size is determined by the relation of sequence length and sampling frequency # Buffer secures data for SEQUENCE_LENGTH[s] self.MAX_DATA_BUF_LEN = self.SEQUENCE_LENGTH * self.SAMPLING_FREQUENCY_HZ self.FPASS = config.filter_params.fpass self.FSTOP = config.filter_params.fstop self.GPASS = config.filter_params.gpass self.GSTOP = config.filter_params.gstop self.is_filter = config.filter_params.is_filter self.is_show_real_time_data = config.is_show_real_time_data self.TIMEZONE = config.timezone self.all_data_columns_list = () for sensor_name in self.sensor_list: self.all_data_columns_list += tuple(self.config["sensors"][sensor_name]["data_columns"]) self.data_buffer = pd.DataFrame() # data buffer for sensor_type in self.sensor_list: sensor_config = self.config.sensors[sensor_type] sensor_instance = SensorFactory.create_sensor(sensor_type, sensor_config) if sensor_instance: self.sensor_instances[sensor_type] = sensor_instance if os.path.exists(self.SAVE_BUF_CSVDATA_PATH): os.remove(self.SAVE_BUF_CSVDATA_PATH) print(f"File '{self.SAVE_BUF_CSVDATA_PATH}' was deleted for initialization") def get_sensor(self, sensor_type): """ Retrieve the sensor instance corresponding to the specified sensor type. Args: sensor_type (str): The type of the sensor to retrieve. Returns: object: The sensor instance corresponding to the specified sensor type. Returns None if the sensor type does not exist. """ return self.sensor_instances.get(sensor_type) def collect_data(self): """ Collect data from all sensors. This method iterates over all sensor instances and collects data from each sensor. The collected data is stored in a dictionary where the keys are sensor types and the values are the data collected from the corresponding sensors. Returns: dict: A dictionary containing the collected data from all sensors. The keys are sensor types and the values are the data from each sensor. Raises: Exception: If an error occurs while collecting data from any sensor, the exception is caught and printed. """ data = {} try: for sensor_type, sensor in self.sensor_instances.items(): # get data from sensors data[sensor_type] = sensor.get_data_from_sensor() return data except Exception as e: print(e) def on_change_start_measurement(self): """ Start the measurement process. This method sets the is_running flag to True, indicating that the measurement process should start. """ self.is_running = True def on_change_stop_measurement(self): """ Stop the measurement process. This method sets the is_running flag to False, indicating that the measurement process should stop. """ self.is_running = False def filtering(self, df, labellist): """ Apply a low-pass filter to the specified columns in the DataFrame. This method applies a Butterworth low-pass filter to each column specified in the labellist. The "Time" column should be excluded from the labellist as it is not needed for the computation. Args: df (pd.DataFrame): The input DataFrame containing the data to be filtered. labellist (list of str): A list of column names to be filtered. The "Time" column should not be included in this list. Returns: pd.DataFrame: A new DataFrame with the filtered data. """ filtered_df = df.copy() for labelname in labellist: # Ensure the column is converted to a numpy array x = df[labelname].to_numpy() filtered_df[labelname] = butterlowpass( x=x, # Correctly pass the numpy array as 'x' fpass=self.FPASS, fstop=self.FSTOP, gpass=self.GPASS, gstop=self.GSTOP, fs=self.SAMPLING_FREQUENCY_HZ, dt=self.SAMPLING_TIME, checkflag=False, labelname=labelname ) return filtered_df def convert_dictdata(self, current_time, sensor_data_dict): """ Convert nested dictionary data from multiple sensors into a single DataFrame. This method converts nested dictionary data obtained from multiple sensors into a single dictionary and then converts it into a pandas DataFrame. The current_time information is associated with the data. Args: current_time (float): The current time at which the data was obtained. sensor_data_dict (dict): A nested dictionary containing data from multiple sensors. Returns: pd.DataFrame: A DataFrame containing the converted data with the current time information. """ converted_data = {'Time': current_time} for sensor, data in sensor_data_dict.items(): converted_data.update(data) converted_data = pd.DataFrame([converted_data]) return converted_data async def update_data_buffer(self, dict_data): """ Add data from sensors to the buffer and save it if necessary. This method adds the provided sensor data to the internal buffer. If the buffer exceeds the specified maximum length, the oldest data is saved to a CSV file and removed from the buffer. Args: dict_data (dict): The data from sensors to be added to the buffer. """ # Add data to the buffer self.data_buffer = pd.concat([self.data_buffer, dict_data], ignore_index=True) # If the buffer exceeds the specified length, save the oldest data if len(self.data_buffer) > self.MAX_DATA_BUF_LEN: # Save the oldest data to a CSV file old_data = self.data_buffer.head(self.MAX_DATA_BUF_LEN) await self.save_data(old_data, self.SAVE_BUF_CSVDATA_PATH) # Update the buffer self.data_buffer = self.data_buffer.tail(len(self.data_buffer) - self.MAX_DATA_BUF_LEN) async def save_data_async(self, df, path): """ Save the DataFrame to a CSV file asynchronously. This method uses asyncio.to_thread to run the synchronous to_csv method in a separate thread, allowing it to be handled asynchronously. Args: df (pd.DataFrame): The DataFrame to be saved. path (str): The file path where the DataFrame should be saved. """ if not os.path.isfile(path): await asyncio.to_thread(df.to_csv, path, sep=',', encoding='utf-8', index=False, header=True, mode='w') else: await asyncio.to_thread(df.to_csv, path, sep=',', encoding='utf-8', index=False, header=False, mode='a') async def save_data(self, df, path): """ Save the DataFrame to a CSV file asynchronously. This method calls save_data_async to save the DataFrame to a CSV file asynchronously. Args: df (pd.DataFrame): The DataFrame to be saved. path (str): The file path where the DataFrame should be saved. """ await self.save_data_async(df, path) async def finish_measurement_and_save_data(self): """ Finish the measurement process and save the data. This method finalizes the measurement process by saving the buffered data to a CSV file. It also applies filtering if specified and saves the filtered data to a separate CSV file. The method handles time zone settings and generates a timestamp for the file names. The buffered data is saved to a temporary CSV file, which is then read back and saved to a final file path with a timestamp. If filtering is enabled, the filtered data is also saved. The temporary CSV file is deleted after the data is saved. Raises: Exception: If an error occurs during the file operations. """ t_delta = datetime.timedelta(hours=9) TIMEZONE = datetime.timezone(t_delta, self.TIMEZONE)# You have to set your timezone now = datetime.datetime.now(TIMEZONE) timestamp = now.strftime('%Y%m%d%H%M%S') final_file_path = self.SAVE_BUF_CSVDATA_PATH.replace(self.SAVE_BUF_CSVDATA_PATH.split('/')[-1], timestamp + "/" + timestamp + '_' + self.SAVE_BUF_CSVDATA_PATH.split('/')[-1]) await self.save_data_async(self.data_buffer, self.SAVE_BUF_CSVDATA_PATH) raw_df = pd.read_csv(self.SAVE_BUF_CSVDATA_PATH, header=0) os.makedirs(self.SAVE_DATA_DIR + "/" + timestamp, exist_ok=True) raw_df.to_csv(final_file_path, sep=',', encoding='utf-8', index=False, header=True) if self.is_filter: filt_df = self.filtering(df=raw_df, labellist=raw_df.columns[1:]) filt_df.to_csv(final_file_path.replace('_raw_data.csv', '_filt_data.csv'), sep=',', encoding='utf-8', index=False, header=True) if os.path.exists(self.SAVE_BUF_CSVDATA_PATH): os.remove(self.SAVE_BUF_CSVDATA_PATH) print(f"File '{self.SAVE_BUF_CSVDATA_PATH}' was deleted") else: print(f"File '{self.SAVE_BUF_CSVDATA_PATH}' is not existed") async def sensor_fusion_main(): """ Main function for sensor fusion. This function initializes the sensor fusion process, starts the measurement loop, collects data from multiple sensors, updates the data buffer, and handles real-time data display. It also calculates and prints the sampling delay and reliability rate upon termination. The main loop runs until the measurement process is stopped, either by an exception or a keyboard interrupt. Raises: Exception: If an error occurs during the measurement process, it is caught and printed. KeyboardInterrupt: If a keyboard interrupt occurs, the measurement process is stopped and the data is saved. """ print("Start sensor fusion main") config = config_manager.load_config(config_path) sensors = Sensors(config["master"]) print("Called an instance of Sensors class") # sensors.start_all_measurements() sampling_counter = 0 current_time = 0 #sensors.is_running = True sensors.on_change_start_measurement() try: main_loop_start_time = None while sensors.is_running: iteration_start_time = perf_counter() # Start time of each iteration if main_loop_start_time is None: main_loop_start_time = iteration_start_time # initialize main loop start time current_time = perf_counter() - main_loop_start_time # Current time data = sensors.collect_data() # Get data from sensors sampling_counter += 1 # Num of sampling converted_data = sensors.convert_dictdata(current_time, data) # Convert data to dataframe format # Update the data buffer. If it reaches the buffer limit, write the data to a CSV file. await sensors.update_data_buffer(converted_data) # Display data in real time. This process is executed on additional thread. if sensors.is_show_real_time_data: formatted_data = format_sensor_fusion_data(data, sensors.all_data_columns_list) print("--------------------------------------------------------------------") print("Current Time is: {:.3f}".format(current_time)) print(formatted_data) # Wait based on the sampling interval and execution time to maintain the sampling frequency. iteration_end_time = perf_counter() iteration_duration = iteration_end_time - iteration_start_time print("Iteration duration is: {0} [s]".format(iteration_duration)) sleep_time = max(0, sensors.SAMPLING_TIME - iteration_duration) if sleep_time > 0: wait_process(sleep_time) except Exception as e: print(e) except KeyboardInterrupt: sensors.on_change_stop_measurement() print("KeyboardInterrupt") await sensors.finish_measurement_and_save_data() finally: print("finish") # Compute delay of sampling main_loop_end_time = perf_counter() - main_loop_start_time print("Program terminated") print("main loop is ended. current time is: {:.3f}".format(current_time)) print("main loop is ended. end time is: {:.3f}".format(main_loop_end_time)) print("sampling num is: {}".format(sampling_counter)) # Compute ideal sampliing time ideal_time = ((sampling_counter - 1) / sensors.SAMPLING_FREQUENCY_HZ) # Cpmpute a delay delay_time = current_time - ideal_time # reliability rate sampling_reliability_rate = (delay_time / (sampling_counter / sensors.SAMPLING_FREQUENCY_HZ)) * 100 print("sampling delay is: {:.3f} s".format(delay_time)) print("sampling delay rate is: {:.3f} %".format(sampling_reliability_rate)) if __name__ == '__main__': asyncio.run(sensor_fusion_main())
sensor_fusion.py manages sensor data collection and processing using asynchronous operations. It dynamically creates sensor instances, collects data, applies filters, and saves the data to CSV files. The script also displays real-time data if needed and monitors performance metrics like sampling delay and reliability. The main function runs a loop that handles data collection, buffer updates, and performance reporting. It uses asynchronous operations to efficiently manage data saving and ensure smooth performance.
# bno055_measurement.py import time import numpy as np import adafruit_bno055 import board import os import sys parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) sys.path.append(parent_dir) from config.config_manager import load_config config_path = os.path.join(parent_dir, 'config', 'measurement_system_config.yaml') class BNO055: def __init__(self, config): self.COLUMNS = config.data_columns self.SAMPLING_FREQUENCY_HZ = config.sampling_frequency_hz self.SAMPLING_TIME = 1 / self.SAMPLING_FREQUENCY_HZ self.SAVE_DATA_DIR = config.save_data_dir self.SEQUENCE_LENGTH = config.sequence_length self.FPASS = config.filter_params.fpass self.FSTOP = config.filter_params.fstop self.GPASS = config.filter_params.gpass self.GSTOP = config.filter_params.gstop self.Isfilter = config.filter_params.is_filter self.IsStart = False self.IsStop = True self.Is_show_real_time_data = config.is_show_real_time_data i2c_instance = board.I2C() # Create i2c instance self.bno055_sensor = adafruit_bno055.BNO055_I2C(i2c_instance) # create BNO055_I2C instance def calibration(self): print("Start calibration!") while not self.bno055_sensor.calibrated: print('SYS: {0}, Gyro: {1}, Accel: {2}, Mag: {3}'.format(*(self.bno055_sensor.calibration_status))) time.sleep(1) def calcEulerfromQuaternion(self, _w, _x, _y, _z): """ Calculate Euler angles (roll, pitch, yaw) from quaternion components. This method converts quaternion components (_w, _x, _y, _z) into Euler angles (roll, pitch, yaw) in degrees. If any of the quaternion components are None, it returns (0.0, 0.0, 0.0) and prints an error message. Args: _w (float): The w component of the quaternion. _x (float): The x component of the quaternion. _y (float): The y component of the quaternion. _z (float): The z component of the quaternion. Returns: tuple: A tuple containing the roll, pitch, and yaw angles in degrees. If an error occurs, it returns (0.0, 0.0, 0.0) and prints an error message. """ if None in (_w, _x, _y, _z): print(f"Error: One or more quaternion values are None: {_w}, {_x}, {_y}, {_z}") return 0.0, 0.0, 0.0 try: sqw = _w ** 2 sqx = _x ** 2 sqy = _y ** 2 sqz = _z ** 2 COEF_EULER2DEG = 57.2957795131 # Yaw term1 = 2.0 * (_x * _y + _z * _w) term2 = sqx - sqy - sqz + sqw yaw = np.arctan2(term1, term2) # Pitch term1 = -2.0 * (_x * _z - _y * _w) term2 = sqx + sqy + sqz + sqw pitch = np.arcsin(term1 / term2) if -1 <= term1 / term2 <= 1 else 0.0 # Roll term1 = 2.0 * (_y * _z + _x * _w) term2 = -sqx - sqy + sqz + sqw roll = np.arctan2(term1, term2) return COEF_EULER2DEG * roll, COEF_EULER2DEG * pitch, COEF_EULER2DEG * yaw except Exception as e: print(f"Error in calcEulerfromQuaternion: {e}") return 0.0, 0.0, 0.0 def get_data_from_sensor(self): """ Retrieve data from the BNO055 sensor and return it as a dictionary. This method collects various sensor readings from the BNO055 sensor, including Euler angles, gyroscope data, linear acceleration, quaternion, magnetic field, and calibration status. It then constructs a dictionary with these values and returns only the columns specified in self.COLUMNS. Returns: dict: A dictionary containing the sensor data. Only the columns specified in self.COLUMNS are included in the returned dictionary. """ # Get data euler_z, euler_y, euler_x = [val for val in self.bno055_sensor.euler] # X: yaw, Y: pitch, Z: roll gyro_x, gyro_y, gyro_z = [val for val in self.bno055_sensor.gyro] # Gyro[rad/s] linear_accel_x, linear_accel_y, linear_accel_z = [val for val in self.bno055_sensor.linear_acceleration] # Linear acceleration[m/s^2] quaternion_1, quaternion_2, quaternion_3, quaternion_4 = [val for val in self.bno055_sensor.quaternion] # Quaternion quat_roll, quat_pitch, quat_yaw = self.calcEulerfromQuaternion(quaternion_1, quaternion_2, quaternion_3, quaternion_4) # Cal Euler angle from quaternion magnetic_x, magnetic_y, magnetic_z = [val for val in self.bno055_sensor.magnetic] # Magnetic field calibstat_sys, calibstat_gyro, calibstat_accel, calibstat_mag = [val for val in self.bno055_sensor.calibration_status] # Status of calibration data_dict = { "linear_accel_x": linear_accel_x, "linear_accel_y": linear_accel_y, "linear_accel_z": linear_accel_z, "gyro_x": gyro_x, "gyro_y": gyro_y, "gyro_z": gyro_z, "euler_x": euler_x, "euler_y": euler_y, "euler_z": euler_z, "quat_roll": quat_roll, "quat_pitch": quat_pitch, "quat_yaw": quat_yaw, "quaternion_1": quaternion_1, "quaternion_2": quaternion_2, "quaternion_3": quaternion_3, "quaternion_4": quaternion_4, "magnetic_x": magnetic_x, "magnetic_y": magnetic_y, "magnetic_z": magnetic_z, "calibstat_sys": calibstat_sys, "calibstat_gyro": calibstat_gyro, "calibstat_accel": calibstat_accel, "calibstat_mag": calibstat_mag } return {column: data_dict[column] for column in self.COLUMNS if column in data_dict} def format_sensor_data(data, labels): """ Format sensor data into a string for display. This method takes a dictionary of sensor data and a list of labels, and formats the data into a string where each label is followed by its corresponding value. If a value is None, it is replaced with the string "None". Each label-value pair is separated by " / ". Args: data (dict): The sensor data to format. labels (list of str): The list of labels to include in the formatted string. Returns: str: A formatted string containing the sensor data. """ formatted_str = "" if isinstance(data, dict): for label in labels: value = data.get(label, None) if value is None: value = "None" else: value = f"{value:.4f}" formatted_str += f"{label}: {value} / " return formatted_str.rstrip(" / ") def test_main(): """ Main function for testing sensor data collection and display. This function initializes the BNO055 sensor, starts a loop to collect data, formats the data for display, and prints it in real-time. It also calculates and prints the sampling delay and reliability rate upon termination. The main loop runs until interrupted by the user. Raises: KeyboardInterrupt: If a keyboard interrupt occurs, the loop is terminated and the final statistics are printed. """ from utils.tools import wait_process from time import perf_counter import matplotlib.pyplot as plt print("Main start") config = load_config(config_path) meas_bno055 = BNO055(config.sensors['bno055']) start_time = perf_counter() sampling_counter = 0 try: main_loop_start_time = perf_counter() while True: iteration_start_time = perf_counter() # Data acquisition process data = meas_bno055.get_data_from_sensor() current_time = perf_counter() - start_time sampling_counter += 1 if meas_bno055.Is_show_real_time_data: formatted_data = format_sensor_data(data, meas_bno055.COLUMNS) # current time print("--------------------------------------------------------------------") print("Current Time is: {:.3f}".format(current_time)) print(formatted_data) # Wait to meet the sampling frequency based on the sampling interval and execution time elapsed_time = perf_counter() - iteration_start_time sleep_time = meas_bno055.SAMPLING_TIME - elapsed_time if sleep_time > 0: wait_process(sleep_time) except KeyboardInterrupt: print("Interrupted by user") finally: # Calculate the sampling delay from the number of samples and the current time main_loop_end_time = perf_counter() - main_loop_start_time print("Program terminated") print("main loop is ended. current time is: {:.3f}".format(current_time)) print("main loop is ended. end time is: {:.3f}".format(main_loop_end_time)) print("sampling num is: {}".format(sampling_counter)) # Since it is 0-based, the number of samples is current_time + 1 # Calculate the ideal sampling time ideal_time = ((sampling_counter - 1) / meas_bno055.SAMPLING_FREQUENCY_HZ) # Calculate the delay delay_time = current_time - ideal_time # The reliability rate is the delay divided by the sampling time sampling_reliability_rate = (delay_time / (sampling_counter / meas_bno055.SAMPLING_FREQUENCY_HZ)) * 100 print("sampling delay is: {:.3f} s".format(delay_time)) print("sampling delay rate is: {:.3f} %".format(sampling_reliability_rate)) if __name__ == '__main__': test_main()
This script interfaces with the BNO055 sensor to collect and process data. It retrieves sensor readings, such as Euler angles and gyroscope data, formats them for display, and prints real-time updates if configured. The script includes a calibration method and calculates Euler angles from quaternion data. It manages data acquisition with a defined sampling frequency and calculates sampling delay and reliability. The main loop continues until interrupted, reporting performance metrics upon termination.
# elm327_measurement.py import obd import os import time from collections import deque import numpy as np import pandas as pd import datetime import asyncio import scipy from scipy import signal import matplotlib as plt import sys from collections import defaultdict import random parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) sys.path.append(parent_dir) from config.config_manager import load_config config_path = os.path.join(parent_dir, 'config', 'measurement_system_config.yaml') class ELM327: def __init__(self, config): """ Initialize the ELM327 class with configuration parameters. Args: config (dict): Configuration parameters for the ELM327. """ self.COLUMNS = config.data_columns self.SAMPLING_FREQUENCY_HZ = config.sampling_frequency_hz self.SAMPLING_TIME = 1 / self.SAMPLING_FREQUENCY_HZ self.SAVE_DATA_DIR = config.save_data_dir self.SEQUENCE_LENGTH = config.sequence_length self.FPASS = config.filter_params.fpass self.FSTOP = config.filter_params.fstop self.GPASS = config.filter_params.gpass self.GSTOP = config.filter_params.gstop self.Isfilter = config.filter_params.is_filter self.res = self.connect_to_elm327() self.is_offline = config.is_offline self.IsStart = False self.IsStop = True self.Is_show_real_time_data = config.is_show_real_time_data def initialize_BLE(self): """ Initialize Bluetooth Low Energy (BLE) for ELM327 connection. """ os.system('sudo hcitool scan') os.system('sudo hciconfig hci0 up') os.system('sudo rfcomm bind 0 8A:2A:D4:FF:38:F3') os.system('sudo rfcomm listen 0 1 &') def connect_to_elm327(self): """ Establish a connection to the ELM327 device. Returns: res (obd.OBDStatus): The connection status of the ELM327 device. """ res = None try: self.initialize_BLE() self.connection = obd.OBD() print(self.connection.status()) res = self.connection.status() if res == obd.OBDStatus.CAR_CONNECTED: print("----------Connection establishment is successful!----------") return res else: print("----------Connection establishment failed!----------") print("End program. Please check settings of the computer and ELM327") except Exception as e: print("----------Exception!----------") print(e) finally: return res def get_data_from_sensor(self): """ Retrieve data from the sensor. Returns: dict: A dictionary containing sensor data. """ if self.is_offline: data = self.get_data_from_sensor_stub() else: # Retrieve data and save it in dictionary format data = {column: self.get_obd2_value(column) for column in self.COLUMNS} return data def get_obd2_value_debug(self, column): """ Retrieve OBD-II value for a specific column with debug information. Args: column (str): The OBD-II command column. Returns: float or None: The value of the OBD-II command, or None if not available. """ command = getattr(obd.commands, column, None) if command: response = self.connection.query(command) if response: print(f"Response for command '{command}': {response}") if response.value is not None: # Check for None print(f"Response value for command '{command}': {response.value}") return response.value.magnitude else: print(f"No value in response for command '{command}'") else: print(f"No response for command '{command}'") else: print(f"No command found for column '{column}'") return None def get_obd2_value(self, column): """ Retrieve OBD-II value for a specific column. Args: column (str): The OBD-II command column. Returns: float or None: The value of the OBD-II command, or None if not available. """ command = getattr(obd.commands, column, None) if command: response = self.connection.query(command) if response.value is not None: # Check for None return response.value.magnitude return None def get_data_from_sensor_stub(self): """ Generate stub data for the sensor. Returns: dict: A dictionary containing stub sensor data. """ data_stub = {column: np.abs(np.random.randn()).astype(np.float32).item() for column in self.COLUMNS} # Randomly insert None or 0.0 if random.choice([True, False]): random_column = random.choice(self.COLUMNS) if random.choice([True, False]): data_stub[random_column] = None else: data_stub[random_column] = 0.0 return data_stub def format_data_for_display(data, labels): """ Format sensor data for display. Args: data (dict): The sensor data to format. labels (list of str): The list of labels to include in the formatted string. Returns: str: A formatted string containing the sensor data. """ formatted_str = "" for label, value in zip(labels, data.values()): if value is None: value = "None" else: value = f"{value:.4f}" formatted_str += f"{label}: {value} / " return formatted_str.rstrip(" / ") def format_sensor_data(data, labels): """ Format sensor data for display. Args: data (dict or list): The sensor data to format. labels (list of str): The list of labels to include in the formatted string. Returns: str: A formatted string containing the sensor data. """ formatted_str = "" if isinstance(data, dict): for label in labels: value = data.get(label, None) if value is None: value = "None" else: value = f"{value:.4f}" formatted_str += f"{label}: {value} / " else: for label, value in zip(labels, data): if value is None: value = "None" else: value = f"{value:.4f}" formatted_str += f"{label}: {value} / " return formatted_str.rstrip(" / ") def test_main(): """ Main function for testing sensor data collection and display. This function initializes the ELM327 sensor, starts a loop to collect data, formats the data for display, and prints it in real-time. It also calculates and prints the sampling delay and reliability rate upon termination. The main loop runs until interrupted by the user. Raises: KeyboardInterrupt: If a keyboard interrupt occurs, the loop is terminated and the final statistics are printed. """ from utils.tools import wait_process from time import perf_counter import matplotlib.pyplot as plt print("Main start") config = load_config(config_path) meas_elm327 = ELM327(config.sensors['elm327']) # res = meas_elm327.connect_to_elm327() start_time = perf_counter() sampling_counter = 0 try: main_loop_start_time = perf_counter() while True: iteration_start_time = perf_counter() # Data acquisition process data = meas_elm327.get_data_from_sensor() current_time = perf_counter() - start_time sampling_counter += 1 if meas_elm327.Is_show_real_time_data: formatted_data = format_sensor_data(data, meas_elm327.COLUMNS) print("--------------------------------------------------------------------") print("Current Time is: {:.3f}".format(current_time)) print(formatted_data) # Wait to meet the sampling frequency based on the sampling interval and execution time elapsed_time = perf_counter() - iteration_start_time sleep_time = meas_elm327.SAMPLING_TIME - elapsed_time if sleep_time > 0: wait_process(sleep_time) except KeyboardInterrupt: print("Interrupted by user") finally: main_loop_end_time = perf_counter() - main_loop_start_time print("Program terminated") print("main loop is ended. current time is: {:.3f}".format(current_time)) print("main loop is ended. end time is: {:.3f}".format(main_loop_end_time)) print("sampling num is: {}".format(sampling_counter)) # Calculate the ideal sampling time ideal_time = ((sampling_counter - 1) / meas_elm327.SAMPLING_FREQUENCY_HZ) # Calculate the delay delay_time = current_time - ideal_time # The reliability rate is the delay divided by the sampling time sampling_reliability_rate = (delay_time / (sampling_counter / meas_elm327.SAMPLING_FREQUENCY_HZ)) * 100 print("sampling delay is: {:.3f} s".format(delay_time)) print("sampling delay rate is: {:.3f} %".format(sampling_reliability_rate)) if __name__ == '__main__': test_main()
This script defines an ELM327 class for interfacing with an ELM327 device to collect sensor data via OBD-II. It initializes the connection, retrieves data, and formats it for display. The test_main function sets up the ELM327 sensor, collects data in a loop, and prints it in real-time. It also calculates and displays sampling delay and reliability rates upon interruption. The script handles both real-time data acquisition and simulated offline data for testing purposes.
In this post, I described the project focused on developing a measurement system to capture vehicle behavior. While the project is still ongoing, I have recognized the potential for capturing vehicle data with a focus on the vehicle dynamics domain. However, when it comes to measuring signals from vehicle ECUs, some limitations need to be considered, as the ELM327 may not reliably achieve sampling rates beyond 5 Hz with good accuracy. This issue could likely be resolved by using a more advanced OBD scanner, such as the OBDLink MX+.
[1] Adafruit. "Adafruit BNO055 Absolute Orientation Sensor."
https://learn.adafruit.com/adafruit-bno055-absolute-orientation-sensor/overview. Accessed September 3, 2024.
[2] Adafruit. "Adafruit_CircuitPython_BNO055." https://github.com/adafruit/Adafruit_CircuitPython_BNO055. Accessed September 3, 2024.
[3] Adafruit. "Adafruit_BNO055." https://github.com/adafruit/Adafruit_BNO055. Accessed September 3, 2024.
[4] Adafruit. "Adafruit BNO055 Demo." https://cdn-shop.adafruit.com/product-videos/1024x768/2472-04.mp4 Accessed September 3, 2024.
[5] Amazon. "Elm327 Launchh OBD2 Professional Bluetooth Scan Tool and Code Reader for Android and PC, Interface OBDII OBD2 Car Auto Diagnostic Scanner, Not Support J1850 VPW & J1850 PWM". https://a.co/d/5BFn4GN. Accessed September 3, 2024.
[6] MathWorks. "Coordinate Systems in Automated Driving Toolbox." https://www.mathworks.com/help/driving/ug/coordinate-systems.html. Accessed September 3, 2024.
[7] Apple. "Getting raw gyroscope events." https://developer.apple.com/documentation/coremotion/getting_raw_gyroscope_events. Accessed September 3, 2024.
[8] phyphox. "phyphox top page."https://phyphox.org/. Accessed September 3, 2024.
[9] Andrea Patrucco, Vehicle dynamics engineer presso Applus+ IDIADA. "Challenging your car's trip computer fuel efficiency figures." https://www.linkedin.com/pulse/challenging-your-cars-trip-computer-fuel-efficiency-figures-patrucco/. Accessed September 3, 2024.
[10] "Choosing OBDII adapter". https://www.carscanner.info/choosing-obdii-adapter/. Accessed September 3, 2024.
[11] "VDDM". https://github.com/Qooniee/VDDM/tree/master/. Accessed September 3, 2024.
以上がRaspberry Pi を使用した車両用カスタム データ ロガーの構築: BNO センサーと ELM センサーの統合の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。