233 lines
7.8 KiB
Python
233 lines
7.8 KiB
Python
# 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')}")
|