#!/usr/bin/env python3 """ meetrec_gui.py — Qt (PySide6) GUI for meetrec. Records a meeting from any PipeWire/ALSA input, shows a live rolling transcript while recording, and on Stop runs the final full transcription. Reuses all recording/transcription logic from meetrec.py (the CLI). Setup (Arch): sudo pacman -S python pipewire pipewire-pulse python-pyside6 # or: pip install pyside6 # (plus the usual deps: pip install numpy sounddevice faster-whisper) Run: python meetrec_gui.py """ import queue import sys import threading from types import SimpleNamespace from PySide6.QtCore import Qt, QThread, QTimer, Signal from PySide6.QtWidgets import ( QApplication, QCheckBox, QComboBox, QHBoxLayout, QLabel, QLineEdit, QMessageBox, QProgressBar, QPushButton, QSpinBox, QTextEdit, QVBoxLayout, QWidget, ) import sounddevice as sd import meetrec from meetrec import ( Recorder, final_transcribe, fmt_hms, live_worker, resolve_source, ) # -------------------------------------------------------------------------- # Worker threads # -------------------------------------------------------------------------- class ModelLoader(QThread): """Loads (and on first run downloads) the Whisper model off the GUI thread.""" loaded = Signal(object) failed = Signal(str) def __init__(self, model_name, engine="faster-whisper", device="auto", compute_type="int8", server_url=None, parent=None): super().__init__(parent) self.model_name = model_name self.engine = engine self.device = device self.compute_type = compute_type self.server_url = server_url def run(self): try: self.model = meetrec.load_engine( self.model_name, engine=self.engine, device=self.device, compute_type=self.compute_type, server_url=self.server_url) self.loaded.emit(self.model) except Exception as e: self.failed.emit(str(e)) class FinalWorker(QThread): """Runs the final full-pass transcription without blocking the GUI.""" info = Signal(str) segment = Signal(float, float, str) done = Signal(str) # output stem failed = Signal(str) def __init__(self, model, wav_path, out_stem, language, parent=None): super().__init__(parent) self.model = model self.wav_path = wav_path self.out_stem = out_stem self.language = language def run(self): try: final_transcribe( self.model, self.wav_path, self.out_stem, self.language, on_info=self.info.emit, on_segment=self.segment.emit, on_done=lambda: self.done.emit(self.out_stem), ) except Exception as e: self.failed.emit(str(e)) # -------------------------------------------------------------------------- # Main window # -------------------------------------------------------------------------- class MeetRecWindow(QWidget): MODELS = ["tiny", "base", "small", "medium", "large-v3"] LANGS = ["autodetect", "en", "de", "fr", "es", "it", "nl", "pl", "pt", "ru", "zh", "ja", "ko"] DEVICES = ["auto", "cpu", "cuda"] COMPUTE_TYPES = ["int8", "int8_float16", "float16", "float32"] ENGINES = ["faster-whisper", "whisper-cpp", "whisper-server"] def __init__(self): super().__init__() self.setWindowTitle("MeetRec — record & transcribe meetings") self.setMinimumSize(700, 600) self.model = None self.model_key = None # (engine, model, device, compute) loaded self.rec = None self.state = "idle" # idle | loading | record | finalizing self.decay = 0.0 self.stop_evt = threading.Event() self.line_q: "queue.Queue" = queue.Queue() self._final_header = False self._args = None self._build_ui() self.ui_timer = QTimer(self) self.ui_timer.setInterval(120) self.ui_timer.timeout.connect(self._tick) self.ui_timer.start() # ------------------------------ UI layout ----------------------------- def _build_ui(self): root = QVBoxLayout(self) # row 1: microphone source r1 = QHBoxLayout() r1.addWidget(QLabel("Microphone:")) self.src = QComboBox() self.src.setEditable(True) # free text = name substring, like the CLI self.src.addItem("default") self.src.setMinimumWidth(280) self._fill_sources() refresh = QPushButton("\u21bb") refresh.setFixedWidth(36) refresh.setToolTip("Rescan input devices") refresh.clicked.connect(self._fill_sources) r1.addWidget(self.src, 1) r1.addWidget(refresh) root.addLayout(r1) # row 2: model + language r2 = QHBoxLayout() self.model_lbl = QLabel("Model:") r2.addWidget(self.model_lbl) self.model_cb = QComboBox() self.model_cb.addItems(self.MODELS) self.model_cb.setCurrentText("small") r2.addWidget(self.model_cb) r2.addStretch(1) r2.addWidget(QLabel("Language:")) self.lang = QComboBox() self.lang.setEditable(True) self.lang.addItems(self.LANGS) self.lang.setCurrentText("autodetect") self.lang.setMinimumWidth(110) r2.addWidget(self.lang) root.addLayout(r2) # row 2b: engine + compute device + precision r2b = QHBoxLayout() r2b.addWidget(QLabel("Engine:")) self.engine_cb = QComboBox() self.engine_cb.addItems(self.ENGINES) self.engine_cb.setCurrentText("faster-whisper") self.engine_cb.activated.connect(self._engine_changed) self.engine_cb.setToolTip("faster-whisper: NVIDIA CUDA or CPU.\n" "whisper-cpp: GPU via Vulkan (AMD, Intel, " "NVIDIA) or CPU — preferred on AMD.") r2b.addWidget(self.engine_cb) self.device_lbl = QLabel("Device:") r2b.addWidget(self.device_lbl) self.device_cb = QComboBox() self.device_cb.addItems(self.DEVICES) self.device_cb.setCurrentText("auto") r2b.addWidget(self.device_cb) self.compute_lbl = QLabel("Compute:") r2b.addWidget(self.compute_lbl) self.compute_cb = QComboBox() self.compute_cb.addItems(self.COMPUTE_TYPES) self.compute_cb.setCurrentText("int8") r2b.addWidget(self.compute_cb) r2b.addStretch(1) root.addLayout(r2b) # row 2c: whisper-server URL r2c = QHBoxLayout() self.server_lbl = QLabel("Server:") r2c.addWidget(self.server_lbl) self.server_url = QLineEdit("http://100.103.83.12:8085") self.server_url.setToolTip( "whisper.cpp server (see server/whisper-server/) — " "the model lives on the server") r2c.addWidget(self.server_url, 1) root.addLayout(r2c) # row 3: live options + output name r3 = QHBoxLayout() r3.addWidget(QLabel("Live:")) self.live_cb = QCheckBox("transcribe while recording every") self.live_cb.setChecked(True) r3.addWidget(self.live_cb) self.live_spin = QSpinBox() self.live_spin.setRange(3, 60) self.live_spin.setValue(8) self.live_spin.setSuffix(" s") r3.addWidget(self.live_spin) r3.addStretch(1) r3.addWidget(QLabel("Output:")) self.out = QLineEdit("meeting") self.out.setPlaceholderText("file stem, e.g. meeting") self.out.setMinimumWidth(160) r3.addWidget(self.out) root.addLayout(r3) # row 4: record / stop r4 = QHBoxLayout() self.record_btn = QPushButton("\u23fa Record") self.record_btn.setMinimumHeight(44) self.record_btn.setStyleSheet( "QPushButton { background: #1a7f37; color: white; font-weight: bold; }" "QPushButton:disabled { background: #888; }") self.record_btn.clicked.connect(self.on_record) self.stop_btn = QPushButton("\u23f9 Stop & transcribe") self.stop_btn.setMinimumHeight(44) self.stop_btn.setEnabled(False) self.stop_btn.setStyleSheet( "QPushButton { background: #c0392b; color: white; font-weight: bold; }" "QPushButton:disabled { background: #888; }") self.stop_btn.clicked.connect(self.on_stop) r4.addWidget(self.record_btn, 1) r4.addWidget(self.stop_btn, 1) root.addLayout(r4) # row 5: level meter + elapsed r5 = QHBoxLayout() self.level = QProgressBar() self.level.setRange(0, 100) self.level.setTextVisible(False) self.level.setFixedHeight(10) self.level.setStyleSheet( "QProgressBar::chunk { background: #2ea043; }") self.elapsed = QLabel("00:00:00") self.elapsed.setAlignment(Qt.AlignRight | Qt.AlignVCenter) self.elapsed.setMinimumWidth(80) self.elapsed.setStyleSheet("font-size: 15px; font-weight: bold;") r5.addWidget(self.level, 1) r5.addWidget(self.elapsed) root.addLayout(r5) # transcript self.transcript = QTextEdit() self.transcript.setReadOnly(True) self.transcript.setPlaceholderText( "Live transcript appears here while recording.\n" "After stopping, the final (higher-quality) transcript is shown " "and saved as .txt / .srt / .json next to the .wav.") root.addWidget(self.transcript, 1) # status bar bar = QHBoxLayout() self.status = QLabel("") bar.addWidget(self.status, 1) clear = QPushButton("Clear transcript") clear.clicked.connect(lambda: self.transcript.clear()) bar.addWidget(clear) root.addLayout(bar) # ------------------------------ helpers ------------------------------- def _fill_sources(self): keep = self.src.currentText() self.src.blockSignals(True) self.src.clear() self.src.addItem("default") seen = set() try: for d in sd.query_devices(): if d["max_input_channels"] > 0 and d["name"] not in seen: seen.add(d["name"]) self.src.addItem(d["name"]) except Exception: pass self.src.blockSignals(False) idx = self.src.findText(keep) if idx >= 0: self.src.setCurrentIndex(idx) else: self.src.setCurrentText(keep) def _engine_changed(self, _idx): # What applies per engine: # model: faster-whisper + whisper-cpp (server owns the # model otherwise) # device/compute: faster-whisper only (CTranslate2) # server url: whisper-server only engine = self.engine_cb.currentText() has_model = engine != "whisper-server" has_dc = engine == "faster-whisper" self.model_lbl.setText("Model:" if has_model else "Model (n/a):") self.device_lbl.setText("Device:" if has_dc else "Device (n/a):") self.compute_lbl.setText("Compute:" if has_dc else "Compute (n/a):") self.model_cb.setEnabled(has_model) self.device_cb.setEnabled(has_dc) self.compute_cb.setEnabled(has_dc) self.server_url.setEnabled(engine == "whisper-server") def _set_inputs_enabled(self, enabled: bool): for w in (self.src, self.model_cb, self.engine_cb, self.device_cb, self.compute_cb, self.server_url, self.lang, self.live_cb, self.live_spin, self.out): w.setEnabled(enabled) if enabled: self._engine_changed(-1) # whisper-cpp: keep device/compute off def _to_idle(self): self.state = "idle" self._set_inputs_enabled(True) self.record_btn.setEnabled(True) self.stop_btn.setEnabled(False) def _append(self, text: str): self.transcript.moveCursor(QTextEdit.End) self.transcript.insertPlainText(text) sb = self.transcript.verticalScrollBar() sb.setValue(sb.maximum()) def _push_line(self, a: float, text: str): """Thread-safe: called from the live worker thread.""" self.line_q.put((a, text)) # ------------------------------ state machine -------------------------- def on_record(self): if self.state != "idle": return lang = self.lang.currentText().strip() live = self.live_spin.value() if self.live_cb.isChecked() else 0 self._args = SimpleNamespace( source=self.src.currentText().strip() or "default", model=self.model_cb.currentText(), output=self.out.text().strip() or "meeting", live=live, language=None if lang in ("", "autodetect") else lang, ) self._set_inputs_enabled(False) self.transcript.clear() self._final_header = False self.decay = 0.0 self.level.setValue(0) self.elapsed.setText("00:00:00") key = (self.engine_cb.currentText(), self._args.model, self.device_cb.currentText(), self.compute_cb.currentText(), self.server_url.text().strip()) if self.model is None or self.model_key != key: self.state = "loading" if key[0] == "whisper-server": self.status.setText(f"Using whisper server {key[4]} ...") else: self.status.setText( f"Loading Whisper model {self._args.model!r} via {key[0]} " "(first run downloads it, this can take a minute)...") self._pending_key = key self.loader = ModelLoader( self._args.model, engine=key[0], device=self.device_cb.currentText(), compute_type=self.compute_cb.currentText(), server_url=key[4] or None, parent=self) self.loader.loaded.connect(self._model_ready) self.loader.failed.connect(self._model_failed) self.loader.start() else: self._start_recording() def _model_ready(self, model): self.model = model self.model_key = self._pending_key self._start_recording() def _model_failed(self, err): self._to_idle() self.status.setText("Failed to load model") QMessageBox.critical(self, "MeetRec", f"Could not load Whisper model:\n\n{err}") def _start_recording(self): try: device = resolve_source(self._args.source) except SystemExit: self._to_idle() QMessageBox.critical( self, "MeetRec", f"No input device matching {self._args.source!r}.\n" "Pick one from the dropdown or fix the name.") return try: self.rec = Recorder(device, self._args.output + ".wav") self.rec.start() except Exception as e: self._to_idle() QMessageBox.critical(self, "MeetRec", f"Could not open the microphone:\n\n{e}") return self.stop_evt.clear() self.state = "record" self.record_btn.setEnabled(False) self.stop_btn.setEnabled(True) try: idx = device if device is not None else sd.default.device[0] name = sd.query_devices(idx)["name"] except Exception: name = self._args.source self.status.setText( f"Recording from: {name} @ {self.rec.rate} Hz → " f"{self._args.output}.wav") if self._args.live: self.live_thread = threading.Thread( target=live_worker, args=(self.model, self.rec, self._args, self._args.output + ".live.srt", self.stop_evt, self._push_line), daemon=True) self.live_thread.start() def on_stop(self): if self.state != "record": return self.stop_evt.set() self.status.setText("Stopped — running final transcription...") if self.rec: self.rec.stop() # flushes the WAV writer (a few ms) self.state = "finalizing" self.stop_btn.setEnabled(False) self.worker = FinalWorker( self.model, self._args.output + ".wav", self._args.output, self._args.language, parent=self) self.worker.info.connect(lambda m: self.status.setText(m)) self.worker.segment.connect(self._final_segment) self.worker.done.connect(self._final_done) self.worker.failed.connect(self._final_failed) self.worker.start() def _final_segment(self, a: float, b: float, text: str): if not self._final_header: self._append("\n— final transcript —\n") self._final_header = True self._append(f"[{fmt_hms(a)}] {text}\n") def _final_done(self, stem: str): self._to_idle() self.status.setText( f"Done. Saved: {stem}.wav {stem}.txt {stem}.srt {stem}.json") def _final_failed(self, err): self._to_idle() self.status.setText("Final transcription failed") QMessageBox.critical(self, "MeetRec", f"Final transcription failed:\n\n{err}\n\n" "The .wav recording is still saved.") # ------------------------------ UI tick -------------------------------- def _tick(self): # drain live lines produced by the worker thread try: while True: a, text = self.line_q.get_nowait() self._append(f"[{fmt_hms(a)}] {text}\n") except queue.Empty: pass if self.state == "record" and self.rec is not None: self.decay = max(self.rec.peak, self.decay * 0.70) self.level.setValue(int(min(1.0, self.decay) * 100)) self.elapsed.setText(fmt_hms(self.rec.duration())) # ------------------------------ close ---------------------------------- def closeEvent(self, ev): if self.state in ("loading", "record", "finalizing"): ret = QMessageBox.question( self, "MeetRec", "Stop and exit now?\n\n" "The .wav file is kept, but the final transcription " "will be aborted (if it hasn't finished).", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if ret != QMessageBox.Yes: ev.ignore() return self.stop_evt.set() if self.rec: self.rec.stop() if self.state == "finalizing" and self.worker: self.worker.wait(5000) elif self.state == "loading" and self.loader: self.loader.wait(5000) ev.accept() def main(): app = QApplication(sys.argv) app.setApplicationName("MeetRec") win = MeetRecWindow() win.show() sys.exit(app.exec()) if __name__ == "__main__": main()