# talk.py — Push-to-Talk (ASR/STT) untuk TUI. # # Alur: # * F9 (configurable) → mulai rekam (ffmpeg subprocess, non-blocking). # * F9 lagi / Enter → stop rekam → transkrip (thread) → hasil DISISIPKAN # ke form input pada posisi kursor, TIDAK submit. # * Backspace / Esc → batal: rekaman dibuang / hasil transkrip diabaikan. # # State disimpan di app: # app.talk_recorder — instance Recorder saat merekam, else None # app.talk_transcribing — True saat request ASR berjalan (thread) # app.talk_cancelled — True jika user membatalkan saat transcribing # app.talk_done/_thread/_talk_result — pola selesai-thread (mirip agent_done) import os import threading import config from tools import asr as asr_mod from .agent import log def _shortcut_label(): """Label tombol rekam untuk pesan UI (dari config.asr.record_key).""" n = (config.ASR_RECORD_KEY or "f9").strip().lower() if n.startswith("ctrl-") and len(n) == 6: return f"Ctrl+{n[5].upper()}" return n.upper() # ------------------------------------------------------------------ helpers --- def talking(app): """True jika sedang merekam atau sedang mentranskrip.""" return getattr(app, "talk_recorder", None) is not None or \ getattr(app, "talk_transcribing", False) def insert_text(app, text) -> bool: """Sisipkan text ke input buffer pada posisi kursor (multi-line aman). Tidak mengubah isi lain dan tidak submit.""" if not text: return False buf = app.input_buffer li, col = app.input_line, app.input_col if "\n" not in text: cur = buf[li] buf[li] = cur[:col] + text + cur[col:] app.input_col = col + len(text) return True lines = text.split("\n") cur = buf[li] head, tail = cur[:col], cur[col:] block = [head + lines[0]] + lines[1:-1] + [lines[-1] + tail] buf[li:li + 1] = block app.input_line = li + len(lines) - 1 app.input_col = len(lines[-1]) return True # ------------------------------------------------------------------- actions --- def start_talk(app, stdscr): # Validasi awal: model set 'asr' harus sudah ada sebelum mikrofon dibuka. if not getattr(app, "asr_ready", False): import lib.moop as _moop if _moop.default_chain(None, "asr"): app.asr_ready = True if not getattr(app, "asr_ready", False): log(app, "error", "Push-to-Talk: model set 'asr' belum diset. " "Model > Select Model → Speech Transcription.") return False try: rec = asr_mod.start_recording( record_dir=config.ASR_RECORD_DIR, backend=config.ASR_RECORD_BACKEND, device=config.ASR_DEVICE, fmt=config.ASR_FORMAT, sample_rate=config.ASR_SAMPLE_RATE, channels=config.ASR_CHANNELS, max_seconds=config.ASR_MAX_SECONDS, windows_mic=getattr(config, "ASR_WINDOWS_MIC", None), ) except asr_mod.RecorderError as e: log(app, "error", f"Push-to-Talk: {e}") return False app.talk_recorder = rec app.talk_cancelled = False log(app, "system", f" \u25cf REC — bicara sekarang. {_shortcut_label()}: stop+transkrip, " "Backspace: batal") return True def _transcribe_worker(app, path): res = asr_mod.transcribe_file(path, timeout=config.ASR_TIMEOUT) if not config.ASR_KEEP_RECORDINGS: try: if os.path.isfile(path): os.remove(path) except OSError: pass app._talk_result = res app.talk_transcribing = False app.talk_done.set() def stop_and_transcribe(app, stdscr): """Stop rekaman lalu transkrip di thread. Hasil masuk form (tanpa submit).""" rec = getattr(app, "talk_recorder", None) if rec is None: if getattr(app, "talk_transcribing", False): log(app, "system", " Transkripsi masih berjalan, tunggu sebentar...") return path = rec.path try: path, duration = rec.stop() except asr_mod.RecorderError as e: app.talk_recorder = None log(app, "error", f"Push-to-Talk: {e}") return app.talk_recorder = None app.talk_transcribing = True app.talk_cancelled = False log(app, "system", f" Transcribing... ({duration:.1f}s)") app.talk_done.clear() app._talk_thread = threading.Thread( target=_transcribe_worker, args=(app, path), daemon=True ) app._talk_thread.start() def cancel_talk(app): """Backspace/Esc: batalkan rekaman (buang file) atau transkripsi (abaikan hasil). Kondisi 'pending' sengaja lebar: selama transcribing ATAU hasil sudah menunggu untuk disisipkan (_talk_result / talk_done), pembatalan tetap berlaku — menutup race saat ASR selesai sesaat sebelum Backspace.""" rec = getattr(app, "talk_recorder", None) if rec is not None: rec.cancel() app.talk_recorder = None log(app, "system", " Rekaman dibatalkan.") return pending = (getattr(app, "talk_transcribing", False) or getattr(app, "_talk_result", None) is not None or app.talk_done.is_set()) if pending: app.talk_cancelled = True log(app, "system", " Hasil transkripsi akan diabaikan.") def talk_toggle(app, stdscr): """Handler aksi F9: mulai rekam / stop+transkrip (toggle).""" if getattr(app, "talk_transcribing", False): return if getattr(app, "talk_recorder", None) is not None: stop_and_transcribe(app, stdscr) else: start_talk(app, stdscr) def talk_key(app, stdscr, key) -> bool: """Konsumsi satu key saat mode talk. Returns True bila key dikonsumsi.""" from . import keycodes if key in (keycodes.KEY_ENTER, 10, 13): # Enter: stop + transkrip, hasil HANYA mengisi form (tidak submit). stop_and_transcribe(app, stdscr) return True if key in (keycodes.KEY_BACKSPACE, 127, keycodes.KEY_ESC): cancel_talk(app) return True if key == keycodes.KEY_CTRL_C: # Ctrl+C tetap jadi exit aplikasi (bukan konsumsi mode talk). return False # Semua key lain diabaikan selama mode talk. return True def poll_talk(app, stdscr): """Dipanggil tiap iterasi loop: tangani auto-stop saat ffmpeg mencapai max_seconds (prosesnya exit sendiri).""" rec = getattr(app, "talk_recorder", None) if rec is not None and not rec.alive(): log(app, "system", " Batas waktu rekaman tercapai.") stop_and_transcribe(app, stdscr) def shutdown(app): """Dipanggil saat TUI keluar: pastikan proses ffmpeg rekaman mati & file dibuang.""" rec = getattr(app, "talk_recorder", None) if rec is not None: try: rec.cancel() except Exception: pass app.talk_recorder = None th = getattr(app, "_talk_thread", None) if th is not None: app.talk_cancelled = True try: th.join(timeout=1) except Exception: pass app._talk_thread = None def finish_talk(app, stdscr): """Dipanggil main loop saat app.talk_done ter-set: sisipkan hasil.""" res = getattr(app, "_talk_result", None) app._talk_result = None if getattr(app, "talk_cancelled", False): app.talk_cancelled = False return if not res: return if res.get("ok"): text = res.get("text", "") if not text: log(app, "system", " Tidak ada ucapan terdeteksi.") return if insert_text(app, text): log(app, "system", f" \u266a [{res.get('provider', '?')} / {res.get('model', '?')}] " "teks masuk ke form (Ctrl+Enter untuk kirim)") if app.scroll_follow: app.scroll = 999999 else: log(app, "error", f"ASR: {res.get('error', 'unknown error')}")