製品レビュー｜電子機器８：光音響NDIR方式CO2センサ（SCD40）

[Terminal]
pip install rpi.gpio

[Terminal]
pip3 install adafruit-circuitpython-scd4x

[Terminal]
pip install smbus2

[IN]
from machine import Pin, I2C

i2c = I2C(0, sda=Pin(16), scl=Pin(17), freq=400000)
print(i2c.scan(), hex(i2c.scan()[0]))  # I2Cデバイスのスキャン

[OUT]
[98] 0x62

[IN]
from machine import Pin, I2C
import time
from micropython import const

class SCD4X:
    #I2Cアドレス
    DEFAULT_ADDRESS = 0x62 #SCD4xのI2Cアドレス
    #SCD4xコマンド
    ##低電力連続計測モード(Section3.8)
    DATA_READY = const(0xE4B8) #データ準備ステータス(Type:Read)/get_data_ready_status
    ##基本コマンド(Section3.5)
    STOP_PERIODIC_MEASUREMENT = const(0x3F86) #連続計測停止(Type:Send command)/stop_periodic_measurement
    START_PERIODIC_MEASUREMENT = const(0x21B1) #連続計測開始(Type:Send command)/start_periodic_measurement
    READ_MEASUREMENT = const(0xEC05) #計測値の読み出し(Type:Read)/read_measurement

    def __init__(self, i2c:I2C, address:int = DEFAULT_ADDRESS):
        self.i2c = i2c #I2Cオブジェクト
        self.address = address #I2Cアドレスを設定
        self._buffer = bytearray(18) #空Byte(18byte):バッファー(データ読み込み用)
        self._cmd = bytearray(2) #空Byte(2byte):コマンド送信用
        self._crc_buffer = bytearray(2) #空Byte(2byte):CRCチェック用

        # cached readings
        self._temperature = None
        self._relative_humidity = None
        self._co2 = None

        self.stop_periodic_measurement() #オブジェクト化時(初期)に連続計測を停止

    @property
    def co2(self):
        if self.data_ready:
            self._read_data()
        return self._co2

    @property
    def temperature(self):
        if self.data_ready:
            self._read_data()
        return self._temperature

    @property
    def relative_humidity(self):
        if self.data_ready:
            self._read_data()
        return self._relative_humidity

    def _read_data(self):
        self._send_command(self.READ_MEASUREMENT, cmd_delay=0.001)
        self._read_reply(self._buffer, 9)
        #CO2濃度データの取得
        self._co2 = (self._buffer[0] << 8) | self._buffer[1]
        #温度データの取得※要計算
        temp = (self._buffer[3] << 8) | self._buffer[4]
        self._temperature = -45 + 175 * (temp / 2 ** 16)
        #湿度データの取得※要計算
        humi = (self._buffer[6] << 8) | self._buffer[7]
        self._relative_humidity = 100 * (humi / 2 ** 16)

    @property
    def data_ready(self):
        self._send_command(self.DATA_READY, cmd_delay=0.001)
        self._read_reply(self._buffer, 3)
        return not ((self._buffer[0] & 0x03 == 0) and (self._buffer[1] == 0))

    def stop_periodic_measurement(self):
        self._send_command(self.STOP_PERIODIC_MEASUREMENT, cmd_delay=0.5)

    def start_periodic_measurement(self):
        self._send_command(self.START_PERIODIC_MEASUREMENT, cmd_delay=0.01)

    def _send_command(self, cmd, cmd_delay=0.00):
        self._cmd[0] = (cmd >> 8) & 0xFF
        self._cmd[1] = cmd & 0xFF  
        
        try:
            self.i2c.writeto(self.address, self._cmd)
        except OSError as e:
            raise RuntimeError("_send_commandメソッドでエラー：I2Cとの接続不良") from e
        time.sleep(cmd_delay)

    def _read_reply(self, buff, num):
        self.i2c.readfrom_into(self.address, buff, num)
        self._check_buffer_crc(self._buffer[0:num])

    def _check_buffer_crc(self, buf):
        for i in range(0, len(buf), 3):
            self._crc_buffer[0] = buf[i]
            self._crc_buffer[1] = buf[i + 1]
            if self._crc8(self._crc_buffer) != buf[i + 2]:
                raise RuntimeError("CRC check failed while reading data")

    #Cyclic Redundancy Check（巡回冗長検査）
    @staticmethod
    def _crc8(buffer):
        crc = 0xFF #CECレジスタの初期化
        for byte in buffer:
            crc ^= byte
            for _ in range(8):
                if crc & 0x80:
                    crc = (crc << 1) ^ 0x31
                else:
                    crc <<= 1
        return crc & 0xFF


if __name__ == "__main__":
    led = Pin("LED", Pin.OUT) #Pico：25, Pico W="LED"
    i2c = I2C(0, sda=Pin(0), scl=Pin(1), freq=100000)
    print(f'I2C接続確認：{[hex(i) for i in i2c.scan()]}')
    scd4x = SCD4X(i2c)
    scd4x.start_periodic_measurement() 
    
    while True:
        time.sleep(5)  # Delay for measurement interval
        if scd4x.co2 is not None:
            print(f'CO2濃度：{scd4x.co2:.2f}ppm')
            print(f'温度：{scd4x.temperature:.2f}℃')
            print(f'湿度：{scd4x.relative_humidity:.2f}% RH', end='\n\n')
        #CO2濃度が2000ppm以上の場合、LEDを点灯
        if scd4x.co2 and scd4x.co2 >= 2000:
            led.on()
        else:
            led.off()

[OUT]
I2C接続確認：['0x62']
CO2濃度：565.00ppm
温度：30.88℃
湿度：46.27% RH

CO2濃度：570.00ppm
温度：30.45℃
湿度：46.97% RH

CO2濃度：569.00ppm
温度：30.21℃
湿度：47.63% RH

[Terminal]
sudo i2cdetect -y 1

[IN]
import time
import board
import adafruit_scd4x

i2c = board.I2C()
scd4x = adafruit_scd4x.SCD4X(i2c)
print("Serial number:", [hex(i) for i in scd4x.serial_number])

scd4x.start_periodic_measurement()
print("Waiting for first measurement....")

while True:
    if scd4x.data_ready:
        print("CO2: %d ppm" % scd4x.CO2)
        print("Temperature: %0.1f *C" % scd4x.temperature)
        print("Humidity: %0.1f %%" % scd4x.relative_humidity, end="\n\n")
    time.sleep(1)

[OUT]
Serial number: ['0x63', '0x2', '0x8f', '0x7', '0x3b', '0x4a']
Waiting for first measurement....
CO2: 556 ppm
Temperature: 28.1 *C
Humidity: 49.7 %

CO2: 552 ppm
Temperature: 28.2 *C
Humidity: 50.2 %

CO2: 556 ppm
Temperature: 28.1 *C
Humidity: 50.6 %

[IN]
import time
import board
import adafruit_scd4x
import datetime
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from IPython.display import display, clear_output

#I2C測定準備
i2c = board.I2C()
scd4x = adafruit_scd4x.SCD4X(i2c)
print("Serial number:", [hex(i) for i in scd4x.serial_number])
scd4x.start_periodic_measurement()
print("Waiting for first measurement....")

#データ取得・可視化
class Datalist:
    def __init__(self):
        self.data = pd.DataFrame(columns=['Timestamp', 'CO2', 'Temperature', 'Humidity'])
        self.range_MINMAX = {'CO2': [400, 2000], 
                        'Temperature': [-10, 60], 
                        'Humidity': [0, 100]}
        self.color_graph = {'CO2': 'blue', 'Temperature': 'red', 'Humidity': 'green'}
        
    def append(self, timestamp, CO2, temperature, humidity):
        new_data = {'Timestamp': timestamp, 'CO2': CO2, 'Temperature': temperature, 'Humidity': humidity}
        df_new = pd.DataFrame(new_data, index=[0])
        self.data = pd.concat([self.data, df_new], ignore_index=True)
        
#データベース(データ格納用)の作成
datalist = Datalist()

#可視化用グラフ
fig, axs = plt.subplots(3, 1, figsize=(10, 10), sharex=True) #Sharex=Trueでx軸を共有

#測定開始
while True:
    if scd4x.data_ready:
        #データ取得
        timestamp = datetime.datetime.now()
        CO2 = scd4x.CO2
        temperature = scd4x.temperature
        humidity = scd4x.relative_humidity
        
        #データ格納
        datalist.append(timestamp, CO2, temperature, humidity)

        #データ可視化
        if not datalist.data.empty:
            #ax毎に描画
            for i, (ax, column) in enumerate(zip(axs, ['CO2', 'Temperature', 'Humidity'])):
                ax.clear() #グラフ初期化
                y_min, y_max = datalist.range_MINMAX[column] #y軸の範囲指定
                ax.plot(datalist.data['Timestamp'], datalist.data[column], label=f'{column}', color=datalist.color_graph[column])
                ax.set_xlabel("Time")
                ax.set_ylabel(f"{column}")
                ax.set_yticks(np.linspace(y_min, y_max, 11))
                ax.legend()
                ax.grid(True)
                ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
                
            # Jupyter Notebookでのリアルタイム表示のためのコード
            clear_output(wait=True)
            display(plt.gcf())
            
    time.sleep(1)

[OUT]
Serial number: ['0x63', '0x2', '0x8f', '0x7', '0x3b', '0x4a']
Waiting for first measurement....