"""Logic to handle and remotely-manage the BMP390 sensor"""
from time import sleep
from typing import List, Sequence, Tuple
from smbus2 import SMBus
from pi_weather.codes import I2C_DEVICE_NOT_FOUND_EXIT_CODE
from pi_weather.config.config_handler import Config
from pi_weather.core.sensors.bmp390.addresses import (
BMP3_FIFO_CONFIG_CHANGE,
BMP3_FIFO_PRESS_FRAME,
BMP3_FIFO_TEMP_FRAME,
BMP3_FIFO_TEMP_PRESS_FRAME,
BMP3_FIFO_TIME_FRAME,
BMP3XX_CALIB_DATA,
BMP3XX_CHIP_ID,
BMP3XX_CMD,
BMP3XX_CMD_FIFO_FLUSH,
BMP3XX_CMD_RESET,
BMP3XX_ERR_REG,
BMP3XX_FIFO_CONF_1,
BMP3XX_FIFO_CONF_2,
BMP3XX_FIFO_DATA,
BMP3XX_FIFO_LENGTH,
BMP3XX_FIFO_WTM_0,
BMP3XX_FIFO_WTM_1,
BMP3XX_IIR_CONFIG,
BMP3XX_IIR_CONFIG_COEF_0,
BMP3XX_IIR_CONFIG_COEF_1,
BMP3XX_IIR_CONFIG_COEF_3,
BMP3XX_INT_CTRL,
BMP3XX_ODR,
BMP3XX_ODR_0P01_HZ,
BMP3XX_ODR_12P5_HZ,
BMP3XX_ODR_25_HZ,
BMP3XX_ODR_50_HZ,
BMP3XX_ODR_100_HZ,
BMP3XX_OSR,
BMP3XX_P_DATA_PA,
BMP3XX_PRESS_OSR_SETTINGS,
BMP3XX_PWR_CTRL,
BMP3XX_TEMP_OSR_SETTINGS,
BMP388_CHIP_ID,
BMP390L_CHIP_ID,
FORCED_MODE,
NORMAL_MODE,
SLEEP_MODE,
)
from pi_weather.core.sensors.bmp390.constants import (
BMP3XX_FIFO_DATA_FRAME_LENGTH,
STANDARD_SEA_LEVEL_PRESSURE_PA,
)
from pi_weather.core.sensors.bmp390.precisions import (
HIGH_PRECISION,
LOW_PRECISION,
NORMAL_PRECISION1,
NORMAL_PRECISION2,
ROUNDING_PRECISION,
ULTRA_LOW_PRECISION,
ULTRA_PRECISION,
)
from pi_weather.utils.convert import uint8_to_int, uint16_to_int
from pi_weather.utils.logs import get_logger
logger = get_logger(__name__)
[docs]
class BMP3XX:
"""
BMP3XX base class
"""
[docs]
def __init__(self) -> None:
"""
Class c-tor func
"""
logger.info("Initializing a BMP3XX class")
# Sea level pressure in Pa.
self.sea_level_pressure: int = STANDARD_SEA_LEVEL_PRESSURE_PA
self._data_calib: Tuple[float | int, ...] = ()
[docs]
def begin(self) -> bool:
"""
Function to initialize the sensor
:return: Initialization status (boolean := True indicates initialization succeed, False otherwise)
:rtype: bool
"""
logger.info("Initializing the sensor")
ret: bool = True
chip_id: List[int] = self._read_reg(BMP3XX_CHIP_ID, 1)
logger.info(chip_id[0])
if chip_id[0] not in (BMP388_CHIP_ID, BMP390L_CHIP_ID):
ret = False
self._get_coefficients()
self.reset()
return ret
@property
def get_pressure(self) -> float | int:
"""
Function to get pressure measurement value from register
(working range := (300‒1250 hPa))
If the reference value is provided before, the absolute value of the current position pressure
is calculated according to the calibrated sea level atmospheric pressure
:return: Pressure measurements (unit: Pa)
:rtype: float | int
"""
logger.info("Getting pressure data point...")
adc_p, adc_t = self._get_reg_temp_press_data()
return self._compensate_data(adc_p, adc_t)[0]
@property
def get_temperature(self) -> float | int:
"""
Function to get pressure measurement value from register
(working range := (-40 ‒ +85 °C))
:return: Temperature measurements (unit: °C)
:rtype: float | int
"""
logger.info("Getting temperature data point...")
adc_p, adc_t = self._get_reg_temp_press_data()
return self._compensate_data(adc_p, adc_t)[1]
@property
def get_sensor_time(self) -> int:
"""
Function to get sensor time data
:return: Sensor time
:rtype: int
"""
return self._get_sensor_time()
# Function called outside (to configure the instance once called)
[docs]
def calibrated_absolute_difference(self, altitude: float) -> bool:
"""
Function which takes the given current location altitude as the reference value
to eliminate the absolute difference for subsequent pressure and altitude data
:param altitude: Altitude in current position
:type altitude: float
:return: Boolean (True if the benchmark value is successful, False otherwise)
:rtype: bool
"""
# The altitude in meters based on the currently set sea level pressure.
ret: bool = False
if STANDARD_SEA_LEVEL_PRESSURE_PA == self.sea_level_pressure:
self.sea_level_pressure = self.get_pressure / pow(
1.0 - (altitude / 44307.7), 5.255302
)
ret = True
return ret
@property
def get_altitude(self) -> float:
"""
Function to calculate and return the altitude based on the atmospheric pressure measured by the sensor
If the reference value is provided before, the absolute value of the current position pressure is
calculated according to the calibrated sea level atmospheric pressure
(see formula: https://www.weather.gov/media/epz/wxcalc/pressureAltitude.pdf)
:return: Altitude (unit: m)
:rtype: float
"""
logger.info("Getting altitude data point...")
return 44307.7 * (
1 - (self.get_pressure / STANDARD_SEA_LEVEL_PRESSURE_PA) ** 0.190284
)
[docs]
def set_power_mode(self, mode: int) -> None:
"""
Function to configure the measurement mode and power mode
- **SLEEP_MODE** (Sleep mode): It will be in sleep mode by default after power-on reset. In this mode,no
measurement is performed and power consumption is minimal. All registers
are accessible for reading the chip ID and compensation coefficient.
- **FORCED_MODE** (Forced mode): In this mode, the sensor will take a single measurement according to the selected
measurement and filtering options. After the measurement is completed, the sensor
will return to sleep mode, and the measurement result can be obtained in the register.
- **NORMAL_MODE** (Normal mode): Continuously loop between the measurement period and the standby period.
The output data rates are related to the ODR mode setting.
:param mode: Measurement mode and power mode that need to be set
:type mode: int
:return: Nothing
:rtype: None
"""
logger.info("Configuring power mode...")
temp = self._read_reg(BMP3XX_PWR_CTRL, 1)[0]
if (mode | 0x03) == temp:
logger.info("Same configuration as before!")
else:
if mode != SLEEP_MODE:
self._write_reg(BMP3XX_PWR_CTRL, (SLEEP_MODE & 0x30) | 0x03)
sleep(
Config().get("sensors", "bmp390", "safety_reg_write_sleep_period")
)
self._write_reg(BMP3XX_PWR_CTRL, (mode & 0x30) | 0x03)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def enable_fifo(self, mode: bool) -> None:
"""
Function to enable or disable FIFO
:param mode: Bool to enable or disable FIFO
:type mode: bool
:return: Nothing
:rtype: None
"""
logger.info("Enabling or disabling FIFO...")
if mode:
# Enable and initialize FIFO configuration
self._write_reg(BMP3XX_FIFO_CONF_1, 0x1D)
self._write_reg(BMP3XX_FIFO_CONF_2, 0x0C)
else:
# 关闭FIFO。
self._write_reg(BMP3XX_FIFO_CONF_1, 0x1C)
self._write_reg(BMP3XX_FIFO_CONF_2, 0x0C)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def set_oversampling(self, press_osr_set: int, temp_osr_set: int) -> None:
"""
Function to set up the oversampling config for pressure + temperature measurements
OSR stands for over-sampling register
There are 6 pressure oversampling modes:
- **BMP3XX_PRESS_OSR_SETTINGS[0]**, Pressure sampling×1, 16 bit / 2.64 Pa (Recommend temperature oversampling×1)
- **BMP3XX_PRESS_OSR_SETTINGS[1]**, Pressure sampling×2, 16 bit / 2.64 Pa (Recommend temperature oversampling×1)
- **BMP3XX_PRESS_OSR_SETTINGS[2]**, Pressure sampling×4, 18 bit / 0.66 Pa (Recommend temperature oversampling×1)
- **BMP3XX_PRESS_OSR_SETTINGS[3]**, Pressure sampling×8, 19 bit / 0.33 Pa (Recommend temperature oversampling×2)
- **BMP3XX_PRESS_OSR_SETTINGS[4]**, Pressure sampling×16, 20 bit / 0.17 Pa (Recommend temperature oversampling×2)
- **BMP3XX_PRESS_OSR_SETTINGS[5]**, Pressure sampling×32, 21 bit / 0.085 Pa (Recommend temperature oversampling×2)
There are 6 temperature oversampling modes:
- **BMP3XX_TEMP_OSR_SETTINGS[0]**, Temperature sampling×1, 16 bit / 0.0050 °C
- **BMP3XX_TEMP_OSR_SETTINGS[1]**, Temperature sampling×2, 16 bit / 0.0025 °C
- **BMP3XX_TEMP_OSR_SETTINGS[2]**, Temperature sampling×4, 18 bit / 0.0012 °C
- **BMP3XX_TEMP_OSR_SETTINGS[3]**, Temperature sampling×8, 19 bit / 0.0006 °C
- **BMP3XX_TEMP_OSR_SETTINGS[4]**, Temperature sampling×16, 20 bit / 0.0003 °C
- **BMP3XX_TEMP_OSR_SETTINGS[5]**, Temperature sampling×32, 21 bit / 0.00015 °C
:param press_osr_set: Pressure oversampling mode to be set
:type press_osr_set: int
:param temp_osr_set: Temperature oversampling mode to be set
:type temp_osr_set: int
:return: Nothing
:rtype: None
"""
logger.info("Setting up OSR mode for both temperature and pressure...")
self._write_reg(BMP3XX_OSR, (press_osr_set | temp_osr_set) & 0x3F)
[docs]
def filter_coefficient(self, iir_config_coef: int) -> None:
"""
Function to set IIR filter coefficient (IIR filtering)
Infinite impulse response (IIR)
The environmental pressure is subject to many short-term changes, caused e.g. by slamming of a door or window, or wind wing into the sensor.
To suppress these disturbances in the output data without causing additional interface traffic and processor work load, the BMP390 features an internal IIR filter.
It effectively reduces the bandwidth of the output signals.
Docs: https://en.wikipedia.org/wiki/Infinite_impulse_response
Configurable mode:
BMP3XX_IIR_CONFIG_COEF_0, BMP3XX_IIR_CONFIG_COEF_1, BMP3XX_IIR_CONFIG_COEF_3,
BMP3XX_IIR_CONFIG_COEF_7, BMP3XX_IIR_CONFIG_COEF_15, BMP3XX_IIR_CONFIG_COEF_31,
BMP3XX_IIR_CONFIG_COEF_63, BMP3XX_IIR_CONFIG_COEF_127
:param iir_config_coef: IIR filter coefficient (IIR filtering)
:type iir_config_coef: int
:return: Nothing
:rtype: None
"""
# The IIR filter coefficient.
logger.info("Setting up the IIR filter coefficient...")
self._write_reg(BMP3XX_IIR_CONFIG, iir_config_coef & 0x0E)
[docs]
def set_output_data_rates(self, odr_set: int) -> bool:
"""
Function to set output data rate
Here are the available modes:
BMP3XX_ODR_200_HZ, BMP3XX_ODR_100_HZ, BMP3XX_ODR_50_HZ, BMP3XX_ODR_25_HZ, BMP3XX_ODR_12P5_HZ,
BMP3XX_ODR_6P25_HZ, BMP3XX_ODR_3P1_HZ, BMP3XX_ODR_1P5_HZ, BMP3XX_ODR_0P78_HZ, BMP3XX_ODR_0P39_HZ,
BMP3XX_ODR_0P2_HZ, BMP3XX_ODR_0P1_HZ, BMP3XX_ODR_0P05_HZ, BMP3XX_ODR_0P02_HZ, BMP3XX_ODR_0P01_HZ,
BMP3XX_ODR_0P006_HZ, BMP3XX_ODR_0P003_HZ, BMP3XX_ODR_0P0015_HZ
:param odr_set: Output Data Rate (ODR) to be set
:type odr_set: int
:return: True if config was successfully set, False otherwise (it remains at the original state)
:rtype: bool
"""
# The IIR filter coefficient.
ret: bool = True
self._write_reg(BMP3XX_ODR, odr_set & 0x1F)
if self._read_reg(BMP3XX_ERR_REG, 1)[0] & 0x04:
logger.warning("Sensor configuration error detected!")
ret = False
return ret
# Function called outside (to configure the instance once called)
[docs]
def set_common_sampling_mode(self, mode: int) -> bool:
"""
Function to set the sampling mode to the sensor
There are 6 commonly used sampling modes:
- **ULTRA_LOW_PRECISION**: Ultra-low precision, suitable for monitoring weather (lowest power consumption), the power is mandatory mode.
- **LOW_PRECISION**: Low precision, suitable for random detection, power is normal mode
- **NORMAL_PRECISION1**: Normal precision 1, suitable for dynamic detection on handheld devices (e.g. on mobile phones), power is normal mode
- **NORMAL_PRECISION2**: Normal precision 2, suitable for drones, power is normal mode
- **HIGH_PRECISION**: High precision, suitable for low-power handled devices (e.g. mobile phones), power is normal mode
- **ULTRA_PRECISION**: Ultra-high precision, suitable for indoor navigation, its acquisition rate will be extremely low, and the acquisition cycle is 1000 ms.
:param mode: Mode (integer)
:type mode: int
:return: Status bool (True if config was set successfully, False otherwise)
:rtype: bool
"""
ret: bool = True
if mode == ULTRA_LOW_PRECISION:
self.set_power_mode(FORCED_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[0], BMP3XX_TEMP_OSR_SETTINGS[0]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_0)
self.set_output_data_rates(BMP3XX_ODR_0P01_HZ)
elif mode == LOW_PRECISION:
self.set_power_mode(NORMAL_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[1], BMP3XX_TEMP_OSR_SETTINGS[0]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_0)
self.set_output_data_rates(BMP3XX_ODR_100_HZ)
elif mode == NORMAL_PRECISION1:
self.set_power_mode(NORMAL_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[2], BMP3XX_TEMP_OSR_SETTINGS[0]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_3)
self.set_output_data_rates(BMP3XX_ODR_50_HZ)
elif mode == NORMAL_PRECISION2:
self.set_power_mode(NORMAL_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[3], BMP3XX_TEMP_OSR_SETTINGS[0]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_1)
self.set_output_data_rates(BMP3XX_ODR_50_HZ)
elif mode == HIGH_PRECISION:
self.set_power_mode(NORMAL_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[3], BMP3XX_TEMP_OSR_SETTINGS[0]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_1)
self.set_output_data_rates(BMP3XX_ODR_12P5_HZ)
elif mode == ULTRA_PRECISION:
self.set_power_mode(NORMAL_MODE)
self.set_oversampling(
BMP3XX_PRESS_OSR_SETTINGS[4], BMP3XX_TEMP_OSR_SETTINGS[1]
)
self.filter_coefficient(BMP3XX_IIR_CONFIG_COEF_3)
self.set_output_data_rates(BMP3XX_ODR_25_HZ)
else:
ret = False
return ret
[docs]
def enable_data_ready_interrupt(self) -> None:
"""
Function to enable the interrupt for the signal triggered when data is ready
:return: Nothing
:rtype: None
"""
logger.info("Enabling interrupt for the data-ready signal...")
self._write_reg(BMP3XX_INT_CTRL, 0x42)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def enable_fifo_wtm_interrupt(self, wtm_value: int) -> None:
"""
Function to enable the interrupt for the water level signal
:param wtm_value: Water level value of FIFO (value range := (0-511))
:type wtm_value: int
:return: Nothing
:rtype: None
"""
logger.info("Enabling interrupt for the water level signal...")
self._write_reg(BMP3XX_INT_CTRL, 0x0A)
self._write_reg(BMP3XX_FIFO_WTM_0, wtm_value & 0xFF)
self._write_reg(BMP3XX_FIFO_WTM_1, (wtm_value >> 8) & 0x01)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def enable_fifo_full_interrupt(self) -> None:
"""
Function to enable the interrupt of the signal that the sensor FIFO is full
As the interrupt pin is unique, the three interrupts are set to be used separately,
please note the other two interrupt functions when using
:return: Nothing
:rtype: None
"""
logger.info("Enabling interrupt for the full sensor FIFO signal...")
self._write_reg(BMP3XX_INT_CTRL, 0x12)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def _get_coefficients(self) -> None:
"""
Function to get the calibration data in the NVM register of the sensor
:return: Nothing
:rtype: None
"""
logger.info("Getting calibration data from NVM register")
calib: List[int] = self._read_reg(BMP3XX_CALIB_DATA, 21)
self._data_calib = (
((calib[1] << 8) | calib[0]) / 2**-8.0, # T1
((calib[3] << 8) | calib[2]) / 2**30.0, # T2
uint8_to_int(calib[4]) / 2**48.0, # T3
(uint16_to_int((calib[6] << 8) | calib[5]) - 2**14.0) / 2**20.0, # P1
(uint16_to_int((calib[8] << 8) | calib[7]) - 2**14.0) / 2**29.0, # P2
uint8_to_int(calib[9]) / 2**32.0, # P3
uint8_to_int(calib[10]) / 2**37.0, # P4
((calib[12] << 8) | calib[11]) / 2**-3.0, # P5
((calib[14] << 8) | calib[13]) / 2**6.0, # P6
uint8_to_int(calib[15]) / 2**8.0, # P7
uint8_to_int(calib[16]) / 2**15.0, # P8
(uint16_to_int(calib[18] << 8) | calib[17]) / 2**48.0, # P9
uint8_to_int(calib[19]) / 2**48.0, # P10
uint8_to_int(calib[20]) / 2**65.0, # P11
)
[docs]
def _compensate_data(self, adc_p: float, adc_t: float) -> Tuple[float, float]:
"""
Function to compensate the original values (pressure + temperature) from the measured data (directly from registers).
It uses the calibration data (from registers data) retrieved at base class init.
# Docs on computation over calib. data := Datasheet, p28, Trimming Coefficient listing in register map with size and sign attributes
:param adc_p: Measured pressure (:= direct register value) (unit: Pa)
:type adc_p: float
:param adc_t: Measured temperature (:= direct register value) (unit: °C)
:type adc_t: float
:return: Tuple of calibrated pressure + temperature data
:rtype: Tuple[float, float]
"""
logger.info("Compensating data (pressure + temperature) ...")
t1, t2, t3, p1, p2, p3, p4, p5, p6, p7, p8, p9, p10, p11 = self._data_calib
# logger.info(self._data_calib)
# Temperature compensation
pd1 = adc_t - t1
pd2 = pd1 * t2
temperature = pd2 + (pd1 * pd1) * t3
# Pressure compensation
pd1 = p6 * temperature
pd2 = p7 * temperature**2.0
pd3 = p8 * temperature**3.0
po1 = p5 + pd1 + pd2 + pd3
pd1 = p2 * temperature
pd2 = p3 * temperature**2.0
pd3 = p4 * temperature**3.0
po2 = adc_p * (p1 + pd1 + pd2 + pd3)
pd1 = adc_p**2.0
pd2 = p9 + p10 * temperature
pd3 = pd1 * pd2
pd4 = pd3 + p11 * adc_p**3.0
pressure = (
po1 + po2 + pd4
) # - self.sea_level_pressure + STANDARD_SEA_LEVEL_PRESSURE_PA
return round(pressure, ROUNDING_PRECISION), round(
temperature, ROUNDING_PRECISION
)
[docs]
def _get_reg_temp_press_data(self) -> Tuple[int, int]:
"""
Function to get and return raw measurement data from the registers (uncompensated and calibrated pressure + temperature)
:return: Raw measurements data (pressure + temperature)
:rtype: Tuple[int, int]
"""
data: List[int] = self._read_reg(BMP3XX_P_DATA_PA, 6)
# 0x04 -> 0x06 included => Pressure data (24-bits data = 3 bytes)
# 0x07 -> 0x09 included => Temperature data (24-bits data = 3 bytes)
# a << b <==> a * 2**b
# data[2] << 2 * 2**3 | data[1] << 1 * 2**3 | data[0] << 0 * 2**3
adc_p: int = data[2] << 16 | data[1] << 8 | data[0]
adc_t: int = data[5] << 16 | data[4] << 8 | data[3]
# logger.info(adc_p)
# logger.info(adc_t)
return adc_p, adc_t
[docs]
def _get_sensor_time(self) -> int:
"""
Function to get and return sensor time from registers
:return: Sensor time data
:rtype: int
"""
data: List[int] = self._read_reg(reg=0x0C, length=3)
t: int = data[2] << 16 | data[1] << 8 | data[0]
return t
[docs]
def get_fifo_temp_press_data(self) -> Tuple[float, float]:
"""
Function to get and return cached data from the FIFO (pressure + temperature)
:return: Calibrated pressure + temperature data
:rtype: Tuple[float, float]
"""
data: List[int] = self._read_reg(
BMP3XX_FIFO_DATA, BMP3XX_FIFO_DATA_FRAME_LENGTH
)
adc_p: float | int = 0
adc_t: float | int = 0
if data[0] == BMP3_FIFO_TEMP_PRESS_FRAME:
adc_t = data[3] << 16 | data[2] << 8 | data[1]
adc_p = data[6] << 16 | data[5] << 8 | data[4]
elif data[0] == BMP3_FIFO_TEMP_FRAME:
adc_t = data[3] << 16 | data[2] << 8 | data[1]
elif data[0] == BMP3_FIFO_PRESS_FRAME:
adc_p = data[3] << 16 | data[2] << 8 | data[1]
elif data[0] == BMP3_FIFO_TIME_FRAME:
logger.info("FIFO time: %d" % (data[3] << 16 | data[2] << 8 | data[1]))
elif data[0] == BMP3_FIFO_CONFIG_CHANGE:
logger.warning("FIFO config change!!!")
else:
logger.warning("FIFO ERROR!!!")
# logger.info(data[0])
# logger.info(adc_p)
# logger.info(adc_t)
if adc_p > 0:
adc_p, adc_t = self._compensate_data(adc_p, adc_t)
return adc_p, adc_t
[docs]
def get_fifo_length(self) -> int:
"""
Function to return the FIFO cached data size
(range of return value is := (0-511))
:return: FIFO cached data size
:rtype: int
"""
logger.info("Getting FIFO length...")
len_: List[int] = self._read_reg(BMP3XX_FIFO_LENGTH, 2)
return len_[0] | (len_[1] << 8)
[docs]
def empty_fifo(self) -> None:
"""
Function to clear cached data in the FIFO without changing its settings
:return: Nothing
:rtype: None
"""
logger.info("Clearing cached data in FIFO...")
self._write_reg(BMP3XX_CMD, BMP3XX_CMD_FIFO_FLUSH)
sleep(Config().get("sensors", "bmp390", "safety_reg_write_sleep_period"))
[docs]
def reset(self) -> None:
"""
Function to reset and restart the sensor, then restoring the sensor configuration
to the default configuration
:return: Nothing
:rtype: None
"""
logger.info("Resetting/Restarting the sensor...")
self._write_reg(BMP3XX_CMD, BMP3XX_CMD_RESET)
[docs]
def _write_reg(self, reg: int, data: int | Sequence[int]) -> None:
"""
Function to write data to a register
:param reg: Reg address
:type reg: int
:param data: Data to write in the register
:type data: int | Sequence[int]
:return: Nothing
:rtype: None
"""
# Low level register writing, not implemented in base class
raise NotImplementedError()
[docs]
def _read_reg(self, reg: int, length: int) -> List[int]:
"""
Function to read data from a register
:param reg: Reg address
:type reg: int
:param length: Data length to read
:type length: int
:return: Read data list
:rtype: List[int]
"""
# Low level register writing, not implemented in base class
raise NotImplementedError()
[docs]
class BMP3XXI2C(BMP3XX):
"""
BMP3XXI2C class
Use the I2C protocol to drive the pressure sensor
"""
[docs]
def __init__(self, i2c_addr: int = 0x77, bus: int = 1):
"""
C-tor init function for BMP3XXI2C
:param i2c_addr: I2C device communication address
:type i2c_addr: int
:param bus: I2C bus number
:type bus: int
"""
try:
self._addr: int = i2c_addr
self.i2c: SMBus = SMBus(bus)
logger.info("Initializing a BMP3XXI2C instance...")
super(BMP3XXI2C, self).__init__() # Important call
except FileNotFoundError as exc_:
logger.fatal(f"I2C device not found on this host machine (error: {exc_})")
exit(I2C_DEVICE_NOT_FOUND_EXIT_CODE)
# Function overloading from base class
[docs]
def _write_reg(self, reg: int, data: int | Sequence[int]) -> None:
"""
Function to write data to a register
:param reg: Reg address
:type reg: int
:param data: Data to write in the register
:type data: int | Sequence[int]
:return: Nothing
:rtype: None
"""
if isinstance(data, int):
data = [data]
# logger.info(data)
logger.info(f"Writing data to register {reg} (data: {data})")
self.i2c.write_i2c_block_data(self._addr, reg, data)
# Function overloading from base class
[docs]
def _read_reg(self, reg: int, length: int) -> List[int]:
"""
Function to read data from a register
:param reg: Reg address
:type reg: int
:param length: Data length to read
:type length: int
:return: Read data list
:rtype: List[int]
"""
logger.info(f"Reading data from register {reg} (length: {length})")
return self.i2c.read_i2c_block_data(self._addr, reg, length)