212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
硬件管理器模块
|
|
提供硬件对象的统一管理和访问
|
|
"""
|
|
from maix import time
|
|
import _thread
|
|
import config
|
|
from at_client import ATClient
|
|
|
|
|
|
class HardwareManager:
|
|
"""硬件管理器（单例）"""
|
|
_instance = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(HardwareManager, cls).__new__(cls)
|
|
cls._instance._initialized = False
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if self._initialized:
|
|
return
|
|
|
|
# 私有硬件对象
|
|
self._uart4g = None # 4G模块UART
|
|
self._bus = None # I2C总线
|
|
self._adc_obj = None # ADC对象
|
|
self._at_client = None # AT客户端
|
|
self._status_led_monitor_started = False
|
|
|
|
self._last_active_time = 0 # 用于记录用户的最后一次活跃的时间
|
|
self._stop_timer = False # 用于停止定时器的标志
|
|
|
|
self._initialized = True
|
|
|
|
|
|
|
|
# ==================== 硬件访问（只读属性）====================
|
|
|
|
@property
|
|
def uart4g(self):
|
|
"""4G模块UART（只读）"""
|
|
return self._uart4g
|
|
|
|
@property
|
|
def bus(self):
|
|
"""I2C总线（只读）"""
|
|
return self._bus
|
|
|
|
@property
|
|
def adc_obj(self):
|
|
"""ADC对象（只读）"""
|
|
return self._adc_obj
|
|
|
|
@property
|
|
def at_client(self):
|
|
"""AT客户端（只读）"""
|
|
return self._at_client
|
|
|
|
# ==================== 初始化方法 ====================
|
|
|
|
def init_uart4g(self, device=None, baudrate=None):
|
|
"""初始化4G模块UART"""
|
|
from maix import uart
|
|
if device is None:
|
|
device = config.UART4G_DEVICE
|
|
if baudrate is None:
|
|
baudrate = config.UART4G_BAUDRATE
|
|
self._uart4g = uart.UART(device, baudrate)
|
|
return self._uart4g
|
|
|
|
def init_bus(self, bus_num=None):
|
|
"""初始化I2C总线"""
|
|
from maix import i2c
|
|
if bus_num is None:
|
|
bus_num = config.I2C_BUS_NUM
|
|
self._bus = i2c.I2C(bus_num, i2c.Mode.MASTER)
|
|
return self._bus
|
|
|
|
def init_adc(self, channel=None, res_bit=None):
|
|
"""初始化ADC"""
|
|
from maix.peripheral import adc
|
|
if channel is None:
|
|
channel = config.ADC_CHANNEL
|
|
if res_bit is None:
|
|
res_bit = adc.RES_BIT_12
|
|
self._adc_obj = adc.ADC(channel, res_bit)
|
|
return self._adc_obj
|
|
|
|
def init_at_client(self, uart_obj=None):
|
|
"""初始化AT客户端"""
|
|
if uart_obj is None:
|
|
if self._uart4g is None:
|
|
raise ValueError("uart4g must be initialized before at_client")
|
|
uart_obj = self._uart4g
|
|
self._at_client = ATClient(uart_obj)
|
|
self._at_client.start()
|
|
return self._at_client
|
|
|
|
def power_off(self):
|
|
"""关闭电源板"""
|
|
try:
|
|
# 物理引脚是 A24，对应 GPIO 功能是 GPIOA24
|
|
# 注意：这里需要先在 config.PIN_MAPPINGS 中配置好 "A24": "GPIOA24"
|
|
from maix import gpio
|
|
# 一代电源板关机信号为高电平
|
|
gpio.GPIO("GPIOA24", gpio.Mode.OUT).value(1)
|
|
except Exception as e:
|
|
print(f"关机失败: {e}")
|
|
|
|
def start_status_led_monitor(self):
|
|
"""后台更新状态灯：正常/充满绿常亮、充电绿闪烁、低电量红常亮。"""
|
|
if self._status_led_monitor_started:
|
|
return
|
|
self._status_led_monitor_started = True
|
|
_thread.start_new_thread(self._status_led_loop, ())
|
|
|
|
def _status_led_loop(self):
|
|
from maix import gpio
|
|
from power import get_bus_voltage, is_charging, voltage_to_percent
|
|
|
|
try:
|
|
green = None
|
|
if getattr(config, "STATUS_LED_GREEN_ENABLED", True):
|
|
green = gpio.GPIO(config.STATUS_LED_GREEN_GPIO, gpio.Mode.OUT)
|
|
red = None
|
|
if getattr(config, "STATUS_LED_RED_ENABLED", True):
|
|
red = gpio.GPIO(config.STATUS_LED_RED_GPIO, gpio.Mode.OUT)
|
|
active = int(config.STATUS_LED_ACTIVE_LEVEL)
|
|
inactive = 0 if active else 1
|
|
if green is not None:
|
|
green.value(inactive)
|
|
if red is not None:
|
|
red.value(inactive)
|
|
last_state = None
|
|
blink_on = False
|
|
blink_period = max(100, int(config.STATUS_LED_CHARGING_BLINK_MS))
|
|
poll_ms = max(100, int(config.STATUS_LED_POLL_MS))
|
|
tick_ms = min(blink_period, poll_ms)
|
|
sensor_elapsed = poll_ms
|
|
blink_elapsed = blink_period
|
|
state = "normal"
|
|
percent = None
|
|
charging = False
|
|
|
|
while self._status_led_monitor_started:
|
|
if sensor_elapsed >= poll_ms:
|
|
voltage = get_bus_voltage()
|
|
percent = voltage_to_percent(voltage) if voltage > 0 else None
|
|
charging = is_charging()
|
|
low = percent is not None and percent <= int(config.STATUS_LED_LOW_BATTERY_PERCENT)
|
|
full = percent is not None and percent >= int(config.STATUS_LED_FULL_BATTERY_PERCENT)
|
|
if charging:
|
|
state = "full" if full else "charging"
|
|
else:
|
|
state = "low" if low else "normal"
|
|
sensor_elapsed = 0
|
|
|
|
if state == "low":
|
|
if green is not None:
|
|
green.value(inactive)
|
|
if red is not None:
|
|
red.value(active)
|
|
elif state == "charging":
|
|
if blink_elapsed >= blink_period:
|
|
blink_on = not blink_on
|
|
blink_elapsed = 0
|
|
if green is not None:
|
|
green.value(active if blink_on else inactive)
|
|
if red is not None:
|
|
red.value(inactive)
|
|
else: # normal or full
|
|
if green is not None:
|
|
green.value(active)
|
|
if red is not None:
|
|
red.value(inactive)
|
|
if state != last_state:
|
|
print(f"[STATUS_LED] state={state} percent={percent} charging={charging}")
|
|
last_state = state
|
|
time.sleep_ms(tick_ms)
|
|
sensor_elapsed += tick_ms
|
|
blink_elapsed += tick_ms
|
|
except Exception as e:
|
|
self._status_led_monitor_started = False
|
|
print(f"[STATUS_LED] monitor failed: {e}")
|
|
|
|
def start_idle_timer(self):
|
|
self._stop_timer = False
|
|
self._last_active_time = time.time()
|
|
|
|
def stop_idle_timer(self):
|
|
self._stop_timer = True
|
|
|
|
def get_idle_time_in_sec(self):
|
|
if self._stop_timer:
|
|
return 0
|
|
diff = time.time() - self._last_active_time
|
|
if diff < 0:
|
|
# 时间可能被重置了，重新计时
|
|
self._last_active_time = time.time()
|
|
return 0
|
|
return diff
|
|
|
|
|
|
# 创建全局单例实例
|
|
hardware_manager = HardwareManager()
|
|
|
|
|