dzxj_dtu/backend/modules/dht11_sensor.py

"""
DHT11 温湿度传感器本地读取模块

通过 libgpiod 操作 GPIO，定时读取 DHT11 数据并上报到应用层。
支持真实传感器读取和模拟模式（用于无硬件环境测试）。
"""

import os
import time
import json
import logging
import threading
import subprocess

logger = logging.getLogger('dht11_sensor')

# 尝试导入 gpiod；如果缺失，真实读取模式会不可用，但模拟模式仍可运行
try:
    import gpiod
    _GPIOD_AVAILABLE = True
    # 简单判断 gpiod 1.x/2.x：1.x 有 Chip/get_line，2.x 没有 get_line
    _GPIOD_V1 = hasattr(gpiod, 'Chip') and hasattr(gpiod.Chip, 'get_line')
    _GPIOD_V2 = hasattr(gpiod, 'Chip') and not _GPIOD_V1
    if _GPIOD_V2:
        logger.warning("检测到 gpiod 2.x，当前 DHT11 驱动使用 gpiod 1.x API；"
                       "请在目标设备使用系统包管理器安装 python3-libgpiod (1.x)")
        _GPIOD_AVAILABLE = False
except Exception as _e:  # pragma: no cover
    gpiod = None
    _GPIOD_AVAILABLE = False
    logger.warning(f"gpiod 库未安装或加载失败，DHT11 真实读取不可用: {_e}")


class DHT11Sensor:
    """
    DHT11 传感器读取器。

    参数:
        gpio_pin: GPIO line 编号（在 gpiochip0 上）。
                  例如 Rockchip 的 GPIO2_A0 通常对应 sysfs 编号 64，
                  但具体编号取决于内核 gpio 描述。请按实际接线配置。
        gpio_chip: GPIO chip 名称，默认 'gpiochip0'。
        poll_interval: 读取间隔（秒），默认 5。
        simulate: 是否使用模拟数据（无真实硬件时测试用）。
        on_data: 数据回调函数，签名 on_data(temperature, humidity) -> None。
    """

    def __init__(self, gpio_pin=None, gpio_chip='gpiochip0', poll_interval=5,
                 simulate=False, on_data=None):
        self.gpio_pin = gpio_pin
        self.gpio_chip = gpio_chip
        self.poll_interval = poll_interval
        self.simulate = simulate
        self.on_data = on_data
        self._running = False
        self._thread = None
        self._last_error = None
        self._last_success_time = None

    def start(self):
        """启动后台读取线程。"""
        if self._running:
            return
        self._running = True
        self._thread = threading.Thread(target=self._read_loop, daemon=True)
        self._thread.start()
        mode = '模拟' if self.simulate else '真实'
        logger.info(f"DHT11 读取线程已启动 (模式={mode}, pin={self.gpio_pin}, chip={self.gpio_chip}, 间隔={self.poll_interval}s)")

    def stop(self):
        """停止后台读取线程。"""
        self._running = False
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=2)
        logger.info("DHT11 读取线程已停止")

    def get_status(self):
        """返回当前传感器状态。"""
        return {
            'running': self._running,
            'simulate': self.simulate,
            'gpio_pin': self.gpio_pin,
            'gpio_chip': self.gpio_chip,
            'poll_interval': self.poll_interval,
            'last_success_time': self._last_success_time,
            'last_error': self._last_error
        }

    def _read_loop(self):
        """后台循环读取。"""
        # 首次启动稍等片刻，让系统其他初始化完成
        time.sleep(1)
        while self._running:
            try:
                if self.simulate:
                    temperature, humidity = self._read_simulated()
                else:
                    temperature, humidity = self._read_real()

                self._last_success_time = time.strftime('%Y-%m-%d %H:%M:%S')
                self._last_error = None

                if self.on_data:
                    try:
                        self.on_data(temperature, humidity)
                    except Exception as cb_err:
                        logger.error(f"DHT11 数据回调异常: {cb_err}")
                else:
                    logger.info(f"DHT11 读取成功: 温度={temperature}°C, 湿度={humidity}%")
            except Exception as e:
                self._last_error = str(e)
                logger.warning(f"DHT11 读取失败: {e}")

            # 按间隔休眠，拆分成小段以便快速退出
            for _ in range(int(self.poll_interval * 2)):
                if not self._running:
                    break
                time.sleep(0.5)

    def _read_simulated(self):
        """生成缓慢变化的模拟数据。"""
        t = time.time()
        # 温度在 20~30 度之间缓慢波动，湿度在 40~70% 之间缓慢波动
        temperature = round(25 + 4 * ((t % 60) / 60 - 0.5), 1)
        humidity = round(55 + 14 * ((t % 120) / 120 - 0.5), 1)
        return temperature, humidity

    def _reader_binary_path(self):
        """返回 C 语言 DHT11 读取器可执行文件路径。"""
        # 优先使用与模块同目录下的 scripts/dht11_reader
        backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        candidates = [
            os.path.join(backend_dir, 'scripts', 'dht11_reader'),
            '/root/dzxj_dtu/backend/scripts/dht11_reader',
            '/usr/local/bin/dht11_reader',
        ]
        for p in candidates:
            if os.path.isfile(p) and os.access(p, os.X_OK):
                return p
        return None

    def _read_with_binary(self):
        """
        调用 C 语言读取器获取温湿度。
        该二进制使用 open-drain + 高速采样，可避开 Python 调度抖动。
        """
        binary = self._reader_binary_path()
        if not binary:
            raise RuntimeError("DHT11 C 读取器未找到，请确保 dht11_reader 已编译")

        pin = int(self.gpio_pin)
        chip = str(self.gpio_chip)
        cmd = [binary, chip, str(pin)]

        proc = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=10
        )

        stderr_text = proc.stderr.strip()
        if stderr_text:
            for line in stderr_text.splitlines():
                logger.debug(f"dht11_reader: {line}")

        if proc.returncode != 0:
            raise RuntimeError(f"dht11_reader 退出码 {proc.returncode}")

        try:
            result = json.loads(proc.stdout.strip().splitlines()[-1])
        except Exception as e:
            raise RuntimeError(f"解析 dht11_reader 输出失败: {e}, stdout={proc.stdout!r}")

        if not result.get('valid'):
            raise RuntimeError(f"DHT11 校验和错误: raw={result.get('raw')}")

        return float(result['temperature']), float(result['humidity'])

    def _read_real(self):
        """
        读取一次 DHT11。

        优先使用 C 语言读取器（时序更稳定），不可用时回退到纯 Python gpiod。
        """
        if self.gpio_pin is None:
            raise ValueError("未配置 DHT11 GPIO pin 编号")

        # 优先使用 C 二进制读取器
        if self._reader_binary_path():
            last_err = None
            for attempt in range(3):
                try:
                    return self._read_with_binary()
                except Exception as e:
                    last_err = e
                    logger.debug(f"C 读取器尝试 {attempt + 1} 失败: {e}")
                    time.sleep(2.5)  # DHT11 两次读取间隔需 >2s
            logger.warning(f"C 读取器连续失败，回退 Python gpiod: {last_err}")

        if not _GPIOD_AVAILABLE:
            raise RuntimeError("gpiod 库不可用，无法读取 DHT11")

        # 纯 Python gpiod 回退
        last_err = None
        for attempt in range(3):
            try:
                return self._read_once()
            except Exception as e:
                last_err = e
                time.sleep(0.2 + attempt * 0.1)
        raise last_err

    def _read_once(self):
        pin = int(self.gpio_pin)
        chip_name = str(self.gpio_chip)

        # 尝试提升实时优先级，减少读取期间的调度抖动
        old_scheduler = None
        old_priority = None
        try:
            import os
            old_scheduler = os.sched_getscheduler(0)
            old_priority = os.sched_getparam(0)
            # SCHED_FIFO 优先级 50（范围 1~99），失败后静默回退
            os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(50))
        except Exception:
            pass

        chip = gpiod.Chip(chip_name)
        line = None
        try:
            line = chip.get_line(pin)

            # 安全：如果该 line 已被内核/其他驱动占用，跳过避免系统异常
            if hasattr(line, 'is_used') and line.is_used():
                raise RuntimeError(f"GPIO {pin} 已被系统占用，无法使用")

            # 阶段1: 主机起始信号
            # DHT11 上电后需要 >1s 稳定时间，每次读取前保持高电平 2s
            line.request(consumer='dht11', type=gpiod.LINE_REQ_DIR_OUT, default_vals=[1])
            line.set_value(1)
            time.sleep(2.0)
            line.set_value(0)
            time.sleep(0.030)      # 拉低 30ms，兼容更多模块
            line.set_value(1)
            time.sleep(0.00005)    # 释放 50us
            line.release()

            # 阶段2: 切换为输入等待传感器响应
            # gpiod 1.4.x 不一定支持 BIAS_PULL_UP，不支持则回退
            try:
                line.request(
                    consumer='dht11',
                    type=gpiod.LINE_REQ_DIR_IN,
                    flags=gpiod.LINE_REQ_FLAG_BIAS_PULL_UP
                )
            except Exception:
                line.request(consumer='dht11', type=gpiod.LINE_REQ_DIR_IN)

            def wait_level(expected, timeout_us):
                """忙等待电平变化，返回是否成功。"""
                deadline = time.perf_counter() + timeout_us / 1_000_000.0
                while line.get_value() != expected:
                    if time.perf_counter() > deadline:
                        return False
                return True

            # 等待 DHT 拉低（响应开始），超时放宽到 1ms
            if not wait_level(0, 1000):
                raise TimeoutError("等待 DHT 响应低电平超时")
            # 等待 DHT 拉高（响应结束）
            if not wait_level(1, 1000):
                raise TimeoutError("等待 DHT 响应高电平超时")
            # 等待 DHT 再次拉低（数据开始）
            if not wait_level(0, 1000):
                raise TimeoutError("等待 DHT 数据开始超时")

            # 阶段3: 读取 40bit 数据
            bits = []
            for _ in range(40):
                # 等待低电平结束（bit 开始）
                if not wait_level(1, 100):
                    raise TimeoutError("等待 bit 高电平超时")

                start = time.perf_counter()
                if not wait_level(0, 100):
                    raise TimeoutError("等待 bit 低电平超时")
                duration = time.perf_counter() - start

                # 高电平 > 40us 视为 1，否则为 0
                bits.append(1 if duration > 0.00004 else 0)

            line.release()
        finally:
            try:
                chip.close()
            except Exception:
                pass
            # 恢复原始调度策略
            try:
                if old_scheduler is not None and old_priority is not None:
                    os.sched_setscheduler(0, old_scheduler, old_priority)
            except Exception:
                pass

        # 阶段4: 解析 40bit 数据
        data_bits = ''.join(str(b) for b in bits)
        data_bytes = [int(data_bits[i:i + 8], 2) for i in range(0, 40, 8)]
        humidity_int, humidity_dec, temp_int, temp_dec, checksum = data_bytes

        calc_sum = (humidity_int + humidity_dec + temp_int + temp_dec) & 0xFF
        if calc_sum != checksum:
            raise ValueError(f"DHT 校验和错误: 计算={calc_sum}, 收到={checksum}")

        humidity = humidity_int + humidity_dec / 10.0
        temperature = temp_int + temp_dec / 10.0
        if temp_int & 0x80:
            temperature = -(temperature & 0x7F)

        return round(temperature, 1), round(humidity, 1)


def create_sensor_from_config(on_data=None):
    """
    根据环境变量 / 默认值创建 DHT11Sensor 实例。

    配置项（环境变量）：
        DHT11_ENABLED: 是否启用，默认 'true'。
        DHT11_GPIO_PIN: GPIO line 编号，默认 None（未配置时只能使用模拟模式）。
        DHT11_GPIO_CHIP: GPIO chip 名称，默认 'gpiochip0'。
        DHT11_POLL_INTERVAL: 读取间隔（秒），默认 5。
        DHT11_SIMULATE: 是否模拟，默认 'false'。
    """
    enabled = os.getenv('DHT11_ENABLED', 'true').lower() in ('1', 'true', 'yes')
    if not enabled:
        return None

    pin_env = os.getenv('DHT11_GPIO_PIN')
    pin = int(pin_env) if pin_env and pin_env.strip() else None

    chip = os.getenv('DHT11_GPIO_CHIP', 'gpiochip0')
    interval = float(os.getenv('DHT11_POLL_INTERVAL', '5'))
    simulate = os.getenv('DHT11_SIMULATE', 'false').lower() in ('1', 'true', 'yes')

    return DHT11Sensor(
        gpio_pin=pin,
        gpio_chip=chip,
        poll_interval=interval,
        simulate=simulate,
        on_data=on_data
    )