self._set_rgb(1, 0, 0)
self._timer.start(self._alert_period_ms, 0, self._timer_cb)
+ def _set_red_flash_limited(self, flashes=3, period_ms=500):
+ """REQ-BAT-03：红快闪若干次后熄灭（约 2 次/秒）。"""
+ try:
+ flashes = int(flashes or 3)
+ except Exception:
+ flashes = 3
+ if flashes < 1:
+ flashes = 1
+ try:
+ period_ms = int(period_ms or 500)
+ except Exception:
+ period_ms = 500
+ if period_ms < 100:
+ period_ms = 100
+ self._mode = "red_flash_n"
+ self._red_flash_steps_left = flashes * 2
+ self._alert_period_ms = period_ms
+ self._blink_on = 1
+ self._timer_stop()
+ self._set_rgb(1, 0, 0)
+ self._timer.start(self._alert_period_ms, 0, self._timer_cb)
+
def _timer_cb(self, *args, **kwargs):
EventMesh.publish("publish_AI_btnInfo", {"func": self._timer_cb_})
self._blink_on = 0 if self._blink_on else 1
self._set_rgb(1 if self._blink_on else 0, 0, 0)
return
+ if self._mode == "red_flash_n":
+ self._red_flash_steps_left -= 1
+ if self._red_flash_steps_left <= 0:
+ self._set_off()
+ return
+ self._blink_on = 0 if self._blink_on else 1
+ self._set_rgb(1 if self._blink_on else 0, 0, 0)
+ try:
+ self._timer.start(self._alert_period_ms, 0, self._timer_cb)
+ except Exception:
+ pass
+ return
if self._mode == "bg_alt":
if self._bg_phase == 0:
self._set_rgb(0, 1, 0)
if aval in ("rb_flash_2", "rb_flash", "rb_flash2"):
self._set_alert_rb_flash(2, 250)
return
+ if aval in ("red_flash_3", "red_flash3"):
+ try:
+ n = 3
+ ms = 500
+ if "vc_config" in globals() and vc_config:
+ n = int(vc_config.get_int("BATTERY_EMPTY_FLASH_COUNT", 3))
+ ms = int(vc_config.get_int("BATTERY_EMPTY_FLASH_PERIOD_MS", 500))
+ self._set_red_flash_limited(n, ms)
+ except Exception:
+ self._set_red_flash_limited(3, 500)
+ return
if aval in (None, "", "none", "clear"):
self._st_alert = None
if "ai_state" in msg:
self._st_signal = msg.get("signal") or "normal"
except Exception:
pass
- if self._mode == "alert_rb":
+ if self._mode in ("alert_rb", "red_flash_n"):
return
if self._led_state_debug_enabled() and isinstance(msg, dict):
try:
pass
def _apply_by_table(self, _msg=None):
- if self._mode == "alert_rb":
+ if self._mode in ("alert_rb", "red_flash_n"):
return
_prev_mode = self._mode
+ # P0：低电/空电（高于 AI 交互灯；充电中不显示，走充电灯态）
+ try:
+ if self._st_charging not in ("charging", "full"):
+ if self._st_battery == "empty":
+ try:
+ ms = 500
+ if "vc_config" in globals() and vc_config:
+ ms = int(vc_config.get_int("BATTERY_EMPTY_FLASH_PERIOD_MS", 500))
+ self._set_red_blink(ms)
+ except Exception:
+ self._set_red_blink(500)
+ return
+ if self._st_battery == "low":
+ self._set_red_blink(1000)
+ return
+ except Exception:
+ pass
# P1~P3：AI 阶段（播放 > 思考 > 聆听）
try:
if self._st_ai == "playing":
return
except Exception:
pass
- # AI ä¼\9aè¯\9dæ\88\96æµ\81æ\92æ\94¾ä¸ï¼\9aai_state ç\9fæ\9a\82 idle æ\97¶ä»\8dä¸\8d让å\85\85ç\94µçº¢/å\85\85满绿æ\8a¢å\8d ï¼\88ä½\8eç\94µ/å¼±ç½\91ä»\8dèµ°å\90\8eé\9d¢å\88\86æ\94¯）
+ # AI 交äº\92ä¸ï¼\88è\81\86å\90¬/æ\80\9dè\80\83/æ\92æ\94¾/æµ\81æ\92ï¼\89ä¸\8d让å\85\85ç\94µç\81¯æ\8a¢å\8d ï¼\9b交äº\92ç»\93æ\9d\9f idle å\90\8eæ\81¢å¤\8då\85\85ç\94µçº¢/å\85\85满绿ï¼\88REQ å\86²çª\81è§\84é\81¿）
_skip_charge_led = 0
try:
- if EventMesh.publish("get_VC_satrt_flag") == 1:
+ if self._st_ai in ("listening", "thinking", "playing"):
_skip_charge_led = 1
except Exception:
pass
if self._st_charging == "charging":
self._set_red_on()
return
- if self._st_battery in ("empty", "low"):
- self._set_red_blink(1000)
- return
if self._st_signal == "weak":
# 弱网：蓝绿「慢交替」，与 thinking 同节拍定义（约 1 s 换一色）
self._set_bg_alt(1000)
self._bat = None
self._sig = None
self._usb = None
+ self._charge_full_latched = 0
+ self._charge_full_streak = 0
+ self._charge_full_streak_need = 2
self._timer = osTimer()
self._period_ms = 10000
self._weak_csq = 10
def _on_usb_state(self, event=None, msg=None):
try:
state = msg
+ prev = self._usb
self._usb = 1 if state == 1 else 0
+ if self._usb == 1 and prev != 1:
+ self._charge_full_latched = 0
+ self._charge_full_streak = 0
+ if self._usb == 0:
+ self._charge_full_latched = 0
+ self._charge_full_streak = 0
except Exception:
pass
try:
usb_status = 0
except Exception:
usb_status = 0
- # charging
+ # charging：插入后先红灯；勿用瞬时 vbatt≥满电阈值直接判满（充电器接入电压即偏高）
if usb_status == 1:
- if v is not None and v >= int(self._charge_full_mv):
+ if getattr(self, "_charge_full_latched", 0) == 1:
chg = "full"
else:
- chg = "charging"
+ if v is not None and v >= int(self._charge_full_mv):
+ self._charge_full_streak = int(getattr(self, "_charge_full_streak", 0)) + 1
+ need = int(getattr(self, "_charge_full_streak_need", 2) or 2)
+ if self._charge_full_streak >= need:
+ self._charge_full_latched = 1
+ chg = "full"
+ else:
+ chg = "charging"
+ else:
+ self._charge_full_streak = 0
+ chg = "charging"
# battery level for LED table
if usb_status != 1 and v is not None:
soc = self._bcfg.soc_from_vbatt(v)
self.usb = USB()
self.usb.setCallback(self.usb_event_cb)
self.usb_status = self.usb.getStatus()
- self.usb_charge_flag = 0
+ self.usb_charge_flag = 0
+ self._usb_evt_last_ms = 0
+ self._usb_evt_last_state = -1
self.Low_power_warning_timer = 0
self.AI_dev_timer = osTimer()
self.AI_dev_timer.start(10000, 1, self.AI_dev_timer_cb)
print("AI_dev:AI_dev_timer")
self.aud = audio.Audio(0)
self.Low_power_warning_flag = 0
+ self._low_warn_last_ms = 0
+ self._empty_shutdown_done = 0
self.VC_TTS_volume = EventMesh.publish("get_storage_config","VC_TTS_volume")#音量配置
if self.VC_TTS_volume == None:
self.VC_TTS_volume = 100
self.Keycall1_press_count = 0
self.ble_switch_flag_time = osTimer()
self.ble_switch_flag_time.start(20000, 0, self.ble_switch_flag_time_init)
+ # AI_hw_status 先于本模块初始化，开机已插 USB 时需补同步一次
+ try:
+ if int(self.usb_status or 0) == 1:
+ EventMesh.publish("hw_usb_state", 1)
+ EventMesh.publish("publish_AI_btnInfo", {"func": self.usb_event, "data": 1})
+ except Exception:
+ pass
def ble_switch_flag_time_init(self,*args):
EventMesh.publish("publish_AI_btnInfo",{"func":self.ble_switch_flag_time_cb})
def ble_switch_flag_time_cb(self,*args):
#ntptime.settime(8)
print("timer:{}!!!!!!!!\r\n".format(utime.localtime()))
def usb_event_cb(self, state):
+ # USB 插拔抖动过滤（日志中 powerKey 与 usb_evt 0/1 连跳会导致误播 OUT/IN）
+ try:
+ now = utime.ticks_ms()
+ last_ms = getattr(self, "_usb_evt_last_ms", 0)
+ last_st = getattr(self, "_usb_evt_last_state", -1)
+ if last_ms and utime.ticks_diff(now, last_ms) < 800 and last_st != state:
+ return
+ self._usb_evt_last_ms = now
+ self._usb_evt_last_state = state
+ except Exception:
+ pass
# 转发给精简硬件监控（AI_hw_status），避免 USB callback 被覆盖导致“后面失效”
try:
EventMesh.publish("hw_usb_state", state)
pass
EventMesh.publish("audio_montage_play",["R_CHARGE_IN.mp3"])
self.Low_power_warning_flag = 0
+ try:
+ self._empty_shutdown_done = 0
+ except Exception:
+ pass
pass
elif status == 0:#拨出usbs
self.usb_status = 0
def AI_dev_timer_cb(self, *args, **kwargs):
#print("AI_dev_timer_cb")
EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_dev_timer_event})
+ def _battery_empty_shutdown_seq(self):
+ """REQ-BAT-03：红快闪 3 次 → 灭灯 → 关机提示音 → 关机（btn 线程执行）。"""
+ try:
+ EventMesh.publish("battery_priority_mute_vc")
+ except Exception:
+ pass
+ try:
+ n, ms = 3, 500
+ if "vc_config" in globals() and vc_config:
+ n = int(vc_config.get_int("BATTERY_EMPTY_FLASH_COUNT", 3))
+ ms = int(vc_config.get_int("BATTERY_EMPTY_FLASH_PERIOD_MS", 500))
+ EventMesh.publish("set_led_state", {"alert": "red_flash_3"})
+ utime.sleep_ms(n * ms * 2 + 200)
+ EventMesh.publish("set_led_state", {"ai_state": "idle", "battery": "normal", "charging": "none"})
+ except Exception:
+ pass
+ try:
+ EventMesh.publish("audio_montage_play", ["didianguanji.mp3"])
+ utime.sleep_ms(4500)
+ self.aud.stopPlayStream()
+ self.aud.stopAll()
+ except Exception:
+ pass
+ try:
+ Power.powerDown()
+ except Exception:
+ pass
+
def AI_dev_timer_event(self):
#EventMesh.publish("audio_montage_play",["liuliangtishi.mp3"])
EventMesh.publish("send_power_init_data")
EventMesh.publish("audio_montage_play",["R_CHARGE_FULL.mp3"])
#self.aud.play(1,0,'U:/static/R_CHARGE_FULL.mp3')#充电已满
self.usb_charge_flag = 1
- if self.usb_status == 0 and Vbatt < _bcfg.low_mv and VC_satrt_flag == 0 and self.Low_power_warning_flag == 0 and voice_record_satrt_flag == 0:
- self.Low_power_warning_flag = 1
- try:
- if not ("vc_config" in globals() and vc_config and vc_config.get_bool("USE_MIN_HW_STATUS", True)):
- EventMesh.publish("set_led_state", {"battery": "low"})
- except Exception:
- pass
- #self.aud.play(1,0,'Low_power_warning.mp3')
- # self.Low_power_warning_timer+=1
- # if self.Low_power_warning_timer >= 2:
- # self.Low_power_warning_timer = 0
- EventMesh.publish("audio_montage_play",["Low_power_warning.mp3"])
- if self.usb_status == 0 and Vbatt >= _bcfg.low_mv and VC_satrt_flag == 0:
- try:
- if not ("vc_config" in globals() and vc_config and vc_config.get_bool("USE_MIN_HW_STATUS", True)):
- EventMesh.publish("set_led_state", {"charging": "none", "battery": "normal"})
- except Exception:
- pass
-
- if self.usb_status == 0 and Vbatt < _bcfg.empty_mv:
- EventMesh.publish("audio_montage_play",["didianguanji.mp3"])
- utime.sleep_ms(4500)
- self.aud.stopPlayStream()
- self.aud.stopAll()
- Power.powerDown()
+ try:
+ low_soc = int(_bcfg.low_soc)
+ empty_soc = int(_bcfg.empty_soc)
+ except Exception:
+ low_soc, empty_soc = 30, 10
+ warn_interval_ms = 300000
+ try:
+ if "vc_config" in globals() and vc_config:
+ warn_interval_ms = int(vc_config.get_int("BATTERY_LOW_WARN_INTERVAL_MS", 300000))
+ except Exception:
+ pass
+ soc = None
+ try:
+ soc = self.get_battery()
+ except Exception:
+ soc = None
+ if self.usb_status == 0 and soc is not None:
+ empty_by_soc = soc < empty_soc
+ empty_by_mv = Vbatt is not None and Vbatt < int(_bcfg.empty_mv)
+ if empty_by_soc or empty_by_mv:
+ if getattr(self, "_empty_shutdown_done", 0) == 0:
+ self._empty_shutdown_done = 1
+ try:
+ print("[BAT] empty shutdown soc={} vbatt={}".format(soc, Vbatt))
+ except Exception:
+ pass
+ EventMesh.publish("publish_AI_btnInfo", {"func": self._battery_empty_shutdown_seq})
+ elif soc < low_soc:
+ if voice_record_satrt_flag == 0:
+ try:
+ now = utime.ticks_ms()
+ last = getattr(self, "_low_warn_last_ms", 0)
+ if (not last) or utime.ticks_diff(now, last) >= warn_interval_ms:
+ self._low_warn_last_ms = now
+ try:
+ print("[BAT] low warn soc={} interval_ms={}".format(soc, warn_interval_ms))
+ except Exception:
+ pass
+ EventMesh.publish("audio_montage_play", ["Low_power_warning.mp3"])
+ except Exception:
+ pass
+ else:
+ try:
+ self._low_warn_last_ms = 0
+ self._empty_shutdown_done = 0
+ except Exception:
+ pass
+ if VC_satrt_flag == 0 and Vbatt is not None and Vbatt >= int(_bcfg.low_mv):
+ try:
+ if not ("vc_config" in globals() and vc_config and vc_config.get_bool("USE_MIN_HW_STATUS", True)):
+ EventMesh.publish("set_led_state", {"charging": "none", "battery": "normal"})
+ except Exception:
+ pass
self.check_rest_timer_count += 1
if self.check_rest_timer_count >= 8640:#24小时没按键，则关机
if soc > 100:
return 100
return int(soc)
- def audio_montage_play(self,event,msg):
- self.wakeup_BT()
- # 仅服务端 TTS PCM 在播时跳过；FSM 处于 TtsReceiving 但尚未出声时不挡本地 montage
+ def _audio_is_battery_priority(self, msg):
try:
- if EventMesh.publish("get_vc_tts_playback_active") == 1:
- print("[audio-arb] skip local montage_play during tts_pcm")
- return
+ for i in (msg or []):
+ n = str(i)
+ if ("Low_power" in n) or ("didianguanji" in n):
+ return True
except Exception:
pass
+ return False
+
+ def audio_montage_play(self,event,msg):
+ self.wakeup_BT()
+ bat_pri = self._audio_is_battery_priority(msg)
+ if bat_pri:
+ try:
+ EventMesh.publish("battery_priority_mute_vc")
+ except Exception:
+ pass
+ else:
+ # 仅服务端 TTS PCM 在播时跳过；低电/关机提示优先级更高（见上）
+ try:
+ if EventMesh.publish("get_vc_tts_playback_active") == 1:
+ print("[audio-arb] skip local montage_play during tts_pcm")
+ return
+ except Exception:
+ pass
self.aud.stopPlayStream()
self.aud.stopAll()
str_data='mp3files='
except Exception:
pass
+ def battery_priority_mute_vc(self, event=None, msg=None):
+ """低电/关机提示音：短暂停 VC TTS，避免被流式播放挡住。"""
+ try:
+ self._ptt_hard_mute_local_playback()
+ except Exception:
+ pass
+ try:
+ if getattr(self, "VC_Play_flag", 0) == 1 or getattr(self, "play_pcm", None) is not None:
+ self.VC_Play_End()
+ except Exception:
+ pass
+
def _vc_enqueue_btn(self, func, data=None):
"""统一经 publish_AI_btnInfo 入队，在 btnInfo_queue_thread 串行执行。"""
try:
EventMesh.subscribe("get_VC_satrt_flag", self.get_VC_satrt_flag)
EventMesh.subscribe("get_vc_stream_active", self.get_vc_stream_active)
EventMesh.subscribe("get_vc_tts_playback_active", self.get_vc_tts_playback_active)
+ EventMesh.subscribe("battery_priority_mute_vc", self.battery_priority_mute_vc)
EventMesh.subscribe("VC_error", self.VC_error)
EventMesh.subscribe("audio_test", self.audio_test)
EventMesh.subscribe("audio_test_close", self.audio_test_close)
projects / mirrors / ASR1603_G5.git / commitdiff