main.py

import network
import time
import machine
import ujson
import urequests

# -------------------------
# Configuration
# -------------------------

WIFI_SSID = "your wifi ssid"
WIFI_PASSWORD = "your wifi password"

POST_URL = "http://192.168.50.246:8080/api/scd41"

# Default Pico I2C0 pins
I2C_SDA = 0
I2C_SCL = 1

SCD41_ADDR = 0x62

# Per datasheet section 3.11: CRC-8, poly 0x31, init 0xFF,
# no reflection, final XOR 0x00. Reference: CRC(0xBEEF) == 0x92.
CRC8_POLYNOMIAL = 0x31
CRC8_INIT = 0xFF

# How many times to re-poll the sensor within one cycle before
# giving up. Periodic mode yields a fresh sample every ~5 s.
MAX_READ_ATTEMPTS = 5

# Forced recalibration (FRC). Set RUN_FRC_ON_START = True to run a
# one-time recalibration at startup, then set it back to False.
# During the whole procedure the sensor MUST be in stable, known
# reference air (e.g. fresh outdoor air ~420 ppm).
RUN_FRC_ON_START = False
FRC_TARGET_PPM = 420
FRC_WARMUP_S = 360  # datasheet requires > 3 minutes of periodic operation


def crc8(data):
    crc = CRC8_INIT
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 0x80:
                crc = ((crc << 1) ^ CRC8_POLYNOMIAL) & 0xFF
            else:
                crc = (crc << 1) & 0xFF
    return crc

# -------------------------
# WiFi
# -------------------------

def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)

    if wlan.isconnected():
        return wlan

    print("Connecting to WiFi...")

    wlan.connect(WIFI_SSID, WIFI_PASSWORD)

    timeout = 20
    while timeout > 0:
        if wlan.isconnected():
            break
        timeout -= 1
        print(".", end="")
        time.sleep(1)

    print()

    if not wlan.isconnected():
        raise RuntimeError("WiFi connection failed")

    print("Connected")
    print("IP:", wlan.ifconfig()[0])

    return wlan

# -------------------------
# SCD41 Driver (minimal)
# -------------------------

class SCD41:

    def __init__(self, i2c):
        self.i2c = i2c

    def _write_cmd(self, cmd):
        buf = bytearray([
            cmd >> 8,
            cmd & 0xff
        ])
        self.i2c.writeto(SCD41_ADDR, buf)

    def _write_cmd_arg(self, cmd, value):
        # Command word, followed by a data word and its CRC byte.
        buf = bytearray([
            cmd >> 8,
            cmd & 0xff,
            value >> 8,
            value & 0xff,
        ])
        buf.append(crc8(buf[2:4]))
        self.i2c.writeto(SCD41_ADDR, buf)

    def _read(self, length):
        return self.i2c.readfrom(SCD41_ADDR, length)

    def stop_periodic_measurement(self):
        # Return the sensor to idle so settings can be read/changed.
        self._write_cmd(0x3F86)
        time.sleep_ms(500)

    def get_asc_enabled(self):
        # Must be in idle mode. Returns True if ASC is enabled.
        self._write_cmd(0x2313)
        time.sleep_ms(2)

        d = self._read(3)

        if crc8(d[0:2]) != d[2]:
            raise ValueError("CRC check failed")

        return ((d[0] << 8) | d[1]) == 1

    def set_asc_enabled(self, enabled):
        # Must be in idle mode. Change is in RAM until persist_settings.
        self._write_cmd_arg(0x2416, 1 if enabled else 0)
        time.sleep_ms(2)

    def persist_settings(self):
        # Store current config to EEPROM (~800 ms). Use sparingly:
        # EEPROM is rated for ~2000 write cycles.
        self._write_cmd(0x3615)
        time.sleep_ms(800)

    def perform_forced_recalibration(self, target_ppm):
        # Must be in idle mode, after > 3 min of periodic operation in a
        # stable, known-CO2 environment (see datasheet section 3.7.1).
        # Returns the applied correction in ppm; raises on failure.
        self._write_cmd_arg(0x362F, target_ppm)
        time.sleep_ms(400)

        d = self._read(3)

        if crc8(d[0:2]) != d[2]:
            raise ValueError("CRC check failed")

        word = (d[0] << 8) | d[1]

        if word == 0xFFFF:
            raise RuntimeError("Forced recalibration failed")

        return word - 0x8000

    def start_periodic_measurement(self):
        self._write_cmd(0x21B1)

    def data_ready(self):
        self._write_cmd(0xE4B8)
        time.sleep_ms(2)

        d = self._read(3)

        value = (d[0] << 8) | d[1]

        return (value & 0x07FF) != 0

    def read_measurement(self):
        self._write_cmd(0xEC05)
        time.sleep_ms(5)

        data = self._read(9)

        # Each 2-byte word is followed by its CRC byte. Validate all
        # three words so a corrupted frame is rejected rather than
        # silently used (see datasheet section 3.11).
        for offset in (0, 3, 6):
            word = data[offset:offset + 2]
            if crc8(word) != data[offset + 2]:
                raise ValueError("CRC check failed")

        co2 = (data[0] << 8) | data[1]

        temp_raw = (data[3] << 8) | data[4]
        rh_raw = (data[6] << 8) | data[7]

        temperature = -45 + 175 * temp_raw / 65535
        humidity = 100 * rh_raw / 65535

        return {
            "co2": co2,
            "temperature": round(temperature, 2),
            "humidity": round(humidity, 2),
        }

# -------------------------
# Forced recalibration routine
# -------------------------

def run_forced_recalibration(sensor, target_ppm):
    # Full documented FRC sequence (datasheet section 3.7.1):
    #   1. run periodic measurement > 3 min in stable, constant CO2 air
    #   2. stop periodic measurement, wait 500 ms
    #   3. issue perform_forced_recalibration and read the correction
    print("FRC: keep the sensor in stable reference air (~%d ppm)." % target_ppm)
    print("FRC: warming up in periodic mode for %d s..." % FRC_WARMUP_S)

    sensor.start_periodic_measurement()
    time.sleep(FRC_WARMUP_S)

    sensor.stop_periodic_measurement()  # includes the required 500 ms wait

    correction = sensor.perform_forced_recalibration(target_ppm)
    print("FRC complete. Applied correction: %d ppm" % correction)

    return correction

# -------------------------
# Main
# -------------------------

connect_wifi()

i2c = machine.I2C(
    0,
    scl=machine.Pin(I2C_SCL),
    sda=machine.Pin(I2C_SDA),
    freq=100000
)

sensor = SCD41(i2c)

# ASC (and other settings) can only be read/changed in idle mode, so
# make sure periodic measurement is stopped first.
sensor.stop_periodic_measurement()

asc_enabled = sensor.get_asc_enabled()
print("Automatic self-calibration (ASC):", "enabled" if asc_enabled else "disabled")

if not asc_enabled:
    print("Enabling ASC and persisting to EEPROM...")
    sensor.set_asc_enabled(True)
    sensor.persist_settings()
    print("ASC now:", "enabled" if sensor.get_asc_enabled() else "disabled")

if RUN_FRC_ON_START:
    run_forced_recalibration(sensor, FRC_TARGET_PPM)

print("Starting SCD41...")
sensor.start_periodic_measurement()

print("Waiting for first measurement...")
time.sleep(5)

def acquire_reading():
    # Try a few times within a cycle to get a valid, non-zero sample.
    # co2 == 0 is the sensor's sentinel for an invalid CO2 measurement
    # (e.g. drafts near an open window) even when T/RH are still good.
    for attempt in range(MAX_READ_ATTEMPTS):
        if sensor.data_ready():
            reading = sensor.read_measurement()

            if reading["co2"] != 0:
                return reading

            print("Discarding invalid CO2=0 reading")
        else:
            print("Measurement not ready")

        # A fresh sample is produced roughly every 5 s in periodic mode.
        time.sleep(5)

    return None


while True:

    try:

        reading = acquire_reading()

        if reading is not None:

            print(reading)

            response = urequests.post(
                POST_URL,
                headers={
                    "Content-Type": "application/json"
                },
                data=ujson.dumps(reading)
            )

            print("HTTP:", response.status_code)
            response.close()

        else:
            print("No valid reading this cycle")

    except Exception as e:
        print("Error:", e)

    time.sleep(60)