self._set_off()
except Exception:
pass
- # 1=打印每次 set_led_state 后的最终生效模式（默认关闭，避免刷屏）
- try:
- if "vc_config" in globals() and vc_config:
- self._dbg = 1 if vc_config.get_bool("LED_DEBUG", False) else 0
- else:
- self._dbg = 0
- except Exception:
- self._dbg = 0
- try:
- if "vc_config" in globals() and vc_config:
- self._key_dbg = 1 if vc_config.get_bool("KEY_LED_DEBUG", False) else 0
- else:
- self._key_dbg = 0
- except Exception:
- self._key_dbg = 0
# LED 电平极性：有些板子是低电平点亮（active-low），也可能仅某一路反相
self._inv_r = 0
self._inv_g = 0
self._apply_by_table(msg)
def _apply_by_table(self, _msg=None):
- prev_mode = self._mode
# P1~P3：AI 阶段（播放 > 思考 > 聆听）
try:
if self._st_ai == "playing":
self._set_green_on()
- return self._dbg_print(prev_mode, _msg)
+ return
if self._st_ai == "thinking":
self._set_bg_alt(1000)
- return self._dbg_print(prev_mode, _msg)
+ return
if self._st_ai == "listening":
self._set_blue_blink(1000)
- return self._dbg_print(prev_mode, _msg)
+ return
except Exception:
pass
# 基础态：充电/电量/信号
try:
if self._st_charging == "full":
self._set_green_on()
- return self._dbg_print(prev_mode, _msg)
+ return
if self._st_charging == "charging":
self._set_red_on()
- return self._dbg_print(prev_mode, _msg)
+ return
if self._st_battery in ("empty", "low"):
self._set_red_blink(1000)
- return self._dbg_print(prev_mode, _msg)
+ return
if self._st_signal == "weak":
self._set_bg_alt(2000)
- return self._dbg_print(prev_mode, _msg)
+ return
except Exception:
pass
self._set_off()
- return self._dbg_print(prev_mode, _msg)
-
- def _dbg_print(self, prev_mode, _msg):
- if getattr(self, "_dbg", 0) != 1:
- return
- try:
- # 仅在模式变化或本次携带 ai_state 时打印，减少刷屏
- if prev_mode != self._mode:
- print("[LED] msg={} st(ai={},chg={},bat={},sig={}) mode:{}->{}".format(
- _msg,
- self._st_ai, self._st_charging, self._st_battery, self._st_signal,
- prev_mode, self._mode))
- else:
- try:
- if isinstance(_msg, dict) and ("ai_state" in _msg or "charging" in _msg):
- print("[LED] msg={} st(ai={},chg={},bat={},sig={}) mode={}".format(
- _msg,
- self._st_ai, self._st_charging, self._st_battery, self._st_signal,
- self._mode))
- except Exception:
- pass
- except Exception:
- pass
def get_led_debug_state(self, event=None, msg=None):
# 供按键/状态机打印当前 LED 聚合状态与最终模式
soc = None
csq = None
rssi = None
+ usb_status = None
chg = "none"
bat = "normal"
sig = "normal"
except Exception:
pass
- # 轮询调试打印（每次 10s tick 一次）
+ # 轮询调试打印（每次 10s tick 一次）：硬件采样一行 + LED 聚合状态一行内打完
try:
if self._dbg == 1:
- print("[HW] usb={} vbatt_mv={} soc={} chg={} bat={} csq={} rssi={} sig={} period_ms={}".format(
- usb_status, v, soc, self._chg, self._bat, csq, rssi, self._sig, self._period_ms))
+ led = None
+ try:
+ led = EventMesh.publish("get_led_debug_state")
+ except Exception:
+ pass
+ print("[HW] usb={} vbatt_mv={} soc={} chg={} bat={} csq={} rssi={} sig={} period_ms={} led={}".format(
+ usb_status, v, soc, self._chg, self._bat, csq, rssi, self._sig, self._period_ms, led))
except Exception:
pass
# print("get_audio_state !!!!!{} {} {}\r\n".format(self.play_end,EventMesh.publish("get_audio_state"),self.play_flag))
# ===== 播放结束检测：用 FSM 驱动 play_end 动作链 =====
try:
- if self.VC_mode != 1:
- if getattr(self.fsm, "await_played", False):
- # 不能只用 stream_is_empty 判定“播完”，否则可能在队列仍有待播数据/统计尚未对齐时提前关播，出现“没声/一闪而过”。
- is_pcm_empty = (self.play_pcm == None) or (self.play_pcm != None and self.play_pcm.stream_is_empty() == 1)
- try:
- q_empty = self.AI_audio_queue.empty()
- except Exception:
- q_empty = True
- try:
- in_len = getattr(self, "tts_in_data_len", 0)
- out_len = getattr(self, "tts_out_data_len", 0)
- except Exception:
- in_len, out_len = 0, 0
- drained = (in_len > 0 and out_len >= in_len and q_empty)
- # 8s 超时：兜底认为播放结束（避免播放状态卡死）
- too_long = False
- try:
- t0 = getattr(self.fsm, "last_tts_completed_ms", 0)
- if t0:
- too_long = utime.ticks_diff(utime.ticks_ms(), t0) > 8000
- except Exception:
- pass
- if (is_pcm_empty and drained) or too_long:
- EventMesh.publish("publish_AI_btnInfo", {"func": self.VC_Play_End})
- self._fsm_apply_actions(self.fsm.on_local_play_end())
- self.VC_state = VC_State_None
+ if getattr(self.fsm, "await_played", False):
+ # PTT(VC_mode==1) 也必须走本段：旧逻辑整段包在 VC_mode!=1 内会导致 tts.completed 后仍 await_played，
+ # 无人触发 VC_Play_End，绿灯/playing 卡死至按键。
+ # 不能只用 stream_is_empty：队列仍可能有待播数据时提前关播会“没声”；in/out 字节在分帧下可能未完全对齐，
+ # 在已 tts.completed 且队列已空时，以服务端结束下发 + PCM 空为准。
+ is_pcm_empty = (self.play_pcm == None) or (self.play_pcm != None and self.play_pcm.stream_is_empty() == 1)
+ try:
+ q_empty = self.AI_audio_queue.empty()
+ except Exception:
+ q_empty = True
+ try:
+ in_len = getattr(self, "tts_in_data_len", 0)
+ out_len = getattr(self, "tts_out_data_len", 0)
+ except Exception:
+ in_len, out_len = 0, 0
+ try:
+ tts_done_ms = getattr(self.fsm, "last_tts_completed_ms", 0)
+ except Exception:
+ tts_done_ms = 0
+ drained_strict = (in_len > 0 and out_len >= in_len and q_empty)
+ drained_after_server = (bool(tts_done_ms) and q_empty)
+ drained = drained_strict or drained_after_server
+ # 8s 超时：兜底认为播放结束（避免播放状态卡死）
+ too_long = False
+ try:
+ t0 = getattr(self.fsm, "last_tts_completed_ms", 0)
+ if t0:
+ too_long = utime.ticks_diff(utime.ticks_ms(), t0) > 8000
+ except Exception:
+ pass
+ if (is_pcm_empty and drained) or too_long:
+ EventMesh.publish("publish_AI_btnInfo", {"func": self.VC_Play_End})
+ self._fsm_apply_actions(self.fsm.on_local_play_end())
+ self.VC_state = VC_State_None
except Exception:
pass
if self.online_kws == 1:
projects / mirrors / ASR1603_G5.git / commitdiff