from app.staging_gui import microphone_semantic_family as _sf_mod
# Windows mic diagnosis (Task6-9). Safe soft-import; gates deactivate gracefully if missing.
from app.staging_gui import windows_mic_diagnostics as _wmd_mod
mic_start_btn: Any = None
mic_stop_btn: Any = None
mic_lock: threading.Lock = field(default_factory=threading.Lock)
hook callbacks, and intercept existing Pausar/Kill commands for mic integration.
4. Implements mic as default-OFF push-to-talk with hard cap 55s, no background listening,
self._mic_state = DesktopMicState()
self._copy_btn_ref: Any = None
self._mic_device_available_live: bool = False
self._mic_real_audio_captured_live: bool = False
self._mic_last_diagnostic_report: Any = None
self._mic_last_retry_proposals: List[Any] = []
self._mic_last_human_diagnostic_text: str = ""
pausar = getattr(window, "_desktop_toggle_btn", None)
kill = getattr(window, "_global_kill_btn", None)
if getattr(window, "_staging_copy_mic_injected__", False):
setattr(window, "_staging_copy_mic_injected__", True)
setattr(window, "_stg_mic", DesktopMicState())
self._inject_microphone_buttons(window, root)
self._hook_pause_kill_for_mic(window, root)
self._install_mic_default_off(window)
# (T3/T4/T5 gate driver, T9 mic pipeline driver, T10/T11 final regression).
#    is first classified by microphone_semantic_family classifier, storing
send_btn = getattr(window, "_chat_send_button", None)
if send_btn is None:
orig_cmd = send_btn.cget("command")
_alt_cmd = send_btn["command"]
send_btn.configure(command=_pre_router_wrapper)
sys.stdout.write(f"[PRE_ROUTER_DEBUG] send_btn.configure err: {_cfg_err!r}\n")
_bc = callable(send_btn.cget("command")) if send_btn else "no_btn"
f"btn_cmd_callable={_bc} "
btn_found=send_btn is not None,
send_btn = getattr(window, "_chat_send_button", None)
if send_btn is None:
parent = getattr(send_btn, "master", None)
copy_btn = tk.Button(
copy_btn = tk.Button(parent, text="Copiar")
copy_btn.configure(command=_copy_handler)
side = getattr(send_btn, "pack_info", lambda: {}).get().get("side", "left")
copy_btn.pack(side=side, padx=4, pady=2, after=send_btn)
copy_btn.pack(after=send_btn)
copy_btn.pack()
self._copy_btn_ref = copy_btn
btn_ok=copy_btn is not None,
# 3. _inject_microphone_buttons
#    Creates 🎤 Hablar (mic START) and ⏹ Detener (mic STOP) buttons into
#    the button row after any existing copy-btn widget, wires their
#    command callbacks into _mic_start_capture / _mic_stop_capture_and_run_stt,
#    stores widget references into window._stg_mic so later lifecycle
#    methods (_install_mic_default_off, stop-event handlers) can reach them.
def _inject_microphone_buttons(self, window: Any, root: Any) -> None:
mic_state = getattr(window, "_stg_mic", None)
if mic_state is None:
mic_state = DesktopMicState()
setattr(window, "_stg_mic", mic_state)
anchor = getattr(window, "_chat_send_button", None) or self._copy_btn_ref
mic_start_btn = tk.Button(
mic_stop_btn = tk.Button(
mic_start_btn = tk.Button(parent, text="🎤 Hablar")
mic_stop_btn = tk.Button(parent, text="⏹ Detener")
injector._mic_start_capture(window, root)
_dbg_instrument("mic_start_error", error=repr(_se))
injector._mic_stop_capture_and_run_stt(
_dbg_instrument("mic_stop_error", error=repr(_se))
mic_start_btn.configure(command=_start_cmd)
mic_stop_btn.configure(command=_stop_cmd)
mic_start_btn.pack(after=anchor, side="left", padx=4, pady=2)
mic_start_btn.pack(after=anchor)
mic_start_btn.pack()
mic_stop_btn.pack(after=mic_start_btn, side="left", padx=4, pady=2)
mic_stop_btn.pack(after=mic_start_btn)
mic_stop_btn.pack()
mic_state.mic_start_btn = mic_start_btn
mic_state.mic_stop_btn = mic_stop_btn
mic_state.root_ref = root
mic_state.input_entry = getattr(window, "_chat_input_entry", None)
mic_state.input_entry = None
"mic_buttons_injected",
start_ok=mic_start_btn is not None,
stop_ok=mic_stop_btn is not None,
# 4. _hook_pause_kill_for_mic
#    button original commands.  If microphone is currently LISTENING or
def _hook_pause_kill_for_mic(self, window: Any, root: Any) -> None:
pausar = getattr(window, "_desktop_toggle_btn", None)
kill = getattr(window, "_global_kill_btn", None)
mic_state = getattr(window, "_stg_mic", None)
if mic_state is not None:
cur_state = getattr(mic_state, "state", _MIC_OFF)
ev = getattr(mic_state, "stop_event", None)
mic_state.state = _MIC_STOPPING
mic_state.last_stop_reason = "PAUSE_BTN_INTERRUPT"
mic_state = getattr(window, "_stg_mic", None)
if mic_state is not None:
cur_state = getattr(mic_state, "state", _MIC_OFF)
ev = getattr(mic_state, "stop_event", None)
mic_state.state = _MIC_STOPPING
mic_state.last_stop_reason = "KILL_SWITCH_INTERRUPT"
# 5. _install_mic_default_off
#    DEFAULT-on-start mic posture.  Ensures state is OFF, START button is
def _install_mic_default_off(self, window: Any) -> None:
mic_state = getattr(window, "_stg_mic", None)
if mic_state is None:
mic_state.state = _MIC_OFF
ev = getattr(mic_state, "stop_event", None)
start_btn = getattr(mic_state, "mic_start_btn", None)
stop_btn = getattr(mic_state, "mic_stop_btn", None)
if start_btn is not None:
start_btn.configure(state="normal")
start_btn.pack()
if stop_btn is not None:
stop_btn.configure(state="disabled")
stop_btn.pack_forget()
"mic_default_off_installed",
start_btn_ok=start_btn is not None,
stop_btn_ok=stop_btn is not None,
state=str(getattr(mic_state, "state", "?")),
#    Hardens the mic OFF-default posture against any background/open-mic
setattr(window, "_stg_mic_listen_gate_armed", True)
# If user closes root while LISTENING → force mic teardown cleanly
mic_state = getattr(window, "_stg_mic", None)
if mic_state is not None:
ev = getattr(mic_state, "stop_event", None)
mic_state.state = _MIC_OFF
stream = getattr(mic_state, "stream", None)
mic_state.stream = None
timer = getattr(mic_state, "hard_cap_timer", None)
gate_armed=bool(getattr(window, "_stg_mic_listen_gate_armed", False)),
def _mic_start_capture(self, window: Any, root: Any) -> None:
"""Open PyAudio stream, begin appending frames to mic_state.frames list,
que quieres hablar", "no puedo acceder al micrófono"}; has_details ∈
def _write_mic_diagnosis(kind: str) -> None:
"Entiendo que quieres hablar con el micrófono, pero no puedo acceder al micrófono "
"  · Sin dispositivo predeterminado: en Sonido → Grabación marca tu micrófono como "
"  · Permisos Windows: valida el acceso al micrófono en Configuración → Privacidad.\n"
"micrófono aparece sin errores.\n"
"Entiendo que quieres hablar, pero no puedo abrir el canal de audio del micrófono.\n\n"
"  · Revisa configuración: otra aplicación puede tener el micrófono ocupado (zoom, "
"  · Dispositivos encontrados: tu micrófono está desconectado o deshabilitado.\n"
"  · Permisos Windows: vuelve a validar acceso al micrófono desde Configuración.\n"
armed = getattr(window, "_stg_mic_listen_gate_armed", None)
_dbg_instrument("mic_start_blocked_gate", armed=armed)
mic_state = getattr(window, "_stg_mic", None)
if mic_state is None:
lock = getattr(mic_state, "mic_lock", None)
if getattr(mic_state, "state", _MIC_OFF) not in (_MIC_OFF,):
"mic_start_skipped_wrong_state",
state=getattr(mic_state, "state", "?"),
mic_state.state = _MIC_PERMISSION_PENDING
mic_state.last_stop_reason = "PYAUDIO_UNAVAILABLE"
_write_mic_diagnosis("PYAUDIO_UNAVAILABLE")
mic_state.pyaudio_mod = pa
mic_state.state = _MIC_PERMISSION_PENDING
mic_state.last_stop_reason = "PYAUDIO_INIT_FAIL"
_write_mic_diagnosis("PYAUDIO_INIT_FAIL")
mic_state.audio_interface = audio_if
mic_state.frames = frames_list
stop_ev = getattr(mic_state, "stop_event", None)
stop_ev_now = getattr(mic_state, "stop_event", None)
mic_state.state = _MIC_PERMISSION_PENDING
mic_state.last_stop_reason = "OPEN_STREAM_FAIL"
_write_mic_diagnosis("OPEN_STREAM_FAIL")
mic_state.stream = stream
mic_state.state = _MIC_LISTENING
sev = getattr(mic_state, "stop_event", None)
self._mic_stop_capture_and_run_stt(
mic_state.hard_cap_timer = cap_timer
sb = getattr(mic_state, "mic_start_btn", None)
spb = getattr(mic_state, "mic_stop_btn", None)
"mic_capture_started",
self._mic_device_available_live = True
def _mic_stop_capture_and_run_stt(
mic_state = getattr(window, "_stg_mic", None)
if mic_state is None:
lock = getattr(mic_state, "mic_lock", None)
mic_state.last_stop_reason = str(reason or "USER_STOP")
timer = getattr(mic_state, "hard_cap_timer", None)
ev = getattr(mic_state, "stop_event", None)
stream = getattr(mic_state, "stream", None)
mic_state.stream = None
cur_state = getattr(mic_state, "state", _MIC_OFF)
mic_state.state = _MIC_STT_PROCESSING
frames = list(getattr(mic_state, "frames", []) or [])
reason=mic_state.last_stop_reason or reason,
mic_state.state = _MIC_OFF
self._install_mic_default_off(window)
self._mic_real_audio_captured_live = bool(frames)
"mic_capture_stopped",
reason=mic_state.last_stop_reason,
mic_state = getattr(self, "_mic_state", None)
for winst in [getattr(self, "_copy_btn_ref", None)]:
def get_mic_readiness_flags(self) -> Dict[str, Any]:
T9-TR4 4-case truth table: mic_ready_ok = bool(device_avail_live) AND
device_avail = bool(getattr(self, "_mic_device_available_live", False))
captured_live = bool(getattr(self, "_mic_real_audio_captured_live", False))
"mic_device_available_live": device_avail,
"mic_real_audio_captured_live": captured_live,
"mic_ready_ok": bool(device_avail and captured_live),
"last_diagnostic_report": getattr(self, "_mic_last_diagnostic_report", None),
"last_retry_proposals": list(getattr(self, "_mic_last_retry_proposals", []) or []),
getattr(self, "_mic_last_human_diagnostic_text", "") or ""
# no secondary worker, late callback (e.g. mic diagnostic event), retry
#   (b) PART1 AFTER_CHAT_FINAL snapshot send_btn + entry binds
if send_btn is not None and hasattr(send_btn, "cget"):
_p78_turn_ctx["SEND_BUTTON_COMMAND_AFTER_CHAT_FINAL"] = str(send_btn.cget("command") or "")
Historical public symbol imported by run_live_desktop_with_copy_and_mic.py