aht21mod.py

# micropython
# MIT License
# Copyright (c) 2026 Roman Shevchik
"""MicroPython драйвер AHT21."""

from micropython import const
from collections import namedtuple
from sensor_pack_2.base_sensor import DeviceEx, IBaseSensorEx, IDentifier, check_value
from sensor_pack_2.bus_service import I2cAdapter
from time import sleep_ms, ticks_ms, ticks_diff
from sensor_pack_2.crc_mod import crc8


# приватные константы модуля
_AHT21_I2C_ADDR = const(0x38)

# команды (раздел 7.4 даташита)
_AHT21_CMD_STATUS = const(0x71)
_AHT21_CMD_TRIGGER = const(0xAC)
_AHT21_CMD_INIT = const(0xBE)
_AHT21_CMD_SOFT_RESET = const(0xBA)

# параметры команды запуска измерения (раздел 7.4)
_AHT21_MEAS_PARAM1 = const(0x33)
_AHT21_MEAS_PARAM2 = const(0x00)

# параметры команды инициализации (раздел 7.4)
_AHT21_INIT_PARAM1 = const(0x08)
_AHT21_INIT_PARAM2 = const(0x00)

# биты статуса
_AHT21_STATUS_BUSY = const(0x80)
_AHT21_STATUS_CALIBRATED = const(0x08)
_AHT21_STATUS_INIT_OK = const(0x18)

# времена ожидания
_AHT21_MEAS_TIME_MS = const(80)
_AHT21_POWER_ON_DELAY_MS = const(100)

# формулы преобразования (раздел 8.1, раздел 8.2)
_AHT21_RH_SCALE = const(1048576)  # 2^20
_AHT21_TEMP_SCALE = const(1048576)  # 2^20
_AHT21_TEMP_OFFSET = const(-50)

# CRC8
_AHT21_CRC_POLY = const(0x31)
_AHT21_CRC_INIT = const(0xFF)

# Named tuple для статуса
AHT21Status = namedtuple("AHT21Status", "busy calibrated raw")


def _decode_raw(raw: bytes) -> tuple:
    """Декодирование 6 байт в физические величины (раздел 8.1, раздел 8.2)"""
    # Влажность: байты [1][2][3], 20 бит
    srh: int = ((raw[1] << 16) | (raw[2] << 8) | raw[3]) >> 4
    rh: float = (srh * 100.0) / _AHT21_RH_SCALE

    # Температура: байты [3][4][5], 20 бит (младшие 4 бита байта[3])
    st: int = ((raw[3] & 0x0F) << 16) | (raw[4] << 8) | raw[5]
    t: float = (st * 200.0) / _AHT21_TEMP_SCALE + _AHT21_TEMP_OFFSET

    return t, rh


class AHT21(IBaseSensorEx, IDentifier):
    """Класс для управления датчиком AHT21 от Guangzhou Aosong Electronics Co., Ltd.,
    поставляется под торговой маркой ASAIR."""

    # Размеры буферов для _get_local_buf
    BUF_SIZE_CMD = const(3)
    BUF_SIZE_DATA = const(7)  # Статус(1) + Данные(6)

    INDEX_TEMPERATURE = const(0)
    INDEX_HUMIDITY = const(1)
    INDEX_BOTH  = None

    def __init__(self, adapter: I2cAdapter, address: int = _AHT21_I2C_ADDR, check_crc: bool = True) -> None:
        # объект DeviceEx для I2C обмена
        self._connection: DeviceEx = DeviceEx(adapter=adapter, address=address, big_byte_order=True)

        # self._config_fields = None
        # self._one_shot_mode_support: bool = True
        # self._resol_changeable: bool = False

        # локальные буферы нужного размера
        self._buf_1 = bytearray(1)
        self._buf_cmd = bytearray(self.BUF_SIZE_CMD)
        self._buf_data = bytearray(self.BUF_SIZE_DATA)

        # кэш и состояние
        self._cached_temp: float | None = None
        self._cached_rh: float | None = None
        self._data_ready: bool = False
        self._check_crc: bool = check_crc

        # задержка после включения питания (раздел 7.1 даташита)
        sleep_ms(_AHT21_POWER_ON_DELAY_MS)
        self._init_sensor()

    def _get_local_buf(self, bytes_for_read: int) -> bytearray | None:
        """Возвращает локальный буфер для операции чтения/записи."""
        check_value(value=bytes_for_read, valid_range=(0, 1, self.BUF_SIZE_CMD, self.BUF_SIZE_DATA),
                    error_msg=f"Invalid value for bytes_for_read: {bytes_for_read}")
        # if bytes_for_read not in (0, 1, self.BUF_SIZE_CMD, self.BUF_SIZE_DATA):
        #    raise ValueError(f"Invalid value for bytes_for_read: {bytes_for_read}")
        if not bytes_for_read:
            return None
        if 1 == bytes_for_read:
            return self._buf_1
        if self.BUF_SIZE_CMD == bytes_for_read:
            return self._buf_cmd
        return self._buf_data

    def _write_cmd(self, cmd: int, p1: int = 0, p2: int = 0) -> None:
        """Отправка команды датчику."""
        buf = self._get_local_buf(self.BUF_SIZE_CMD)
        buf[0] = cmd
        buf[1] = p1
        buf[2] = p2
        self._connection.write(buf)

    def _read_status_raw(self) -> int:
        """Чтение байта статуса (команда 0x71 -> чтение 1 байта)."""
        buf = self._get_local_buf(1)
        buf[0] = _AHT21_CMD_STATUS
        self._connection.write(buf)
        self._connection.read_to_buf(buf)
        return buf[0]

    def wait_busy_clear(self, timeout_ms: int = 200) -> bool:
        """Ожидание снятия флага Busy."""
        start: int = ticks_ms()
        while ticks_diff(ticks_ms(), start) < timeout_ms:
            if not self.get_data_status(raw=True) & _AHT21_STATUS_BUSY:
                return True
            sleep_ms(5)
        return False

    def _read_raw_data(self) -> bytes:
        """Чтение данных измерения (7 байт: статус + 5 данных + CRC)."""
        buf = self._get_local_buf(self.BUF_SIZE_DATA)
        # читаю 7 байт сразу
        self._connection.read_to_buf(buf)

        if self._check_crc:
            # CRC находится в 7-м байте (индекс 6)
            print(f"DBG: buf: {buf}")
            calculated_crc = crc8(buf[1:6], _AHT21_CRC_POLY, _AHT21_CRC_INIT)
            crc_from_buf = buf[6]
            if calculated_crc != crc_from_buf:
                raise ValueError(f"CRC mismatch! Calc: {calculated_crc:02X}, Recv: {crc_from_buf:02X}")

        return buf

    def _init_sensor(self) -> bool:
        """Инициализация и проверка калибровки (раздел 7.4)."""
        status: int = self._read_status_raw()
        if status != _AHT21_STATUS_INIT_OK:
            self._write_cmd(_AHT21_CMD_INIT, _AHT21_INIT_PARAM1, _AHT21_INIT_PARAM2)
            sleep_ms(10)
        return bool(self._read_status_raw() & _AHT21_STATUS_CALIBRATED)

    # Реализация интерфейс IBaseSensorEx
    def get_conversion_cycle_time(self) -> int:
        """Время преобразования в мс для текущих настроек.
        + 50 задаёт максимально допустимое время ожидания, равное = 130 мс.
        В реальной работе датчику может потребоваться чуть больше времени из-за:
            * задержек шины I²C и обработки ACK/NACK
            * джиттера планировщика MicroPython и дискретности ticks_ms() (обычно 1–10 мс)
            * внутренних вариаций времени АЦП при разных температурах/влажности
            * возможных кратковременных пауз на шине от других задач в RTOS.
        Используй self.get_data_status(raw=True) & _AHT21_STATUS_BUSY для проверки наличия!
        Если флаг в 1, то датчик занят преобразованием!
        Можете обнулить добавку и возвращать только _AHT21_MEAS_TIME_MS,
        но тогда используйте метод wait_busy_clear"""
        return 50 + _AHT21_MEAS_TIME_MS

    def start_measurement(self) -> None:
        """Настраивает параметры датчика и запускает процесс измерения."""
        self._write_cmd(_AHT21_CMD_TRIGGER, _AHT21_MEAS_PARAM1, _AHT21_MEAS_PARAM2)
        self._data_ready = False

    def get_measurement_value(self, value_index: int | None) -> float | tuple:
        """Возвращает измеренное значение по индексу."""
        if not self._data_ready:
            buf_raw = self._read_raw_data()
            self._cached_temp, self._cached_rh = _decode_raw(buf_raw)
            self._data_ready = True

        if value_index == self.INDEX_TEMPERATURE:
            return self._cached_temp
        if value_index == self.INDEX_HUMIDITY:
            return self._cached_rh
        return self._cached_temp, self._cached_rh

    def get_data_status(self, raw: bool = True) -> int | AHT21Status:
        """Возвращает состояние готовности данных.
        При raw=False возвращает namedtuple: AHT21Status(busy, calibrated, raw)."""
        status: int = self._read_status_raw()
        if raw:
            return status
        return AHT21Status(
            busy=bool(status & _AHT21_STATUS_BUSY),
            calibrated=bool(status & _AHT21_STATUS_CALIBRATED),
            raw=status
        )

    def is_single_shot_mode(self) -> bool:
        return True

    def is_continuously_mode(self) -> bool:
        return False

    # интерфейс IDentifier
    def soft_reset(self) -> None:
        """Программный сброс устройства (раздел 7.4)"""
        buf = self._get_local_buf(1)
        buf[0] = _AHT21_CMD_SOFT_RESET
        self._connection.write(buf)
        sleep_ms(20)
        self._cached_temp = self._cached_rh = None
        self._data_ready = False

    # Iterator
    # Поскольку разработчики не завезли автоматический режим работы датчика, у меня нет для вас итератора :-)