1c8c6ad7e8
Engines now read user.name from the config object at transcription time instead of caching it at init, so name changes take effect immediately. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1059 lines
43 KiB
Python
1059 lines
43 KiB
Python
"""PySide6 main application window for the local transcription app."""
|
|
|
|
from PySide6.QtWidgets import (
|
|
QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
|
|
QPushButton, QLabel, QFileDialog, QMessageBox,
|
|
QDialog, QTextEdit, QCheckBox
|
|
)
|
|
from PySide6.QtCore import Qt, QThread, Signal, QTimer
|
|
from PySide6.QtGui import QFont
|
|
import webbrowser
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
# Add parent directory to path for imports (resolve symlinks)
|
|
sys.path.append(str(Path(__file__).resolve().parent.parent))
|
|
|
|
from client.config import Config
|
|
from client.device_utils import DeviceManager
|
|
from client.transcription_engine_realtime import RealtimeTranscriptionEngine, TranscriptionResult
|
|
from client.deepgram_transcription import DeepgramTranscriptionEngine
|
|
from client.server_sync import ServerSyncClient
|
|
from gui.settings_dialog_qt import SettingsDialog
|
|
from server.web_display import TranscriptionWebServer
|
|
from version import __version__
|
|
import asyncio
|
|
from threading import Thread
|
|
|
|
|
|
class WebServerThread(Thread):
|
|
"""Thread for running the web server."""
|
|
|
|
def __init__(self, web_server):
|
|
super().__init__(daemon=True)
|
|
self.web_server = web_server
|
|
self.loop = None
|
|
self.error = None
|
|
|
|
def run(self):
|
|
"""Run the web server in async event loop."""
|
|
try:
|
|
self.loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(self.loop)
|
|
self.loop.run_until_complete(self.web_server.start())
|
|
except Exception as e:
|
|
self.error = e
|
|
print(f"ERROR: Web server failed to start: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
class EngineStartThread(QThread):
|
|
"""Thread for starting the RealtimeSTT engine without blocking the GUI."""
|
|
|
|
finished = Signal(bool, str) # success, message
|
|
|
|
def __init__(self, transcription_engine):
|
|
super().__init__()
|
|
self.transcription_engine = transcription_engine
|
|
|
|
def run(self):
|
|
"""Initialize the engine in background thread (does NOT start recording)."""
|
|
try:
|
|
success = self.transcription_engine.initialize()
|
|
if success:
|
|
self.finished.emit(True, "Engine initialized successfully")
|
|
else:
|
|
self.finished.emit(False, "Failed to initialize engine")
|
|
except Exception as e:
|
|
self.finished.emit(False, f"Error initializing engine: {e}")
|
|
|
|
|
|
class UpdateDialog(QDialog):
|
|
"""Dialog showing available update information."""
|
|
|
|
def __init__(self, parent, current_version: str, release_info):
|
|
"""
|
|
Initialize the update dialog.
|
|
|
|
Args:
|
|
parent: Parent window
|
|
current_version: Current application version
|
|
release_info: ReleaseInfo object with update details
|
|
"""
|
|
super().__init__(parent)
|
|
self.release_info = release_info
|
|
self.skip_version = False
|
|
|
|
self.setWindowTitle("Update Available")
|
|
self.setModal(True)
|
|
self.setMinimumSize(500, 400)
|
|
self.resize(550, 450)
|
|
|
|
layout = QVBoxLayout()
|
|
layout.setSpacing(15)
|
|
layout.setContentsMargins(20, 20, 20, 20)
|
|
self.setLayout(layout)
|
|
|
|
# Title
|
|
title_label = QLabel("Update Available!")
|
|
title_font = QFont()
|
|
title_font.setPointSize(16)
|
|
title_font.setBold(True)
|
|
title_label.setFont(title_font)
|
|
layout.addWidget(title_label)
|
|
|
|
# Version info
|
|
version_label = QLabel(f"Current: {current_version} \u2192 New: {release_info.version}")
|
|
version_font = QFont()
|
|
version_font.setPointSize(12)
|
|
version_label.setFont(version_font)
|
|
layout.addWidget(version_label)
|
|
|
|
# Release notes
|
|
notes_label = QLabel("Release Notes:")
|
|
notes_label.setStyleSheet("font-weight: bold; margin-top: 10px;")
|
|
layout.addWidget(notes_label)
|
|
|
|
self.notes_text = QTextEdit()
|
|
self.notes_text.setReadOnly(True)
|
|
self.notes_text.setPlainText(release_info.release_notes or "No release notes available.")
|
|
layout.addWidget(self.notes_text)
|
|
|
|
# Skip version checkbox
|
|
self.skip_checkbox = QCheckBox(f"Skip version {release_info.version}")
|
|
self.skip_checkbox.setToolTip("Don't show this update again")
|
|
layout.addWidget(self.skip_checkbox)
|
|
|
|
# Buttons
|
|
button_layout = QHBoxLayout()
|
|
button_layout.addStretch()
|
|
|
|
self.later_button = QPushButton("Remind Me Later")
|
|
self.later_button.clicked.connect(self._on_later)
|
|
button_layout.addWidget(self.later_button)
|
|
|
|
self.download_button = QPushButton("Download Update")
|
|
self.download_button.setStyleSheet("background-color: #2ecc71; color: white; font-weight: bold;")
|
|
self.download_button.clicked.connect(self._on_download)
|
|
button_layout.addWidget(self.download_button)
|
|
|
|
layout.addLayout(button_layout)
|
|
|
|
def _on_later(self):
|
|
"""Handle 'Remind Me Later' button click."""
|
|
self.skip_version = self.skip_checkbox.isChecked()
|
|
self.reject()
|
|
|
|
def _on_download(self):
|
|
"""Handle 'Download Update' button click."""
|
|
self.skip_version = self.skip_checkbox.isChecked()
|
|
if self.release_info.download_url:
|
|
webbrowser.open(self.release_info.download_url)
|
|
self.accept()
|
|
|
|
|
|
class MainWindow(QMainWindow):
|
|
"""Main application window using PySide6."""
|
|
|
|
def __init__(self, splash_screen=None):
|
|
"""Initialize the main window."""
|
|
super().__init__()
|
|
|
|
# Store splash screen reference
|
|
self.splash_screen = splash_screen
|
|
|
|
# Application state
|
|
self.is_transcribing = False
|
|
self.config = Config()
|
|
self.device_manager = DeviceManager()
|
|
|
|
# Components (initialized later)
|
|
self.transcription_engine: RealtimeTranscriptionEngine = None
|
|
self.engine_start_thread: EngineStartThread = None
|
|
|
|
# Track current model settings
|
|
self.current_model_size: str = None
|
|
self.current_device_config: str = None
|
|
|
|
# Web server components
|
|
self.web_server: TranscriptionWebServer = None
|
|
self.web_server_thread: WebServerThread = None
|
|
|
|
# Server sync components
|
|
self.server_sync_client: ServerSyncClient = None
|
|
|
|
# Store all transcriptions for saving (separate from display)
|
|
self.transcriptions: list = []
|
|
|
|
# Configure window
|
|
self.setWindowTitle("Local Transcription")
|
|
self.resize(700, 300)
|
|
self.setMinimumSize(600, 280)
|
|
|
|
# Set application icon
|
|
# In PyInstaller frozen executables, use _MEIPASS for bundled files
|
|
import sys
|
|
if getattr(sys, 'frozen', False):
|
|
# Running in PyInstaller bundle
|
|
icon_path = Path(sys._MEIPASS) / "LocalTranscription.png"
|
|
else:
|
|
# Running in normal Python
|
|
icon_path = Path(__file__).resolve().parent.parent / "LocalTranscription.png"
|
|
|
|
if icon_path.exists():
|
|
from PySide6.QtGui import QIcon
|
|
self.setWindowIcon(QIcon(str(icon_path)))
|
|
|
|
# Update splash
|
|
self._update_splash("Creating user interface...")
|
|
|
|
# Create UI
|
|
self._create_widgets()
|
|
|
|
# Update splash
|
|
self._update_splash("Starting web server...")
|
|
|
|
# Start web server if enabled
|
|
self._start_web_server_if_enabled()
|
|
|
|
# Update splash
|
|
self._update_splash("Loading Whisper model...")
|
|
|
|
# Initialize components (in background)
|
|
self._initialize_components()
|
|
|
|
# Schedule update check 3 seconds after startup (non-blocking)
|
|
QTimer.singleShot(3000, self._startup_update_check)
|
|
|
|
def _update_splash(self, message: str):
|
|
"""Update splash screen message if it exists."""
|
|
if self.splash_screen:
|
|
from PySide6.QtCore import Qt
|
|
from PySide6.QtGui import QColor
|
|
from PySide6.QtWidgets import QApplication
|
|
self.splash_screen.showMessage(message, Qt.AlignBottom | Qt.AlignCenter, QColor("#888888"))
|
|
QApplication.processEvents()
|
|
|
|
def _create_widgets(self):
|
|
"""Create all UI widgets."""
|
|
# Central widget
|
|
central_widget = QWidget()
|
|
self.setCentralWidget(central_widget)
|
|
|
|
main_layout = QVBoxLayout()
|
|
central_widget.setLayout(main_layout)
|
|
|
|
# Header
|
|
header_widget = QWidget()
|
|
header_widget.setFixedHeight(80)
|
|
header_layout = QHBoxLayout()
|
|
header_widget.setLayout(header_layout)
|
|
|
|
title_label = QLabel("Local Transcription")
|
|
title_font = QFont()
|
|
title_font.setPointSize(24)
|
|
title_font.setBold(True)
|
|
title_label.setFont(title_font)
|
|
header_layout.addWidget(title_label)
|
|
|
|
header_layout.addStretch()
|
|
|
|
self.settings_button = QPushButton("⚙ Settings")
|
|
self.settings_button.setFixedSize(120, 40)
|
|
self.settings_button.clicked.connect(self._open_settings)
|
|
header_layout.addWidget(self.settings_button)
|
|
|
|
main_layout.addWidget(header_widget)
|
|
|
|
# Status bar
|
|
status_widget = QWidget()
|
|
status_widget.setFixedHeight(40)
|
|
status_layout = QHBoxLayout()
|
|
status_layout.setContentsMargins(0, 0, 0, 0)
|
|
status_widget.setLayout(status_layout)
|
|
|
|
self.status_label = QLabel("⚫ Initializing...")
|
|
status_font = QFont()
|
|
status_font.setPointSize(12)
|
|
self.status_label.setFont(status_font)
|
|
status_layout.addWidget(self.status_label)
|
|
|
|
device_info = self.device_manager.get_device_info()
|
|
device_text = device_info[0][1] if device_info else "No device"
|
|
self.device_label = QLabel(f"Device: {device_text}")
|
|
status_layout.addWidget(self.device_label)
|
|
|
|
user_name = self.config.get('user.name', 'User')
|
|
self.user_label = QLabel(f"User: {user_name}")
|
|
status_layout.addWidget(self.user_label)
|
|
|
|
status_layout.addStretch()
|
|
|
|
main_layout.addWidget(status_widget)
|
|
|
|
# Web display links section
|
|
links_widget = QWidget()
|
|
links_layout = QVBoxLayout()
|
|
links_layout.setContentsMargins(0, 5, 0, 5)
|
|
links_layout.setSpacing(5)
|
|
links_widget.setLayout(links_layout)
|
|
|
|
# Local web display link
|
|
web_host = self.config.get('web_server.host', '127.0.0.1')
|
|
web_port = self.config.get('web_server.port', 8080)
|
|
web_url = f"http://{web_host}:{web_port}"
|
|
self.web_link = QLabel(f'🌐 Local Web Display: <a href="{web_url}">{web_url}</a>')
|
|
self.web_link.setOpenExternalLinks(True)
|
|
self.web_link.setToolTip("Click to open in browser (for OBS)")
|
|
self.web_link.setStyleSheet("QLabel a { color: #4CAF50; }")
|
|
links_layout.addWidget(self.web_link)
|
|
|
|
# Multi-user sync display link (shown when server sync is enabled)
|
|
self.sync_link = QLabel("")
|
|
self.sync_link.setOpenExternalLinks(True)
|
|
self.sync_link.setStyleSheet("QLabel a { color: #2196F3; }")
|
|
self.sync_link.setVisible(False)
|
|
links_layout.addWidget(self.sync_link)
|
|
self._update_sync_link()
|
|
|
|
main_layout.addWidget(links_widget)
|
|
|
|
# Control buttons
|
|
control_widget = QWidget()
|
|
control_widget.setFixedHeight(80)
|
|
control_layout = QHBoxLayout()
|
|
control_widget.setLayout(control_layout)
|
|
|
|
self.start_button = QPushButton("▶ Start Transcription")
|
|
self.start_button.setFixedSize(240, 50)
|
|
button_font = QFont()
|
|
button_font.setPointSize(14)
|
|
button_font.setBold(True)
|
|
self.start_button.setFont(button_font)
|
|
self.start_button.clicked.connect(self._toggle_transcription)
|
|
self.start_button.setStyleSheet("background-color: #2ecc71; color: white;")
|
|
control_layout.addWidget(self.start_button)
|
|
|
|
self.clear_button = QPushButton("🗑 Clear")
|
|
self.clear_button.setFixedSize(120, 50)
|
|
self.clear_button.clicked.connect(self._clear_transcriptions)
|
|
control_layout.addWidget(self.clear_button)
|
|
|
|
self.save_button = QPushButton("💾 Save")
|
|
self.save_button.setFixedSize(120, 50)
|
|
self.save_button.clicked.connect(self._save_transcriptions)
|
|
control_layout.addWidget(self.save_button)
|
|
|
|
control_layout.addStretch()
|
|
|
|
main_layout.addWidget(control_widget)
|
|
|
|
# Version label (bottom right)
|
|
version_label = QLabel(f"v{__version__}")
|
|
version_label.setStyleSheet("QLabel { color: #666; font-size: 10px; }")
|
|
version_label.setAlignment(Qt.AlignRight)
|
|
main_layout.addWidget(version_label)
|
|
|
|
def _initialize_components(self):
|
|
"""Initialize RealtimeSTT transcription engine."""
|
|
# Update status
|
|
self.status_label.setText("⚙ Initializing...")
|
|
|
|
# Set device based on config
|
|
device_config = self.config.get('transcription.device', 'auto')
|
|
self.device_manager.set_device(device_config)
|
|
|
|
# Get audio device
|
|
audio_device_str = self.config.get('audio.input_device', 'default')
|
|
audio_device = None if audio_device_str == 'default' else int(audio_device_str)
|
|
|
|
# Initialize transcription engine with RealtimeSTT
|
|
model = self.config.get('transcription.model', 'base.en')
|
|
language = self.config.get('transcription.language', 'en')
|
|
device = self.device_manager.get_device_for_whisper()
|
|
compute_type = self.config.get('transcription.compute_type', 'default')
|
|
|
|
# Track current settings
|
|
self.current_model_size = model
|
|
self.current_device_config = device_config
|
|
|
|
user_name = self.config.get('user.name', 'User')
|
|
|
|
# Check for continuous/fast speaker mode
|
|
continuous_mode = self.config.get('transcription.continuous_mode', False)
|
|
|
|
# Get timing settings - use faster values if continuous mode is enabled
|
|
if continuous_mode:
|
|
# Faster settings for speakers who talk without pauses
|
|
post_speech_silence = 0.15 # Reduced from default 0.3
|
|
min_gap = 0.0 # No gap between recordings
|
|
min_recording = 0.3 # Shorter minimum recording
|
|
else:
|
|
post_speech_silence = self.config.get('transcription.post_speech_silence_duration', 0.3)
|
|
min_gap = self.config.get('transcription.min_gap_between_recordings', 0.0)
|
|
min_recording = self.config.get('transcription.min_length_of_recording', 0.5)
|
|
|
|
remote_mode = self.config.get('remote.mode', 'local')
|
|
|
|
if remote_mode in ('managed', 'byok'):
|
|
# Use Deepgram-based remote transcription
|
|
self.transcription_engine = DeepgramTranscriptionEngine(
|
|
config=self.config,
|
|
input_device_index=audio_device
|
|
)
|
|
self.transcription_engine.set_callbacks(
|
|
realtime_callback=self._on_realtime_transcription,
|
|
final_callback=self._on_final_transcription
|
|
)
|
|
self.transcription_engine.set_error_callback(self._on_remote_error)
|
|
self.transcription_engine.set_credits_low_callback(self._on_credits_low)
|
|
else:
|
|
# Use local Whisper transcription
|
|
self.transcription_engine = RealtimeTranscriptionEngine(
|
|
model=model,
|
|
device=device,
|
|
language=language,
|
|
compute_type=compute_type,
|
|
enable_realtime_transcription=self.config.get('transcription.enable_realtime_transcription', False),
|
|
realtime_model=self.config.get('transcription.realtime_model', 'tiny.en'),
|
|
realtime_processing_pause=self.config.get('transcription.realtime_processing_pause', 0.1),
|
|
silero_sensitivity=self.config.get('transcription.silero_sensitivity', 0.4),
|
|
silero_use_onnx=self.config.get('transcription.silero_use_onnx', True),
|
|
webrtc_sensitivity=self.config.get('transcription.webrtc_sensitivity', 3),
|
|
post_speech_silence_duration=post_speech_silence,
|
|
min_length_of_recording=min_recording,
|
|
min_gap_between_recordings=min_gap,
|
|
pre_recording_buffer_duration=self.config.get('transcription.pre_recording_buffer_duration', 0.2),
|
|
beam_size=self.config.get('transcription.beam_size', 5),
|
|
initial_prompt=self.config.get('transcription.initial_prompt', ''),
|
|
no_log_file=self.config.get('transcription.no_log_file', True),
|
|
input_device_index=audio_device,
|
|
app_config=self.config
|
|
)
|
|
|
|
# Set up callbacks for transcription results
|
|
self.transcription_engine.set_callbacks(
|
|
realtime_callback=self._on_realtime_transcription,
|
|
final_callback=self._on_final_transcription
|
|
)
|
|
|
|
# Start engine in background thread (downloads models, initializes VAD, etc.)
|
|
self.engine_start_thread = EngineStartThread(self.transcription_engine)
|
|
self.engine_start_thread.finished.connect(self._on_engine_ready)
|
|
self.engine_start_thread.start()
|
|
|
|
def _on_engine_ready(self, success: bool, message: str):
|
|
"""Handle engine initialization completion."""
|
|
if success:
|
|
remote_mode = self.config.get('remote.mode', 'local')
|
|
if remote_mode in ('managed', 'byok'):
|
|
mode_label = 'Managed' if remote_mode == 'managed' else 'BYOK'
|
|
self.device_label.setText(f"Device: Deepgram ({mode_label})")
|
|
elif self.transcription_engine:
|
|
actual_device = self.transcription_engine.device
|
|
compute_type = self.transcription_engine.compute_type
|
|
device_display = f"{actual_device.upper()} ({compute_type})"
|
|
self.device_label.setText(f"Device: {device_display}")
|
|
|
|
host = self.config.get('web_server.host', '127.0.0.1')
|
|
port = self.config.get('web_server.port', 8080)
|
|
self.status_label.setText(f"✓ Ready | Web: http://{host}:{port}")
|
|
self.start_button.setEnabled(True)
|
|
else:
|
|
self.status_label.setText("❌ Engine initialization failed")
|
|
QMessageBox.critical(self, "Error", message)
|
|
self.start_button.setEnabled(False)
|
|
|
|
def _start_web_server_if_enabled(self):
|
|
"""Start web server."""
|
|
try:
|
|
host = self.config.get('web_server.host', '127.0.0.1')
|
|
port = self.config.get('web_server.port', 8080)
|
|
show_timestamps = self.config.get('display.show_timestamps', True)
|
|
fade_after_seconds = self.config.get('display.fade_after_seconds', 10)
|
|
max_lines = self.config.get('display.max_lines', 50)
|
|
font_family = self.config.get('display.font_family', 'Arial')
|
|
font_size = self.config.get('display.font_size', 16)
|
|
fonts_dir = self.config.fonts_dir # Custom fonts directory
|
|
|
|
# Font source settings
|
|
font_source = self.config.get('display.font_source', 'System Font')
|
|
websafe_font = self.config.get('display.websafe_font', 'Arial')
|
|
google_font = self.config.get('display.google_font', 'Roboto')
|
|
|
|
# Color settings
|
|
user_color = self.config.get('display.user_color', '#4CAF50')
|
|
text_color = self.config.get('display.text_color', '#FFFFFF')
|
|
background_color = self.config.get('display.background_color', '#000000B3')
|
|
|
|
# Try up to 5 ports if the default is in use
|
|
ports_to_try = [port] + [port + i for i in range(1, 5)]
|
|
server_started = False
|
|
|
|
for try_port in ports_to_try:
|
|
print(f"Attempting to start web server at http://{host}:{try_port}")
|
|
self.web_server = TranscriptionWebServer(
|
|
host=host,
|
|
port=try_port,
|
|
show_timestamps=show_timestamps,
|
|
fade_after_seconds=fade_after_seconds,
|
|
max_lines=max_lines,
|
|
font_family=font_family,
|
|
font_size=font_size,
|
|
fonts_dir=fonts_dir,
|
|
font_source=font_source,
|
|
websafe_font=websafe_font,
|
|
google_font=google_font,
|
|
user_color=user_color,
|
|
text_color=text_color,
|
|
background_color=background_color
|
|
)
|
|
self.web_server_thread = WebServerThread(self.web_server)
|
|
self.web_server_thread.start()
|
|
|
|
# Give it a moment to start and check for errors
|
|
import time
|
|
time.sleep(0.5)
|
|
|
|
if self.web_server_thread.error:
|
|
error_str = str(self.web_server_thread.error)
|
|
# Check if it's a port-in-use error
|
|
if "address already in use" in error_str.lower() or "errno 98" in error_str.lower():
|
|
print(f"Port {try_port} is in use, trying next port...")
|
|
self.web_server = None
|
|
self.web_server_thread = None
|
|
continue
|
|
else:
|
|
# Different error, don't retry
|
|
print(f"Web server failed to start: {self.web_server_thread.error}")
|
|
self.web_server = None
|
|
self.web_server_thread = None
|
|
break
|
|
else:
|
|
# Success!
|
|
print(f"✓ Web server started successfully at http://{host}:{try_port}")
|
|
if try_port != port:
|
|
print(f" Note: Using port {try_port} instead of configured port {port}")
|
|
server_started = True
|
|
break
|
|
|
|
if not server_started:
|
|
print(f"WARNING: Could not start web server on any port from {ports_to_try[0]} to {ports_to_try[-1]}")
|
|
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to initialize web server: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
self.web_server = None
|
|
self.web_server_thread = None
|
|
|
|
def _toggle_transcription(self):
|
|
"""Start or stop transcription."""
|
|
if not self.is_transcribing:
|
|
self._start_transcription()
|
|
else:
|
|
self._stop_transcription()
|
|
|
|
def _start_transcription(self):
|
|
"""Start transcription."""
|
|
try:
|
|
# Check if engine is ready
|
|
if not self.transcription_engine or not self.transcription_engine.is_ready():
|
|
QMessageBox.critical(self, "Error", "Transcription engine not ready")
|
|
return
|
|
|
|
# Start recording
|
|
success = self.transcription_engine.start_recording()
|
|
if not success:
|
|
QMessageBox.critical(self, "Error", "Failed to start recording")
|
|
return
|
|
|
|
# Initialize server sync if enabled
|
|
if self.config.get('server_sync.enabled', False):
|
|
self._start_server_sync()
|
|
|
|
# Update UI
|
|
self.is_transcribing = True
|
|
self.start_button.setText("⏸ Stop Transcription")
|
|
self.start_button.setStyleSheet("background-color: #e74c3c; color: white;")
|
|
self.status_label.setText("🔴 Transcribing...")
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", f"Failed to start transcription:\n{e}")
|
|
print(f"Error starting transcription: {e}")
|
|
|
|
def _stop_transcription(self):
|
|
"""Stop transcription."""
|
|
try:
|
|
# Stop recording
|
|
if self.transcription_engine:
|
|
self.transcription_engine.stop_recording()
|
|
|
|
# Stop server sync if running
|
|
if self.server_sync_client:
|
|
self.server_sync_client.stop()
|
|
self.server_sync_client = None
|
|
|
|
# Update UI
|
|
self.is_transcribing = False
|
|
self.start_button.setText("▶ Start Transcription")
|
|
self.start_button.setStyleSheet("background-color: #2ecc71; color: white;")
|
|
self.status_label.setText("✓ Ready")
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", f"Failed to stop transcription:\n{e}")
|
|
print(f"Error stopping transcription: {e}")
|
|
|
|
def _on_realtime_transcription(self, result: TranscriptionResult):
|
|
"""Handle realtime (preview) transcription from RealtimeSTT."""
|
|
if not self.is_transcribing:
|
|
return
|
|
|
|
try:
|
|
# Broadcast preview to local web server
|
|
if self.web_server and self.web_server_thread and self.web_server_thread.loop:
|
|
asyncio.run_coroutine_threadsafe(
|
|
self.web_server.broadcast_preview(
|
|
result.text,
|
|
result.user_name,
|
|
result.timestamp
|
|
),
|
|
self.web_server_thread.loop
|
|
)
|
|
|
|
# Send preview to server sync if enabled
|
|
if self.server_sync_client:
|
|
self.server_sync_client.send_preview(result.text, result.timestamp)
|
|
|
|
except Exception as e:
|
|
print(f"Error handling realtime transcription: {e}")
|
|
|
|
def _on_final_transcription(self, result: TranscriptionResult):
|
|
"""Handle final transcription from RealtimeSTT."""
|
|
if not self.is_transcribing:
|
|
return
|
|
|
|
try:
|
|
# Store transcription for saving
|
|
self.transcriptions.append(result)
|
|
|
|
# Broadcast to web server if enabled
|
|
if self.web_server and self.web_server_thread:
|
|
asyncio.run_coroutine_threadsafe(
|
|
self.web_server.broadcast_transcription(
|
|
result.text,
|
|
result.user_name,
|
|
result.timestamp
|
|
),
|
|
self.web_server_thread.loop
|
|
)
|
|
|
|
# Send to server sync if enabled
|
|
if self.server_sync_client:
|
|
import time
|
|
sync_start = time.time()
|
|
print(f"[GUI] Sending to server sync: '{result.text[:50]}...'")
|
|
self.server_sync_client.send_transcription(
|
|
result.text,
|
|
result.timestamp
|
|
)
|
|
sync_queue_time = (time.time() - sync_start) * 1000
|
|
print(f"[GUI] Queued for sync in: {sync_queue_time:.1f}ms")
|
|
|
|
except Exception as e:
|
|
print(f"Error handling final transcription: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
def _on_remote_error(self, error_msg: str):
|
|
"""Handle error from remote transcription service."""
|
|
print(f"Remote transcription error: {error_msg}")
|
|
self.status_label.setText(f"⚠ Remote error: {error_msg}")
|
|
|
|
# Fallback to local if enabled
|
|
if self.config.get('remote.fallback_to_local', True) and self.is_transcribing:
|
|
print("Falling back to local transcription...")
|
|
self.status_label.setText("⚠ Remote failed — falling back to local")
|
|
|
|
def _on_credits_low(self, seconds_remaining: int):
|
|
"""Handle low credits warning from proxy."""
|
|
minutes = seconds_remaining // 60
|
|
self.status_label.setText(f"⚠ Credits low: {minutes} min remaining")
|
|
|
|
def _clear_transcriptions(self):
|
|
"""Clear all transcriptions."""
|
|
if not self.transcriptions:
|
|
QMessageBox.information(self, "No Transcriptions", "There are no transcriptions to clear.")
|
|
return
|
|
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"Clear Transcriptions",
|
|
f"Are you sure you want to clear {len(self.transcriptions)} transcription(s)?",
|
|
QMessageBox.Yes | QMessageBox.No
|
|
)
|
|
|
|
if reply == QMessageBox.Yes:
|
|
self.transcriptions.clear()
|
|
QMessageBox.information(self, "Cleared", "All transcriptions have been cleared.")
|
|
|
|
def _save_transcriptions(self):
|
|
"""Save transcriptions to file."""
|
|
if not self.transcriptions:
|
|
QMessageBox.warning(self, "No Transcriptions", "There are no transcriptions to save.")
|
|
return
|
|
|
|
filepath, _ = QFileDialog.getSaveFileName(
|
|
self,
|
|
"Save Transcriptions",
|
|
"",
|
|
"Text files (*.txt);;All files (*.*)"
|
|
)
|
|
|
|
if filepath:
|
|
try:
|
|
show_timestamps = self.config.get('display.show_timestamps', True)
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
for result in self.transcriptions:
|
|
line_parts = []
|
|
if show_timestamps:
|
|
time_str = result.timestamp.strftime("%H:%M:%S")
|
|
line_parts.append(f"[{time_str}]")
|
|
if result.user_name and result.user_name.strip():
|
|
line_parts.append(f"{result.user_name}:")
|
|
line_parts.append(result.text)
|
|
f.write(" ".join(line_parts) + "\n")
|
|
QMessageBox.information(self, "Saved", f"Transcriptions saved to:\n{filepath}")
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", f"Failed to save transcriptions:\n{e}")
|
|
|
|
def _open_settings(self):
|
|
"""Open settings dialog."""
|
|
# Get audio devices using sounddevice
|
|
import sounddevice as sd
|
|
audio_devices = []
|
|
try:
|
|
device_list = sd.query_devices()
|
|
for i, device in enumerate(device_list):
|
|
if device['max_input_channels'] > 0:
|
|
audio_devices.append((i, device['name']))
|
|
except:
|
|
pass
|
|
|
|
if not audio_devices:
|
|
audio_devices = [(0, "Default")]
|
|
|
|
# Get compute devices
|
|
compute_devices = self.device_manager.get_device_info()
|
|
compute_devices.insert(0, ("auto", "Auto-detect"))
|
|
|
|
# Open settings dialog
|
|
dialog = SettingsDialog(
|
|
self,
|
|
self.config,
|
|
audio_devices,
|
|
compute_devices,
|
|
on_save=self._on_settings_saved
|
|
)
|
|
dialog.exec()
|
|
|
|
def _on_settings_saved(self):
|
|
"""Handle settings being saved."""
|
|
# Update user label
|
|
user_name = self.config.get('user.name', 'User')
|
|
self.user_label.setText(f"User: {user_name}")
|
|
|
|
# Update web server settings
|
|
if self.web_server:
|
|
self.web_server.show_timestamps = self.config.get('display.show_timestamps', True)
|
|
self.web_server.fade_after_seconds = self.config.get('display.fade_after_seconds', 10)
|
|
self.web_server.max_lines = self.config.get('display.max_lines', 50)
|
|
self.web_server.font_family = self.config.get('display.font_family', 'Arial')
|
|
self.web_server.font_size = self.config.get('display.font_size', 16)
|
|
# Update font source settings
|
|
self.web_server.font_source = self.config.get('display.font_source', 'System Font')
|
|
self.web_server.websafe_font = self.config.get('display.websafe_font', 'Arial')
|
|
self.web_server.google_font = self.config.get('display.google_font', 'Roboto')
|
|
# Update color settings
|
|
self.web_server.user_color = self.config.get('display.user_color', '#4CAF50')
|
|
self.web_server.text_color = self.config.get('display.text_color', '#FFFFFF')
|
|
self.web_server.background_color = self.config.get('display.background_color', '#000000B3')
|
|
|
|
# Update sync link visibility based on server sync settings
|
|
self._update_sync_link()
|
|
|
|
# Restart server sync if it was running and settings changed
|
|
if self.is_transcribing and self.server_sync_client:
|
|
# Stop old client
|
|
self.server_sync_client.stop()
|
|
self.server_sync_client = None
|
|
# Start new one if enabled
|
|
if self.config.get('server_sync.enabled', False):
|
|
self._start_server_sync()
|
|
|
|
# Check if model/device settings changed - reload engine if needed
|
|
new_model = self.config.get('transcription.model', 'base.en')
|
|
new_device_config = self.config.get('transcription.device', 'auto')
|
|
|
|
# Only reload if model size or device changed
|
|
if self.current_model_size != new_model or self.current_device_config != new_device_config:
|
|
self._reload_engine()
|
|
else:
|
|
QMessageBox.information(self, "Settings Saved", "Settings have been applied successfully!")
|
|
|
|
def _reload_engine(self):
|
|
"""Reload the transcription engine with new settings."""
|
|
try:
|
|
# Stop transcription if running
|
|
was_transcribing = self.is_transcribing
|
|
if was_transcribing:
|
|
self._stop_transcription()
|
|
|
|
# Update status
|
|
self.status_label.setText("⚙ Reloading engine...")
|
|
self.start_button.setEnabled(False)
|
|
|
|
# Wait for any existing engine thread to finish and disconnect
|
|
if self.engine_start_thread and self.engine_start_thread.isRunning():
|
|
print("Waiting for previous engine thread to finish...")
|
|
self.engine_start_thread.wait()
|
|
|
|
# Disconnect any existing signals to prevent duplicate connections
|
|
if self.engine_start_thread:
|
|
try:
|
|
self.engine_start_thread.finished.disconnect()
|
|
except:
|
|
pass # Already disconnected or never connected
|
|
|
|
# Stop current engine
|
|
if self.transcription_engine:
|
|
try:
|
|
self.transcription_engine.stop()
|
|
except Exception as e:
|
|
print(f"Warning: Error stopping engine: {e}")
|
|
|
|
# Re-initialize components with new settings
|
|
self._initialize_components()
|
|
|
|
except Exception as e:
|
|
error_msg = f"Error during engine reload: {e}"
|
|
print(error_msg)
|
|
import traceback
|
|
traceback.print_exc()
|
|
self.status_label.setText("❌ Engine reload failed")
|
|
self.start_button.setEnabled(False)
|
|
QMessageBox.critical(self, "Error", error_msg)
|
|
|
|
|
|
def _start_server_sync(self):
|
|
"""Start server sync client."""
|
|
try:
|
|
url = self.config.get('server_sync.url', '')
|
|
room = self.config.get('server_sync.room', 'default')
|
|
passphrase = self.config.get('server_sync.passphrase', '')
|
|
user_name = self.config.get('user.name', 'User')
|
|
fonts_dir = self.config.fonts_dir # Custom fonts directory
|
|
|
|
# Font settings (shared with display settings)
|
|
# Note: "System Font" only works locally, so we treat it as "None" for server sync
|
|
font_source = self.config.get('display.font_source', 'System Font')
|
|
if font_source == "System Font":
|
|
font_source = "None" # System fonts don't work on remote displays
|
|
websafe_font = self.config.get('display.websafe_font', '')
|
|
google_font = self.config.get('display.google_font', '')
|
|
custom_font_file = self.config.get('display.custom_font_file', '')
|
|
|
|
# Color settings
|
|
user_color = self.config.get('display.user_color', '#4CAF50')
|
|
text_color = self.config.get('display.text_color', '#FFFFFF')
|
|
background_color = self.config.get('display.background_color', '#000000B3')
|
|
|
|
if not url:
|
|
print("Server sync enabled but no URL configured")
|
|
return
|
|
|
|
print(f"Starting server sync: {url}, room: {room}, user: {user_name}, font: {font_source}")
|
|
|
|
self.server_sync_client = ServerSyncClient(
|
|
url=url,
|
|
room=room,
|
|
passphrase=passphrase,
|
|
user_name=user_name,
|
|
fonts_dir=fonts_dir,
|
|
font_source=font_source,
|
|
websafe_font=websafe_font if websafe_font else None,
|
|
google_font=google_font if google_font else None,
|
|
custom_font_file=custom_font_file if custom_font_file else None,
|
|
user_color=user_color,
|
|
text_color=text_color,
|
|
background_color=background_color
|
|
)
|
|
self.server_sync_client.start()
|
|
|
|
except Exception as e:
|
|
print(f"Error starting server sync: {e}")
|
|
QMessageBox.warning(
|
|
self,
|
|
"Server Sync Warning",
|
|
f"Failed to start server sync:\n{e}\n\nTranscription will continue locally."
|
|
)
|
|
|
|
def _update_sync_link(self):
|
|
"""Update the multi-user sync link visibility and URL."""
|
|
server_sync_enabled = self.config.get('server_sync.enabled', False)
|
|
server_url = self.config.get('server_sync.url', '')
|
|
room = self.config.get('server_sync.room', 'default')
|
|
|
|
if server_sync_enabled and server_url:
|
|
# Extract base URL from the API endpoint (e.g., http://server:3000/api/send -> http://server:3000)
|
|
try:
|
|
from urllib.parse import urlparse, urlencode
|
|
parsed = urlparse(server_url)
|
|
base_url = f"{parsed.scheme}://{parsed.netloc}"
|
|
|
|
# Get display settings to pass as URL parameters
|
|
params = {
|
|
'room': room,
|
|
'fontfamily': self.config.get('display.font_family', 'Arial'),
|
|
'fontsize': self.config.get('display.font_size', 16),
|
|
'fade': self.config.get('display.fade_after_seconds', 10),
|
|
'timestamps': 'true' if self.config.get('display.show_timestamps', True) else 'false',
|
|
'maxlines': self.config.get('display.max_lines', 50)
|
|
}
|
|
display_url = f"{base_url}/display?{urlencode(params)}"
|
|
# Show shorter text with just address and room
|
|
display_text = f"{base_url} (room: {room})"
|
|
self.sync_link.setText(f'🔗 Multi-User Display: <a href="{display_url}">{display_text}</a>')
|
|
self.sync_link.setToolTip(f"Click to open: {display_url}")
|
|
self.sync_link.setVisible(True)
|
|
except Exception as e:
|
|
print(f"Error parsing server URL: {e}")
|
|
self.sync_link.setVisible(False)
|
|
else:
|
|
self.sync_link.setVisible(False)
|
|
|
|
def closeEvent(self, event):
|
|
"""Handle window closing."""
|
|
# Stop transcription if running
|
|
if self.is_transcribing:
|
|
self._stop_transcription()
|
|
|
|
# Stop web server
|
|
if self.web_server_thread and self.web_server_thread.is_alive():
|
|
try:
|
|
print("Shutting down web server...")
|
|
if self.web_server_thread.loop:
|
|
self.web_server_thread.loop.call_soon_threadsafe(self.web_server_thread.loop.stop)
|
|
except Exception as e:
|
|
print(f"Warning: Error stopping web server: {e}")
|
|
|
|
# Stop transcription engine
|
|
if self.transcription_engine:
|
|
try:
|
|
self.transcription_engine.stop()
|
|
except Exception as e:
|
|
print(f"Warning: Error stopping engine: {e}")
|
|
|
|
# Wait for engine start thread
|
|
if self.engine_start_thread and self.engine_start_thread.isRunning():
|
|
self.engine_start_thread.wait()
|
|
|
|
event.accept()
|
|
|
|
def _startup_update_check(self):
|
|
"""Check for updates on startup (called via QTimer)."""
|
|
# Only check if auto_check is enabled
|
|
if not self.config.get('updates.auto_check', True):
|
|
return
|
|
|
|
# Check if enough time has passed since last check
|
|
last_check_str = self.config.get('updates.last_check', '')
|
|
check_interval = self.config.get('updates.check_interval_hours', 24)
|
|
|
|
if last_check_str:
|
|
try:
|
|
last_check = datetime.fromisoformat(last_check_str)
|
|
hours_since_check = (datetime.now() - last_check).total_seconds() / 3600
|
|
if hours_since_check < check_interval:
|
|
print(f"Skipping update check - last checked {hours_since_check:.1f} hours ago")
|
|
return
|
|
except (ValueError, TypeError):
|
|
pass # Invalid date format, proceed with check
|
|
|
|
# Perform async update check
|
|
self._check_for_updates(show_no_update_message=False)
|
|
|
|
def _check_for_updates(self, show_no_update_message: bool = True):
|
|
"""
|
|
Check for updates.
|
|
|
|
Args:
|
|
show_no_update_message: Whether to show a message if no update is available
|
|
"""
|
|
from client.update_checker import UpdateChecker
|
|
|
|
gitea_url = self.config.get('updates.gitea_url', 'https://repo.anhonesthost.net')
|
|
owner = self.config.get('updates.owner', 'streamer-tools')
|
|
repo = self.config.get('updates.repo', 'local-transcription')
|
|
|
|
if not gitea_url or not owner or not repo:
|
|
if show_no_update_message:
|
|
QMessageBox.warning(self, "Update Check", "Update checking is not configured.")
|
|
return
|
|
|
|
checker = UpdateChecker(
|
|
current_version=__version__,
|
|
gitea_url=gitea_url,
|
|
owner=owner,
|
|
repo=repo
|
|
)
|
|
|
|
def on_update_check_complete(release_info, error):
|
|
# Update last check time
|
|
self.config.set('updates.last_check', datetime.now().isoformat())
|
|
|
|
if error:
|
|
print(f"Update check failed: {error}")
|
|
if show_no_update_message:
|
|
QMessageBox.warning(self, "Update Check Failed", f"Could not check for updates:\n{error}")
|
|
return
|
|
|
|
if release_info:
|
|
# Check if this version is skipped
|
|
skipped_versions = self.config.get('updates.skipped_versions', [])
|
|
if release_info.version in skipped_versions:
|
|
print(f"Skipping update notification for version {release_info.version} (user skipped)")
|
|
return
|
|
|
|
# Show update dialog on main thread
|
|
QTimer.singleShot(0, lambda: self._show_update_dialog(release_info))
|
|
else:
|
|
if show_no_update_message:
|
|
QMessageBox.information(
|
|
self,
|
|
"No Updates",
|
|
f"You are running the latest version ({__version__})."
|
|
)
|
|
|
|
# Run check in background thread
|
|
checker.check_for_update_async(on_update_check_complete)
|
|
|
|
def _show_update_dialog(self, release_info):
|
|
"""
|
|
Show the update dialog.
|
|
|
|
Args:
|
|
release_info: ReleaseInfo object with update details
|
|
"""
|
|
dialog = UpdateDialog(self, __version__, release_info)
|
|
dialog.exec()
|
|
|
|
# If user chose to skip this version, save it
|
|
if dialog.skip_version:
|
|
skipped_versions = self.config.get('updates.skipped_versions', [])
|
|
if release_info.version not in skipped_versions:
|
|
skipped_versions.append(release_info.version)
|
|
self.config.set('updates.skipped_versions', skipped_versions)
|