"""Logic to handle and manage the STCC4 sensor"""
from time import sleep
from typing import Dict, List, Optional, Tuple
from smbus2 import SMBus
from pi_weather.core.sensors.stcc4.addresses import (
STCC4_DISABLE_TESTING_MODE,
STCC4_ENABLE_TESTING_MODE,
STCC4_ENTER_SLEEP_MODE,
STCC4_EXIT_SLEEP_MODE, # := WAKEUP operation
STCC4_GET_PRODUCT_ID,
STCC4_MEASURE_SINGLE_SHOT,
STCC4_PERFORM_FACTORY_RESET,
STCC4_PERFORM_FORCED_RECALIBRATION,
STCC4_PERFORM_SOFT_RESET,
STCC4_READ_MEASUREMENT,
STCC4_SET_PRESSURE_COMPENSATION,
STCC4_SET_RHT_COMPENSATION,
STCC4_START_CONTINUOUS_MEASUREMENT,
STCC4_STOP_CONTINUOUS_MEASUREMENT,
)
from pi_weather.core.sensors.stcc4.constants import (
COMPENSATION_HUMIDITY_MAX,
COMPENSATION_HUMIDITY_MIN,
COMPENSATION_PRESSURE_MAX,
COMPENSATION_PRESSURE_MIN,
COMPENSATION_TEMP_MAX,
COMPENSATION_TEMP_MIN,
RECALIB_TARGET_PPM_MAX,
RECALIB_TARGET_PPM_MIN,
SENSOR_SAMPLING_PERIOD,
)
from pi_weather.utils.logs import get_logger
logger = get_logger(__name__)
[docs]
class STCC4:
"""Base class for STCC4 CO2 sensor"""
[docs]
def __init__(self) -> None:
"""
C-tor init function
"""
...
[docs]
def calculation_crc(self, data: List | Tuple) -> int:
"""
Calculate the CRC (Cyclic Redundancy Check) for input data
Data words are succeeded by an 8-bit checksum (denoted as Cyclic Redundancy Check (CRC))
:param data: List or tuple of 16-bit integers
:return: Calculated CRC value
"""
crc = 0xFF
for value in data:
high_byte = (value >> 8) & 0xFF
low_byte = value & 0xFF
# Process high byte
crc ^= high_byte
for _ in range(8):
if crc & 0x80:
crc = (crc << 1) ^ 0x31
else:
crc <<= 1
crc &= 0xFF # Keep it as 8-bit
# Process low byte
crc ^= low_byte
for _ in range(8):
if crc & 0x80:
crc = (crc << 1) ^ 0x31
else:
crc <<= 1
crc &= 0xFF # Keep it as 8-bit
return crc
[docs]
def get_product_id(self) -> int: # Optional[bytes]:
"""
Get the sensor ID
:return: Sensor ID as bytes if successful, None otherwise
"""
raise NotImplementedError
[docs]
def start_measurement(self) -> bool:
"""
Start continuous measurement
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def stop_measurement(self) -> bool:
"""
Stop continuous measurement
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def get_measurement(
self,
) -> Optional[Dict[str, int | float]]: # Optional[Tuple[int, float, float, int]]:
"""
Read measurement data
:return: Tuple of (co2_concentration, temperature, humidity, sensor_status).
:rtype: Optional[Dict[str, int | float]]
"""
raise NotImplementedError
[docs]
def set_rht_compensation(self, temperature: int, humidity: int) -> bool:
"""
Set temperature and humidity compensation
:param temperature: Temperature compensation value, range of 10 to 40 degrees Celsius.
:param humidity: Humidity compensation value, range of 20 to 80 percent relative humidity.
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def set_pressure_compensation(self, pressure: int) -> bool:
"""
Set pressure compensation
:param pressure: Pressure compensation value, range of 400 to 1100 hPa
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def single_measurement(self) -> bool:
"""
Perform a single shot measurement
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def fall_asleep(self) -> bool:
"""
Put the sensor into sleep mode
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def wakeup(self) -> bool:
"""
Wake up the sensor from sleep mode
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def soft_reset(self) -> bool:
"""
Perform a soft reset of the sensor
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def factory_reset(self) -> bool:
"""
Perform a factory reset of the sensor
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def enable_testing_mode(self) -> bool:
"""
Enable testing mode
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def disable_testing_mode(self) -> bool:
"""
Disable testing mode
:return: True if successful, False otherwise
"""
raise NotImplementedError
[docs]
def forced_recalibration(self, target_ppm: int) -> Optional[int]:
"""
Perform forced recalibration
:param target_ppm: Target PPM value for recalibration, must be between 0 and 32000 ppm.
:return: Correction value if successful, None otherwise
"""
raise NotImplementedError
[docs]
class STCC4I2C(STCC4):
"""
STCC4I2C class
Use the I2C protocol to drive the STCC4 sensor
"""
[docs]
def __init__(self, i2c_addr: int = 0x64, bus: int = 3) -> None:
"""
C-tor init function
:param i2c_addr: I2C device address
:type i2c_addr: int
:param bus: I2C bus identifier
:type bus: int
"""
super().__init__()
self.i2c_address: int = i2c_addr
self.i2c_bus: int = bus
try:
logger.info(f"Initializing I2C connection bus #{self.i2c_bus}")
self._bus = SMBus(bus=self.i2c_bus)
except Exception as exc_:
logger.error(
f"An error occurred while trying to init a I2C bus connection: {exc_}"
)
self._bus = None
[docs]
def _write_cmd16(self, cmd: int) -> bool:
"""
Write a 16-bit command to the sensor
:param cmd: Command to write
:return: True if successful, False otherwise
"""
if self._bus is None:
logger.error("I2C bus is invalid.")
return False
try:
# Split 16-bit command into two bytes (big endian)
bytes_to_send = [(cmd >> 8) & 0xFF, cmd & 0xFF]
self._bus.write_i2c_block_data(
self.i2c_address, bytes_to_send[0], [bytes_to_send[1]]
)
return True
except Exception as exc_:
logger.error(
f"An error occurred when trying to write 16-bit command: {exc_}"
)
return False
[docs]
def _write_cmd8(self, cmd: int) -> bool:
"""
Write an 8-bit command to the sensor
:param cmd: Command to write
:return: True if successful, False otherwise
"""
if self._bus is None:
logger.error("I2C bus is invalid.")
return False
try:
self._bus.write_byte(self.i2c_address, cmd)
return True
except Exception as exc_:
logger.error(
f"An error occurred when trying to write 8-bit command: {exc_}"
)
return False
[docs]
def _write_data(self, cmd: int, data: List | Tuple) -> bool:
"""
Write data to the sensor
:param cmd: Command to write before data
:param data: List or tuple of 16-bit integers to write
:return: True if successful, False otherwise
"""
logger.info(f"Reg:= {cmd} \t Data:= {data}")
if not self._write_cmd16(cmd):
return False
try:
for value in data:
# Split value into two bytes
high_byte = (value >> 8) & 0xFF
low_byte = value & 0xFF
# Calculate CRC for this value
crc = self.calculation_crc([value])
# Write data bytes and CRC
# self._bus.write_i2c_block_data(
# self.i2c_address,
# high_byte,
# [low_byte, crc]
# )
self._bus.write_byte(self.i2c_address, high_byte)
self._bus.write_byte(self.i2c_address, low_byte)
self._bus.write_byte(self.i2c_address, crc)
# Small delay between writes
sleep(0.01)
return True
except Exception as exc_:
logger.error(
f"An error occurred when trying to write data (reg:= {cmd}, data to write:= {data}): {exc_}"
)
return False
[docs]
def _read_data(self, cmd: int, length: int) -> Optional[bytes]:
"""
Read data from the sensor
:param cmd: Command to write before reading
:param length: Number of bytes to read
:return: Read bytes if successful, None otherwise
"""
if self._bus is None:
logger.error("I2C bus is invalid.")
return None
if not self._write_cmd16(cmd):
return None
try:
# Read the data
data = self._bus.read_i2c_block_data(self.i2c_address, 0, length)
return bytes(data)
except Exception as exc_:
logger.error(
f"An error occurred when trying to read data (reg:= {cmd}, length:= {length}): {exc_}"
)
return None
[docs]
def get_product_id(self) -> int:
"""
Function retrieving and returning the sensor product ID
:return: Sensor product ID
:rtype: int
"""
logger.info("Getting sensor product ID")
for i in range(5):
r_buf = self._read_data(STCC4_GET_PRODUCT_ID, 18)
if r_buf is None or len(r_buf) < 18:
sleep(0.2)
continue
# Extract data pairs and their CRCs
data1 = (r_buf[0] << 8) | r_buf[1]
data2 = (r_buf[3] << 8) | r_buf[4]
crc1 = r_buf[2]
crc2 = r_buf[5]
# Calculate CRCs
calculated_crc1 = self.calculation_crc([data1])
calculated_crc2 = self.calculation_crc([data2])
# Check CRCs
if crc1 == calculated_crc1 and crc2 == calculated_crc2:
id_value = (
(r_buf[0] << 24) | (r_buf[1] << 16) | (r_buf[3] << 8) | r_buf[4]
)
if id_value == 0x901018A:
return id_value
sleep(0.2)
return 0
[docs]
def start_measurement(self) -> bool:
"""
Function to start measurements
Starts continuous measurement with 1s sampling interval.
Note that the sensor and the microcontroller are subject to clock tolerances.
The effective sampling interval is 1s +/- 150 ms.
:return: True if successful, False otherwise
:rtype: bool
"""
if not self._write_cmd16(STCC4_START_CONTINUOUS_MEASUREMENT):
logger.error("Failed to start continuous measurements")
return False
sleep(1)
logger.info("Starting continuous measurements...")
return True
[docs]
def stop_measurement(self) -> bool:
"""
Function to stop continuous measurements
The `stop_continuous_measurement` command will finish the currently running measurement
before returning to idle mode.
During the execution time, the sensor will not acknowledge its I2C address nor accept commands.
:return: True if successful, False otherwise
:rtype: bool
"""
if not self._write_cmd16(STCC4_STOP_CONTINUOUS_MEASUREMENT):
logger.error("Failed to stop continuous measurements")
return False
sleep(1)
logger.info("Stopping continuous measurements...")
return True
[docs]
def get_measurement(
self, force_sampling_sleep: bool = False
) -> Optional[Dict[str, int | float]]: # Optional[Tuple[int, float, float, int]]:
"""
Function to get instantaneous measurements
:param force_sampling_sleep: Bool to force a sampling sleep (see constants.py)
:type force_sampling_sleep: bool
:return: Tuple with CO2 concentration, temperature, humidity and sensor status
:rtype: Optional[Dict[str, int | float]]
"""
if force_sampling_sleep:
logger.info("Sleeping during sampling interval...")
sleep(SENSOR_SAMPLING_PERIOD)
logger.info("Reading measurement...")
raw_data = self._read_data(STCC4_READ_MEASUREMENT, 12)
sleep(0.2) # to avoid: IOError: [Errno 121] Remote I/O error
if raw_data is None or len(raw_data) < 12:
logger.error("Failed to read measurement")
return None
# Parsing CO2 concentration
co2_concentration = (raw_data[0] << 8) | raw_data[1]
# Parsing temperature (raw value to °C)
temp_raw = (raw_data[3] << 8) | raw_data[4]
temperature = -45.0 + ((175.0 * temp_raw) / 65535.0)
# Parsing humidity (raw value to %RH)
hum_raw = (raw_data[6] << 8) | raw_data[7]
humidity = -6.0 + ((125.0 * hum_raw) / 65535.0)
# Parsing sensor status
sensor_status = (raw_data[9] << 8) | raw_data[10]
return {
"co2_concentration": co2_concentration,
"temperature": temperature,
"humidity": humidity,
"sensor_status": sensor_status,
}
# (co2_concentration, temperature, humidity, sensor_status)
[docs]
def set_rht_compensation(self, temperature: float, humidity: float) -> bool:
"""
Function to set relative humidity + temperature compensation values
:param temperature: External temperature (T) compensation value
:type temperature: float
:param humidity: External relative humidity (RH) compensation value
:type humidity: float
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info(
f"Setting RHT compensation (RH := {humidity}, T := {temperature}) ..."
)
if (
temperature < COMPENSATION_TEMP_MIN or temperature > COMPENSATION_TEMP_MAX
): # or humidity < 20 or humidity > 80:
logger.error("Temperature compensation must be between 10 and 40")
return False
if humidity < COMPENSATION_HUMIDITY_MIN or humidity > COMPENSATION_HUMIDITY_MAX:
logger.error("Humidity compensation must be between 20 and 80.")
return False
# Convert temperature to raw value
temp_raw = int((temperature + 45) * 65535 / 175)
# Convert humidity to raw value
hum_raw = int((humidity + 6) * 65535 / 125)
return self._write_data(STCC4_SET_RHT_COMPENSATION, [temp_raw, hum_raw])
[docs]
def set_pressure_compensation(self, pressure: int) -> bool:
"""
Function to set pressure compensation to STCC4 sensor
:param pressure: External pressure (P) compensation value (int in hPa)
:type pressure: int
:return: True if command was successful, False otherwise
:rtype: bool
"""
# Limits in hPa
if pressure < COMPENSATION_PRESSURE_MIN or pressure > COMPENSATION_PRESSURE_MAX:
logger.error("Pressure compensation must be between 400 and 1100 hPa")
return False
pressure_raw = pressure * 50
return self._write_data(STCC4_SET_PRESSURE_COMPENSATION, [pressure_raw])
[docs]
def single_measurement(self) -> bool:
"""
Function to perform a single-short measurement
The `measure_single_shot` command conducts an on-demand measurement of CO2 gas concentration.
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info("Performing a single shot measurement...")
return self._write_cmd16(STCC4_MEASURE_SINGLE_SHOT)
[docs]
def fall_asleep(self) -> bool:
"""
Function to make the sensor fall asleep
The `enter_sleep_mode` command sets the sensor from idle to sleep mode through the I2C interface.
The written relative humidity, temperature, pressure compensation values and
ASC state are retained while in sleep mode.
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info("Performing a fall asleep...")
return self._write_cmd16(STCC4_ENTER_SLEEP_MODE)
[docs]
def wakeup(self) -> bool:
"""
Function to wake up the sensor
The `exit_sleep_mode` command wakes the sensor up from sleep mode to idle mode upon receiving
its I2C address, a write data direction bit and a payload byte 0x00.
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info("Waking up the sensor...")
return self._write_cmd8(STCC4_EXIT_SLEEP_MODE)
[docs]
def soft_reset(self) -> bool:
"""
Function to perform a soft reset
The `perform_soft_reset` command triggers a soft reset of the sensor through
the I2C general call reset as implemented according to the NXP I2C-bus specification and user manual
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info("Performing a soft reset...")
return self._write_cmd8(STCC4_PERFORM_SOFT_RESET)
[docs]
def factory_reset(self) -> bool:
"""
Function to perform a factory reset
The `perform_factory_reset` command resets the FRC and ASC algorithm history
:return: True if command was successful, False otherwise
:rtype: bool
"""
logger.info("Performing a factory reset...")
if not self._write_cmd16(STCC4_PERFORM_FACTORY_RESET):
logger.error("Failed to perform a factory reset on sensor")
return False
raw_data = self._read_data(STCC4_PERFORM_FACTORY_RESET, 2)
if raw_data is None or len(raw_data) < 2:
return False
response = (raw_data[0] << 8) | raw_data[1]
return response == 0
[docs]
def enable_testing_mode(self) -> bool:
"""
Function to enable testing mode
:return: Nothing
:rtype: None
"""
logger.info("Enabling the testing mode of the sensor...")
return self._write_cmd16(STCC4_ENABLE_TESTING_MODE)
[docs]
def disable_testing_mode(self) -> bool:
"""
Function to disable testing mode
:return: Nothing
:rtype: None
"""
logger.info("Disabling the testing mode of the sensor...")
return self._write_cmd16(STCC4_DISABLE_TESTING_MODE)
[docs]
def forced_recalibration(self, target_ppm: int) -> Optional[int]:
"""
Function to forced recalibration
The perform_forced_recalibration command (FRC) is used to correct the sensor’s CO2 concentration
output with an externally provided target CO2 concentration.
Ensure the sensor reading and environmental conditions, including CO2 concentration,
are stable for the duration of the recommended operation sequence.
See Section 3.5 for the signal output and input conversion.
:param target_ppm: Target PPM
:type target_ppm: int
:return: Applied FRC correction
:rtype: int
"""
if target_ppm < RECALIB_TARGET_PPM_MIN or target_ppm > RECALIB_TARGET_PPM_MAX:
logger.error("Forced recalibration target PPM is out of range (0-32000)")
return None
if not self._write_data(STCC4_PERFORM_FORCED_RECALIBRATION, [target_ppm]):
return None
sleep(0.2)
# Reading the applied FRC correction
raw_data = self._read_data(STCC4_PERFORM_FORCED_RECALIBRATION, 3)
if raw_data is None or len(raw_data) < 3:
return None
frc_correction = (raw_data[0] << 8) | raw_data[1]
return frc_correction