dzxj_dtu/backend/scripts/detect_dht11.py

#!/usr/bin/env python3
"""
DHT11 GPIO 自动探测脚本

在目标板子上运行，自动尝试多个候选 GPIO line，找到能成功读取 DHT11 的 pin。
探测成功后输出 JSON，供部署脚本读取并写入环境变量配置。
"""

import os
import sys
import json
import time

# 把 backend 目录加入路径，以便复用 dht11_sensor 模块
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
BACKEND_DIR = os.path.dirname(SCRIPT_DIR)
sys.path.insert(0, BACKEND_DIR)

from modules.dht11_sensor import DHT11Sensor

# 候选 GPIO line 编号：RK3562 GPIO0_C3_d 对应 gpiochip0 line 19。
# 如需扫描更多，可通过环境变量 DHT11_GPIO_PIN 指定一个或多个（逗号分隔）。
DEFAULT_CANDIDATES = [19]


def detect(gpio_pin=None, gpio_chip='gpiochip0', max_attempts=3, simulate=False):
    """
    探测 DHT11 传感器。

    如果 gpio_pin 给出，只尝试该 pin；否则扫描 DEFAULT_CANDIDATES。
    返回 (pin, temperature, humidity) 或 None。
    simulate=True 用于无硬件环境测试脚本流程。
    """
    if simulate:
        sensor = DHT11Sensor(gpio_pin=gpio_pin or 0, gpio_chip=gpio_chip, simulate=True)
        temp, hum = sensor._read_simulated()
        return {
            'gpio_pin': int(gpio_pin) if gpio_pin else 0,
            'gpio_chip': gpio_chip,
            'temperature': temp,
            'humidity': hum,
            'success': True
        }

    if gpio_pin:
        # 支持逗号分隔的多个候选，例如 "2,64,72"
        candidates = [int(p.strip()) for p in str(gpio_pin).split(',')]
    else:
        candidates = DEFAULT_CANDIDATES

    for pin in candidates:
        sensor = DHT11Sensor(gpio_pin=pin, gpio_chip=gpio_chip, simulate=False)
        for attempt in range(max_attempts):
            try:
                temp, hum = sensor._read_real()
                if temp is not None and hum is not None:
                    return {
                        'gpio_pin': pin,
                        'gpio_chip': gpio_chip,
                        'temperature': temp,
                        'humidity': hum,
                        'success': True
                    }
            except Exception as e:
                print(f"  探测 GPIO {pin} 失败: {e}", file=sys.stderr)
                time.sleep(2.5)  # DHT11 两次读取间隔需 >2s
    return None


def main():
    pin_env = os.getenv('DHT11_GPIO_PIN')
    chip_env = os.getenv('DHT11_GPIO_CHIP', 'gpiochip0')
    simulate = os.getenv('DHT11_SIMULATE', 'false').lower() in ('1', 'true', 'yes')

    print(f"开始探测 DHT11 (chip={chip_env}, simulate={simulate})...", file=sys.stderr)
    if pin_env:
        print(f"用户指定优先探测 GPIO: {pin_env}", file=sys.stderr)

    result = detect(gpio_pin=pin_env, gpio_chip=chip_env, simulate=simulate)

    if result:
        # 同时输出机器可读行和 JSON，方便部署脚本解析
        print(
            f"DHT11_OK pin={result['gpio_pin']} chip={result['gpio_chip']} "
            f"temp={result['temperature']} hum={result['humidity']}"
        )
        print(json.dumps(result, ensure_ascii=False))
        print(
            f"探测成功: GPIO pin={result['gpio_pin']}, "
            f"温度={result['temperature']}°C, 湿度={result['humidity']}%",
            file=sys.stderr
        )
        sys.exit(0)
    else:
        print("DHT11_FAIL")
        error = {
            'success': False,
            'error': '未能在任何候选 GPIO 上读取到 DHT11，请检查接线或手动指定 DHT11_GPIO_PIN',
            'candidates': DEFAULT_CANDIDATES if not pin_env else [int(pin_env)]
        }
        print(json.dumps(error, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()