"""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.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: {web_url}')
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)
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,
user_name=user_name
)
# 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:
# Update device label with actual device used
if 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 _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: {display_text}')
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)