I'm creating a virtual assistant. 

Here is my current code and other files thus far.

================================================
FILE: main.py
================================================
#!/usr/bin/env python3
# main.py

import sys
import logging
import queue
import threading
from pathlib import Path
import traceback
import signal
import tkinter as tk
from tkinter import messagebox

# Ensure modules/ directory is on Python’s import path
sys.path.insert(0, str(Path(__file__).resolve().parent))

# --- GUI Imports ---
try:
    import customtkinter
except ImportError as e:
    print(f"FATAL ERROR: Failed to import CustomTkinter: {e}", file=sys.stderr)
    print("Please ensure CustomTkinter is installed: uv add customtkinter", file=sys.stderr)
    sys.exit(1)

# --- Core Module Imports ---
try:
    from modules.config import Config, ConfigError
    from modules.system_manager import SystemManager, LoggingSetupError, RequirementError
    from modules.audio_manager import AudioManager, AudioManagerError
    from modules.stt_manager import STTManager, STTManagerError
    from modules.llm_manager import LLMManager, LLMManagerError
    from modules.tts_manager import TTSManager, TTSManagerError
    from modules.context_manager import ContextManager, ContextManagerError
except ImportError as e:
    print(f"FATAL ERROR: Failed to import core modules: {e}", file=sys.stderr)
    print("Ensure you are running from the project root directory and all modules exist.", file=sys.stderr)
    sys.exit(1)

logger = logging.getLogger(__name__)
gui_queue = queue.Queue()


class MiraiGuiApp(customtkinter.CTk):
    APP_NAME = "MiraiAssist"
    APP_VERSION = "0.1.0"

    def __init__(self):
        super().__init__()
        # --- State placeholders ---
        self.config: Config = None
        self.system_manager: SystemManager = None
        self.audio_manager: AudioManager = None
        self.stt_manager: STTManager = None
        self.context_manager: ContextManager = None
        self.llm_manager: LLMManager = None
        self.tts_manager: TTSManager = None

        self.initialization_complete = False
        self.initialization_error = None
        self._ptt_key_pressed = False
        self._is_processing_audio = False
        self._assistant_streaming = False
        self._processing_lock = threading.Lock()

        # --- Build UI ---
        self.title(self.APP_NAME)
        self.protocol("WM_DELETE_WINDOW", self.quit_app)
        self.minsize(700, 550)
        self.geometry("800x600")
        self._create_menu()
        customtkinter.set_appearance_mode("system")
        self._create_widgets()

        # --- Kick off backend init ---
        self.update_status("Initializing backend…")
        threading.Thread(target=self.start_backend_initialization, daemon=True, name="BackendInitThread").start()
        self.poll_queue()

    def _create_menu(self):
        menubar = tk.Menu(self)
        self.configure(menu=menubar)
        file_menu = tk.Menu(menubar, tearoff=0)
        menubar.add_cascade(label="File", menu=file_menu)
        file_menu.add_command(label="Exit", command=self.quit_app)
        options = tk.Menu(menubar, tearoff=0)
        menubar.add_cascade(label="Options", menu=options)
        theme = tk.Menu(options, tearoff=0)
        options.add_cascade(label="Theme", menu=theme)
        theme.add_command(label="Light", command=lambda: self.change_theme("light"))
        theme.add_command(label="Dark",  command=lambda: self.change_theme("dark"))
        theme.add_command(label="System",command=lambda: self.change_theme("system"))
        help_menu = tk.Menu(menubar, tearoff=0)
        menubar.add_cascade(label="Help", menu=help_menu)
        help_menu.add_command(label="About", command=self.show_about_dialog)

    def _create_widgets(self):
        # layout
        self.grid_columnconfigure(0, weight=1)
        self.grid_rowconfigure(0, weight=1)
        self.grid_rowconfigure(1, weight=0)

        main = customtkinter.CTkFrame(self, fg_color="transparent")
        main.grid(row=0, column=0, padx=10, pady=(0,5), sticky="nsew")
        main.grid_columnconfigure(0, weight=1)
        main.grid_rowconfigure(0, weight=3)
        main.grid_rowconfigure(1, weight=1)
        main.grid_rowconfigure(2, weight=0)

        # history
        hf = customtkinter.CTkFrame(main)
        hf.grid(row=0, column=0, sticky="nsew", pady=(0,5))
        hf.grid_columnconfigure(0, weight=1)
        hf.grid_rowconfigure(0, weight=1)
        self.history_textbox = customtkinter.CTkTextbox(hf, wrap=tk.WORD, state=tk.DISABLED, corner_radius=4, font=("Segoe UI",11))
        self.history_textbox.grid(row=0, column=0, padx=5, pady=5, sticky="nsew")
        self.history_textbox.tag_config("user_tag",      foreground="#00529B")
        self.history_textbox.tag_config("assistant_tag", foreground="#006400")
        self.history_textbox.tag_config("prefix_tag",    foreground="gray30")

        # logs
        lf = customtkinter.CTkFrame(main)
        lf.grid(row=1, column=0, sticky="nsew", pady=5)
        lf.grid_columnconfigure(0, weight=1)
        lf.grid_rowconfigure(0, weight=1)
        self.log_textbox = customtkinter.CTkTextbox(lf, wrap=tk.WORD, state=tk.DISABLED,
                                                   corner_radius=4, height=120, font=("Consolas",10))
        self.log_textbox.grid(row=0, column=0, padx=5, pady=5, sticky="nsew")
        for lvl, col in [("debug","gray60"),("info","black"),("warning","#FFA500"),
                         ("error","#FF4136"),("critical","#D0021B"),("success","#00A000")]:
            self.log_textbox.tag_config(lvl, foreground=col)

        # control
        cf = customtkinter.CTkFrame(main, fg_color="transparent")
        cf.grid(row=2, column=0, sticky="ew", pady=5)
        cf.grid_columnconfigure(0, weight=1)
        self.record_button = customtkinter.CTkButton(cf, text="Record", command=self.handle_record_button, height=35)
        self.record_button.grid(row=0, column=0, pady=5)
        self.record_button.configure(state=tk.DISABLED)

        # status
        self.status_label = customtkinter.CTkLabel(self, text="Status:", anchor="w", padx=10)
        self.status_label.grid(row=1, column=0, sticky="ew", pady=(0,5))

    # --- GUI Updaters ---
    def update_status(self, msg: str):
        self.after(0, lambda: self.status_label.configure(text=f"Status: {msg}"))

    def update_gui_log(self, msg: str, sev: str="info"):
        def _():
            self.log_textbox.configure(state=tk.NORMAL)
            self.log_textbox.insert(tk.END, f"{msg}\n", sev)
            self.log_textbox.configure(state=tk.DISABLED)
            self.log_textbox.see(tk.END)
        self.after(0, _)

    def update_history(self, role: str, text: str):
        def _():
            prefix = "You:\n" if role=="user" else "Assistant:\n"
            tag = "user_tag" if role=="user" else "assistant_tag"
            self.history_textbox.configure(state=tk.NORMAL)
            self.history_textbox.insert(tk.END, prefix, ("prefix_tag", tag))
            self.history_textbox.insert(tk.END, f"{text}\n\n", tag)
            self.history_textbox.configure(state=tk.DISABLED)
            self.history_textbox.see(tk.END)
        self.after(0, _)

    def show_about_dialog(self):
        messagebox.showinfo(f"About {self.APP_NAME}",
                            f"{self.APP_NAME} v{self.APP_VERSION}\n\nModular Voice Assistant\n(c) 2025 Nighthawk")

    def change_theme(self, mode: str):
        if mode not in ("light","dark","system"): return
        customtkinter.set_appearance_mode(mode)
        if self.config and self.config.is_loaded:
            self.config.update_value("gui", "theme_preference", mode)
            try: self.config.save_config()
            except Exception as e:
                messagebox.showerror("Config Error", f"Could not save theme: {e}")

    # --- PTT Handling ---
    def handle_record_button(self):
        if not self.initialization_complete:
            return
        if self.audio_manager._is_recording:
            self.audio_manager.stop_recording()
            self._reset_record_button()
        else:
            self.audio_manager.start_recording()
            self.record_button.configure(text="Stop Recording")

    def _bind_ptt(self):
        key = self.config.get_activation_config().get("push_to_talk_key","ctrl+space").lower().split("+")
        mods, k = key[:-1], key[-1]
        tk_mod = [m.capitalize() for m in mods]
        press   = f"<{'-'.join(tk_mod+[k])}>"
        release = f"<KeyRelease-{k}>"
        self.bind(press,   self._on_ptt_press,   add="+")
        self.bind(release, self._on_ptt_release, add="+")
        self._reset_record_button()
        self.update_gui_log(f"PTT Active (Hold {'+'.join(key).upper()})", "info")

    def _on_ptt_press(self, _):
        if self._is_processing_audio or self._ptt_key_pressed: return
        self._ptt_key_pressed = True
        if self.audio_manager and not self.audio_manager._is_recording:
            self.audio_manager.start_recording()

    def _on_ptt_release(self, _):
        if not self._ptt_key_pressed: return
        self._ptt_key_pressed = False
        if self.audio_manager and self.audio_manager._is_recording:
            self.audio_manager.stop_recording()
            self._reset_record_button()

    def _reset_record_button(self):
        key = self.config.get_activation_config().get("push_to_talk_key","ctrl+space").upper()
        self.record_button.configure(text=f"Record (Hold {key})")

    # --- Main Queue Loop ---
    def poll_queue(self):
        try:
            while True:
                msg = gui_queue.get_nowait()
                self._handle_queue(msg)
        except queue.Empty:
            pass
        finally:
            if self.winfo_exists():
                self.after(50, self.poll_queue)

    def _handle_queue(self, msg: dict):
        t, p, sev = msg.get("type"), msg.get("payload"), msg.get("severity","info")
        if t=="log":
            self.update_gui_log(str(p), sev)
        elif t=="status":
            self.update_status(str(p))
        elif t=="init_result":
            if p=="success":
                self.initialization_complete = True
                self.record_button.configure(state=tk.NORMAL)
                self.update_status("Ready")
                self._bind_ptt()
            else:
                self.initialization_error = str(p)
                self.update_status(f"ERROR: {p}")
                self.update_gui_log(f"Init failed: {p}", "critical")
        elif t=="audio_ready":
            fp = p.get("filepath"); dur=p.get("duration",0.0)
            if fp:
                self._start_stt(fp, dur)
            else:
                self.update_status("ERROR: Audio Prep Failed")
                self._unlock()
        elif t=="stt_result":
            text = p.get("text")
            if text is None:
                self.update_gui_log("STT failed.", "error")
                self._unlock()
            else:
                if text.strip():
                    self.update_history("user", text)
                    self.context_manager.add_message("user", text)
                    self.update_status("Thinking...")
                    self._start_assistant_stream()
                    self.llm_manager.run_llm_in_background(text)
                else:
                    self.update_gui_log("Empty transcription.", "warning")
                    self._unlock()
        elif t=="llm_chunk":
            self._append_assistant_stream(p.get("delta",""))
        elif t=="llm_result":
            if p.get("error"):
                self.update_gui_log(f"LLM Error: {p.get('error_message')}", "error")
                self.update_status("Ready")
                self._unlock()
            else:
                full = p.get("text","")
                self._finish_assistant_stream()
                self.context_manager.add_message("assistant", full)
                self.update_status("Synthesizing…")
                threading.Thread(target=self._run_tts, args=(full,), daemon=True).start()
        elif t=="shutdown_request":
            self.quit_app()

    # --- STT → LLM → TTS helpers ---
    def _start_stt(self, filepath: str, duration: float):
        if not self._processing_lock.acquire(blocking=False):
            return
        self._is_processing_audio = True
        self.update_status(f"Transcribing {duration:.1f}s…")
        threading.Thread(target=self._stt_thread, args=(filepath,), daemon=True).start()

    def _stt_thread(self, filepath: str):
        try:
            txt = self.stt_manager.transcribe(filepath)
            gui_queue.put({"type":"stt_result","payload":{"text":txt}})
        except Exception as e:
            logger.error(f"STT thread error: {e}", exc_info=True)
            gui_queue.put({"type":"stt_result","payload":{"text":None}})
        finally:
            self.after(0, self._unlock)

    def _start_assistant_stream(self):
        self._assistant_streaming = True
        self.history_textbox.configure(state=tk.NORMAL)
        self.history_textbox.insert(tk.END, "Assistant:\n", ("prefix_tag","assistant_tag"))
        self.history_textbox.configure(state=tk.DISABLED)

    def _append_assistant_stream(self, delta: str):
        if not self._assistant_streaming: return
        self.history_textbox.configure(state=tk.NORMAL)
        self.history_textbox.insert(tk.END, delta, "assistant_tag")
        self.history_textbox.configure(state=tk.DISABLED)
        self.history_textbox.see(tk.END)

    def _finish_assistant_stream(self):
        self._assistant_streaming = False
        self.history_textbox.configure(state=tk.NORMAL)
        self.history_textbox.insert(tk.END, "\n\n", "assistant_tag")
        self.history_textbox.configure(state=tk.DISABLED)

    def _run_tts(self, text: str):
        try:
            paths = self.tts_manager.synthesize(text)
            for p in paths:
                self.audio_manager.play_audio_file(p)
        except Exception as e:
            logger.error(f"TTS error: {e}", exc_info=True)
            self.update_gui_log(f"TTS failed: {e}", "error")
        finally:
            self.after(0, lambda: self.update_status("Ready"))

    def _unlock(self):
        if self._processing_lock.locked():
            self._processing_lock.release()
        self._is_processing_audio = False

    # --- Init sequence ---
    def start_backend_initialization(self):
        try:
            Path("logs").mkdir(exist_ok=True)
            Path("data").mkdir(exist_ok=True)

            gui_queue.put({"type":"status","payload":"Loading config…"})
            self.config = Config()
            self.config.load()
            gui_queue.put({"type":"log","payload":"Configuration loaded."})

            gui_queue.put({"type":"status","payload":"Init SystemManager…"})
            self.system_manager = SystemManager(self.config)
            gui_queue.put({"type":"log","payload":"System Manager initialized."})

            gui_queue.put({"type":"status","payload":"Configuring logging…"})
            self.system_manager.setup_logging()
            gui_queue.put({"type":"log","payload":"Logging configured.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Logging system info…"})
            self.system_manager.log_system_info()
            gui_queue.put({"type":"log","payload":"System info logged."})

            gui_queue.put({"type":"status","payload":"Checking requirements…"})
            self.system_manager.check_requirements()
            gui_queue.put({"type":"log","payload":"Requirements passed.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Initializing ContextManager…"})
            self.context_manager = ContextManager(self.config)
            gui_queue.put({"type":"log","payload":"ContextManager initialized.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Initializing AudioManager…"})
            self.audio_manager = AudioManager(self.config, gui_queue)
            gui_queue.put({"type":"log","payload":"AudioManager initialized.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Initializing STTManager…"})
            self.stt_manager = STTManager(self.config)
            gui_queue.put({"type":"log","payload":"STTManager initialized.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Initializing LLMManager…"})
            self.llm_manager = LLMManager(self.config, gui_queue)
            # attach context for future prompt handling
            self.llm_manager.set_context_manager(self.context_manager)
            gui_queue.put({"type":"log","payload":"LLMManager initialized.","severity":"success"})

            gui_queue.put({"type":"status","payload":"Initializing TTSManager…"})
            self.tts_manager = TTSManager(self.config)
            gui_queue.put({"type":"log","payload":"TTSManager initialized.","severity":"success"})

            gui_queue.put({"type":"init_result","payload":"success"})
        except Exception as e:
            msg = getattr(e, "message", str(e))
            logger.critical(f"BackendInitThread error: {msg}", exc_info=True)
            gui_queue.put({"type":"init_result","payload":msg})

    # --- Shutdown ---
    def quit_app(self):
        logger.info("Shutdown requested.")
        self.update_status("Shutting down…")
        if self.audio_manager:
            try: self.audio_manager.stop()
            except: pass
        self.destroy()


# --- Signal hook ---
app_instance = None
def handle_sigint(signum, frame):
    if app_instance:
        gui_queue.put({"type":"shutdown_request"})
    else:
        sys.exit(1)

def run_application():
    global app_instance
    try:
        signal.signal(signal.SIGINT, handle_sigint)
        app_instance = MiraiGuiApp()
        app_instance.mainloop()
    except KeyboardInterrupt:
        if app_instance:
            app_instance.quit_app()
    except Exception as e:
        tb = traceback.format_exc()
        print(f"FATAL ERROR: {e}\n{tb}", file=sys.stderr)
        logging.getLogger(__name__).critical("Unhandled error", exc_info=True)

if __name__ == "__main__":
    run_application()


================================================
FILE: modules/__init__.py
================================================
# modules/__init__.py

# This file marks the 'modules' directory as a Python package.
# It can remain empty or be used for package-level initializations if needed later.

================================================
FILE: modules/audio_manager.py
================================================
# modules/audio_manager.py

import logging
import queue
import threading
import time
import wave
from pathlib import Path
from typing import Optional, List, Dict, Any, Union

import numpy as np
import pyaudio # Needs pip install PyAudio

from .config import Config

logger = logging.getLogger(__name__)

# Custom Exception
class AudioManagerError(Exception):
    """Custom exception for AudioManager specific errors."""
    pass

class AudioManager:
    """
    Handles audio input (recording via PTT) and output (playback).

    Uses PyAudio for cross-platform audio stream management.
    Communicates status and results via a shared queue with the GUI.
    """

    def __init__(self, config: Config, gui_queue: queue.Queue):
        """
        Initialize the AudioManager.

        Args:
            config: The application configuration object.
            gui_queue: Queue for sending messages (status, results) to the GUI thread.

        Raises:
            AudioManagerError: If PyAudio initialization fails or no devices are found.
        """
        logger.info("Initializing AudioManager...")
        self.config = config
        self.gui_queue = gui_queue

        # Config shortcuts
        self.audio_config = config.get_audio_config()
        # self.activation_config = config.get_activation_config() # Keep for future PTT key config

        # Audio settings
        self.sample_rate: int = self.audio_config.get("sample_rate", 16000) # Default 16kHz common for STT
        self.chunk_size: int = self.audio_config.get("record_chunk_size", 1024)
        self.channels: int = 1 # Force mono for simplicity
        self.format: int = pyaudio.paInt16 # Use 16-bit PCM
        self.format_np = np.int16 # Numpy equivalent dtype

        # State variables
        self._p: Optional[pyaudio.PyAudio] = None
        self._input_stream: Optional[pyaudio.Stream] = None
        self._output_stream: Optional[pyaudio.Stream] = None
        self._recording_thread: Optional[threading.Thread] = None
        self._playback_thread: Optional[threading.Thread] = None
        self._recording_frames: List[bytes] = []
        self._is_recording: bool = False
        self._is_playing: bool = False
        self._stop_playback_event = threading.Event() # Used to signal playback thread to stop

        try:
            self._p = pyaudio.PyAudio()
            logger.debug("PyAudio instance created.")
        except Exception as e:
            logger.critical(f"Failed to initialize PyAudio: {e}", exc_info=True)
            raise AudioManagerError(f"PyAudio initialization failed: {e}") from e

        # Get device indices after PyAudio is initialized
        self.input_device_index: Optional[int] = self._get_device_index("input_device")
        self.output_device_index: Optional[int] = self._get_device_index("output_device")

        if self.input_device_index is None:
            logger.error("No suitable input device found. Recording will not work.")
            # raise AudioManagerError("No suitable input device found.") # Optionally make this fatal
        if self.output_device_index is None:
            logger.error("No suitable output device found. Playback will not work.")
            # raise AudioManagerError("No suitable output device found.") # Optionally make this fatal

        self._log_device_info()
        logger.info("AudioManager initialized successfully.")


    def _get_device_index(self, device_key: str) -> Optional[int]:
        """Finds device index based on config (null, index, or partial name)."""
        if not self._p: return None # PyAudio not initialized

        device_config: Union[None, int, str] = self.audio_config.get(device_key)
        device_type = "input" if "input" in device_key else "output"
        is_input = device_type == "input"
        logger.debug(f"Finding {device_type} device for config value: {device_config}")

        # 1. Handle None -> Use Default Device
        if device_config is None:
            try:
                if is_input:
                    info = self._p.get_default_input_device_info()
                else:
                    info = self._p.get_default_output_device_info()
                index = info['index']
                logger.info(f"Using default {device_type} device: {info['name']} (Index: {index})")
                return index
            except IOError as e:
                logger.error(f"Could not get default {device_type} device: {e}. Please ensure audio devices are connected and enabled.")
                # Try to find *any* valid device as a fallback
                return self._find_first_valid_device(is_input)
            except Exception as e:
                 logger.error(f"Unexpected error getting default {device_type} device: {e}", exc_info=True)
                 return self._find_first_valid_device(is_input)

        # 2. Handle Integer -> Use Specific Index
        if isinstance(device_config, int):
            try:
                info = self._p.get_device_info_by_index(device_config)
                required_channels = info.get('maxInputChannels' if is_input else 'maxOutputChannels', 0)
                if required_channels > 0:
                    logger.info(f"Using configured {device_type} device index: {info['name']} (Index: {device_config})")
                    return device_config
                else:
                     logger.warning(f"Device index {device_config} is not a valid {device_type} device (channels={required_channels}). Falling back.")
                     return self._find_first_valid_device(is_input)
            except OSError:
                logger.warning(f"Device index {device_config} not found. Falling back.")
                return self._find_first_valid_device(is_input)
            except Exception as e:
                 logger.error(f"Error checking device index {device_config}: {e}", exc_info=True)
                 return self._find_first_valid_device(is_input)

        # 3. Handle String -> Find by Partial Name Match
        if isinstance(device_config, str):
            try:
                target_name = device_config.lower()
                logger.debug(f"Searching for {device_type} device containing name: '{target_name}'")
                best_match = -1
                device_count = self._p.get_device_count()
                for i in range(device_count):
                    info = self._p.get_device_info_by_index(i)
                    required_channels = info.get('maxInputChannels' if is_input else 'maxOutputChannels', 0)
                    device_name = info.get('name', '').lower()
                    if required_channels > 0 and target_name in device_name:
                        best_match = i
                        logger.debug(f" > Match found: Index {i}, Name: {info.get('name')}")
                        break # Found first match

                if best_match != -1:
                    info = self._p.get_device_info_by_index(best_match)
                    logger.info(f"Using configured {device_type} device name match: {info['name']} (Index: {best_match})")
                    return best_match
                else:
                    logger.warning(f"No {device_type} device found matching name '{device_config}'. Falling back.")
                    return self._find_first_valid_device(is_input)
            except Exception as e:
                 logger.error(f"Error searching for device name '{device_config}': {e}", exc_info=True)
                 return self._find_first_valid_device(is_input)

        # Fallback for invalid config type
        logger.warning(f"Invalid configuration type for {device_key}: {type(device_config)}. Falling back.")
        return self._find_first_valid_device(is_input)

    def _find_first_valid_device(self, is_input: bool) -> Optional[int]:
        """Finds the first available device of the specified type."""
        if not self._p: return None
        device_type = "input" if is_input else "output"
        logger.debug(f"Searching for first available {device_type} device...")
        try:
            device_count = self._p.get_device_count()
            for i in range(device_count):
                 info = self._p.get_device_info_by_index(i)
                 required_channels = info.get('maxInputChannels' if is_input else 'maxOutputChannels', 0)
                 if required_channels > 0:
                     logger.info(f"Fallback: Found valid {device_type} device: {info['name']} (Index: {i})")
                     return i
            logger.error(f"Fallback failed: No valid {device_type} devices found.")
            return None
        except Exception as e:
            logger.error(f"Error during fallback device search: {e}", exc_info=True)
            return None

    def _log_device_info(self):
        """Logs information about the selected audio devices."""
        if not self._p: return
        try:
            if self.input_device_index is not None:
                 info = self._p.get_device_info_by_index(self.input_device_index)
                 logger.info(f"Selected Input Device: Index={self.input_device_index}, Name='{info.get('name')}', Rate={info.get('defaultSampleRate')}, Channels={info.get('maxInputChannels')}")
            else:
                 logger.warning("No valid input device selected/found.")

            if self.output_device_index is not None:
                 info = self._p.get_device_info_by_index(self.output_device_index)
                 logger.info(f"Selected Output Device: Index={self.output_device_index}, Name='{info.get('name')}', Rate={info.get('defaultSampleRate')}, Channels={info.get('maxOutputChannels')}")
            else:
                 logger.warning("No valid output device selected/found.")
        except Exception as e:
            logger.error(f"Failed to log device info: {e}", exc_info=True)

    # --- Recording ---

    def _recording_loop(self):
        """Target function for the recording thread."""
        logger.info(f"Recording thread started (Device: {self.input_device_index}).")
        stream = None # Define stream here for finally block
        try:
            if self.input_device_index is None:
                 raise AudioManagerError("Cannot record: No valid input device selected.")

            stream = self._p.open(format=self.format,
                                  channels=self.channels,
                                  rate=self.sample_rate,
                                  input=True,
                                  frames_per_buffer=self.chunk_size,
                                  input_device_index=self.input_device_index)
            self._input_stream = stream # Store reference if needed elsewhere
            logger.info(f"Input stream opened (Rate: {self.sample_rate}, Chunk: {self.chunk_size}).")
            self._recording_frames.clear() # Ensure buffer is clean

            while self._is_recording: # Check flag controlled by stop_recording
                try:
                    data = stream.read(self.chunk_size, exception_on_overflow=False)
                    self._recording_frames.append(data)
                except IOError as e:
                    if e.errno == pyaudio.paInputOverflowed:
                        logger.warning("Input overflowed during recording.")
                    else:
                        logger.error(f"IOError during recording stream read: {e}")
                        # Decide if we should stop recording here
                        # self._is_recording = False # maybe?
                except Exception as e:
                     logger.error(f"Unexpected error in recording loop: {e}", exc_info=True)
                     # self._is_recording = False # maybe?

            logger.info("Recording flag turned off, exiting loop.")

        except AudioManagerError as e: # Catch specific error
            logger.error(f"{e}")
            self.gui_queue.put({"type": "status", "payload": f"ERROR: {e}"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: {e}", "severity": "error"})
        except Exception as e:
            logger.error(f"Failed to open or read from input stream: {e}", exc_info=True)
            self.gui_queue.put({"type": "status", "payload": "ERROR: Mic Access Failed"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: Mic stream failed: {e}", "severity": "error"})
        finally:
            if stream:
                try:
                    if stream.is_active(): stream.stop_stream()
                    stream.close()
                    logger.info("Input stream stopped and closed.")
                except Exception as e:
                    logger.error(f"Error closing input stream: {e}", exc_info=True)
            self._input_stream = None
            self._is_recording = False # Ensure flag is definitely false on exit
            logger.info("Recording thread finished.")


    def start_recording(self):
        """Starts PTT recording in a background thread."""
        if self._is_recording:
            logger.warning("Recording requested but already in progress.")
            return
        if self.input_device_index is None:
             logger.error("Cannot start recording: No valid input device.")
             self.gui_queue.put({"type": "status", "payload": "ERROR: No Mic Device"})
             return

        logger.info("PTT Pressed - Starting recording...")
        self._is_recording = True
        # self._stop_playback_event.clear() # Recording doesn't use this event
        self.gui_queue.put({"type": "status", "payload": "Recording..."})
        self.gui_queue.put({"type": "log", "payload": "Recording started...", "severity": "info"})

        # Start the recording thread
        self._recording_thread = threading.Thread(target=self._recording_loop, daemon=True, name="RecordingThread")
        self._recording_thread.start()

    def stop_recording(self):
        """Stops PTT recording, saves audio, and notifies GUI."""
        if not self._is_recording:
            # This might happen if stop is called rapidly after start
            logger.debug("Stop recording called but not currently recording.")
            return

        logger.info("PTT Released - Stopping recording...")
        self._is_recording = False # Signal the recording loop to stop

        # Wait for the recording thread to finish processing its last chunk and exit
        if self._recording_thread and self._recording_thread.is_alive():
             logger.debug("Waiting for recording thread to finish...")
             self._recording_thread.join(timeout=1.0) # Generous timeout
             if self._recording_thread.is_alive():
                  logger.warning("Recording thread did not exit cleanly after 1 second.")
        self._recording_thread = None

        if not self._recording_frames:
            logger.warning("No audio frames were recorded.")
            self.gui_queue.put({"type": "status", "payload": "Ready"}) # Reset status
            self.gui_queue.put({"type": "log", "payload": "Recording stopped (no data).", "severity": "warning"})
            return

        output_filename = self.audio_config.get("input_filename", "temp_input.wav")
        output_path = Path(output_filename)
        # Ensure parent directory exists
        output_path.parent.mkdir(parents=True, exist_ok=True)

        logger.info(f"Saving recorded audio ({len(self._recording_frames)} frames) to {output_path.resolve()}...")

        try:
            wf = wave.open(str(output_path), 'wb')
            wf.setnchannels(self.channels)
            wf.setsampwidth(self._p.get_sample_size(self.format))
            wf.setframerate(self.sample_rate)
            wf.writeframes(b''.join(self._recording_frames))
            wf.close()
            duration = len(self._recording_frames) * self.chunk_size / self.sample_rate
            logger.info(f"Audio saved successfully ({duration:.2f}s).")

            # Notify GUI that audio is ready for processing
            self.gui_queue.put({
                "type": "audio_ready",
                "payload": {"filepath": str(output_path.resolve()), "duration": duration}
            })
            self.gui_queue.put({"type": "status", "payload": "Processing Audio..."})
            self.gui_queue.put({"type": "log", "payload": f"Recording stopped ({duration:.2f}s). Audio ready.", "severity": "info"})

        except Exception as e:
            logger.error(f"Failed to save recorded audio: {e}", exc_info=True)
            self.gui_queue.put({"type": "status", "payload": "ERROR: Save Failed"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: Failed to save audio: {e}", "severity": "error"})
        finally:
             self._recording_frames.clear() # Clear buffer after saving (or attempting to)


    # --- Playback ---

    def _playback_loop(self, filepath: str):
        """Target function for the playback thread."""
        logger.info(f"Playback thread started for: {filepath}")
        wf = None
        stream = None
        try:
            if self.output_device_index is None:
                 raise AudioManagerError("Cannot play audio: No valid output device selected.")

            wf = wave.open(filepath, 'rb')
            logger.debug(f"Opened WAV file: Channels={wf.getnchannels()}, Rate={wf.getframerate()}, Width={wf.getsampwidth()}")

            stream = self._p.open(format=self._p.get_format_from_width(wf.getsampwidth()),
                                  channels=wf.getnchannels(),
                                  rate=wf.getframerate(),
                                  output=True,
                                  frames_per_buffer=self.chunk_size,
                                  output_device_index=self.output_device_index)
            self._output_stream = stream
            logger.info(f"Output stream opened (Device: {self.output_device_index}).")

            data = wf.readframes(self.chunk_size)
            while data and not self._stop_playback_event.is_set():
                stream.write(data)
                data = wf.readframes(self.chunk_size)

            if self._stop_playback_event.is_set():
                 logger.info("Playback loop interrupted by stop event.")
            else:
                 logger.info("Playback loop finished normally.")

        except FileNotFoundError:
             logger.error(f"Playback failed: File not found at {filepath}")
             self.gui_queue.put({"type": "status", "payload": "ERROR: Playback File Not Found"})
             self.gui_queue.put({"type": "log", "payload": f"ERROR: Playback file not found: {filepath}", "severity": "error"})
        except AudioManagerError as e:
            logger.error(f"{e}")
            self.gui_queue.put({"type": "status", "payload": f"ERROR: {e}"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: {e}", "severity": "error"})
        except Exception as e:
            logger.error(f"Error during playback: {e}", exc_info=True)
            self.gui_queue.put({"type": "status", "payload": f"ERROR: Playback Failed"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: Playback failed: {e}", "severity": "error"})
        finally:
            if stream:
                try:
                    # Wait for stream to finish processing buffers before stopping/closing
                    while stream.is_active():
                        time.sleep(0.1)
                        logger.debug("Waiting for playback stream to finish...")
                    stream.stop_stream()
                    stream.close()
                    logger.info("Output stream stopped and closed.")
                except Exception as e:
                    logger.error(f"Error closing output stream: {e}", exc_info=True)
            if wf:
                wf.close()

            self._output_stream = None
            was_stopped = self._stop_playback_event.is_set() # Check if stopped externally
            self._is_playing = False # Ensure flag reset BEFORE putting final status
            # Send final status ONLY if playback finished normally (wasn't stopped)
            if not was_stopped:
                 self.gui_queue.put({"type": "status", "payload": "Ready"})
                 self.gui_queue.put({"type": "log", "payload": "Playback finished.", "severity": "info"})
            logger.info("Playback thread finished.")


    def play_audio_file(self, filepath: str):
        """Plays an audio file in a background thread."""
        if self._is_playing:
            logger.warning("Playback requested but already in progress. Stopping previous playback.")
            self.stop_playback() # Stop current playback cleanly
            # Give a moment for resources to potentially release
            time.sleep(0.2) # Increased delay slightly

        if not Path(filepath).exists():
            logger.error(f"Cannot play audio: File not found at {filepath}")
            self.gui_queue.put({"type": "status", "payload": "ERROR: Playback File Not Found"})
            self.gui_queue.put({"type": "log", "payload": f"ERROR: Playback file not found: {filepath}", "severity": "error"})
            return
        if self.output_device_index is None:
             logger.error("Cannot play audio: No valid output device.")
             self.gui_queue.put({"type": "status", "payload": "ERROR: No Speaker Device"})
             return

        logger.info(f"Starting playback for: {filepath}")
        self._is_playing = True
        self._stop_playback_event.clear() # Ensure stop signal is clear for new playback
        self.gui_queue.put({"type": "status", "payload": "Playing Audio..."})
        self.gui_queue.put({"type": "log", "payload": f"Playing {Path(filepath).name}...", "severity": "info"})

        self._playback_thread = threading.Thread(target=self._playback_loop, args=(filepath,), daemon=True, name="PlaybackThread")
        self._playback_thread.start()

    def stop_playback(self):
        """Signals the playback thread to stop and waits briefly."""
        if not self._is_playing:
            # logger.debug("No playback in progress to stop.")
            return

        logger.info("Requesting playback stop...")
        self._stop_playback_event.set() # Signal thread

        # Wait briefly for the thread to potentially finish its current write and exit loop
        if self._playback_thread and self._playback_thread.is_alive():
            logger.debug("Waiting briefly for playback thread...")
            self._playback_thread.join(timeout=0.5) # Adjust timeout if needed
            if self._playback_thread.is_alive():
                 logger.warning("Playback thread did not exit cleanly after stop signal.")
        self._playback_thread = None
        self._is_playing = False # Update state *after* trying to join

        # Send status update regardless of whether thread joined immediately
        self.gui_queue.put({"type": "status", "payload": "Playback Stopped"})
        self.gui_queue.put({"type": "log", "payload": "Playback stopped by user.", "severity": "warning"})
        logger.info("Playback stop requested.")

    # --- Shutdown ---
    def stop(self):
        """Stops all audio activities and cleans up resources."""
        logger.info("AudioManager stopping...")
        self._shutting_down = True # Use internal flag if needed
        self._stop_playback_event.set() # Signal playback thread

        if self._is_recording:
            logger.info("Stopping active recording...")
            self.stop_recording() # Ensure recording state is cleared and thread joined

        if self._is_playing:
            logger.info("Stopping active playback...")
            self.stop_playback() # Ensure playback state is cleared and thread joined

        # Explicitly join threads again just in case stop methods were interrupted
        if self._recording_thread and self._recording_thread.is_alive():
            logger.warning("Recording thread still alive during final stop. Joining...")
            self._recording_thread.join(timeout=1.0)
        if self._playback_thread and self._playback_thread.is_alive():
            logger.warning("Playback thread still alive during final stop. Joining...")
            self._playback_thread.join(timeout=1.0)

        # Terminate PyAudio
        if self._p:
            logger.info("Terminating PyAudio instance.")
            try:
                self._p.terminate()
            except Exception as e:
                 logger.error(f"Error terminating PyAudio: {e}", exc_info=True)
            finally:
                 self._p = None

        # TODO: Add porcupine_handle.delete() here when hotword is added back

        logger.info("AudioManager stopped successfully.")

================================================
FILE: modules/config.py
================================================
# modules/config.py
import sys
import yaml
import os
from pathlib import Path
import logging
from typing import Any, Dict, Optional

# Note: Logging might not be fully configured when this module is first imported.
# Initial log messages might go to stderr if setup_logging hasn't run.
logger = logging.getLogger(__name__)

class ConfigError(Exception):
    """Custom exception for configuration errors."""
    pass

class Config:
    """
    Manages loading and accessing configuration settings from a YAML file using a Singleton pattern.

    Handles basic validation and provides convenient access methods for
    different configuration sections. Allows saving updates back to the file.
    """
    _instance = None
    _config_path: Path = Path("config.yaml")
    _config_data: Dict[str, Any] = {}
    _loaded = False

    def __new__(cls, config_path: Optional[str | Path] = None):
        if cls._instance is None:
            cls._instance = super(Config, cls).__new__(cls)
            # Ensure config path is set before attempting load
            if config_path:
                cls._config_path = Path(config_path)
            # Don't load here automatically, require explicit load call
            # after logging might be partially set up.
        return cls._instance

    def load(self) -> None:
        """Loads configuration from the YAML file. Should be called explicitly after instantiation."""
        if self._loaded:
            logger.debug("Configuration already loaded.")
            return

        if not self._config_path.exists():
            message = f"Configuration file not found: {self._config_path.resolve()}"
            # Cannot use logger here reliably before setup
            print(f"FATAL ERROR: {message}", file=sys.stderr)
            raise ConfigError(message)

        try:
            with open(self._config_path, 'r', encoding='utf-8') as f:
                loaded_data = yaml.safe_load(f)
            if not isinstance(loaded_data, dict):
                raise ValueError("Configuration file is not a valid YAML dictionary.")
            self._config_data = loaded_data
            self._loaded = True
            # Use logger here as load() is called after basic logging might be up
            logger.info(f"Configuration loaded successfully from {self._config_path.resolve()}")
        except yaml.YAMLError as e:
            message = f"Error parsing configuration file {self._config_path.resolve()}: {e}"
            print(f"FATAL ERROR: {message}", file=sys.stderr) # Fallback print
            logger.critical(message, exc_info=True)
            raise ConfigError(message) from e
        except Exception as e:
            message = f"Failed to read configuration file {self._config_path.resolve()}: {e}"
            print(f"FATAL ERROR: {message}", file=sys.stderr) # Fallback print
            logger.critical(message, exc_info=True)
            raise ConfigError(message) from e

    def get(self, section: str, key: Optional[str] = None, default: Any = None) -> Any:
        """
        Gets a configuration value from a specific section and key.

        Args:
            section (str): The top-level section key (e.g., 'audio', 'llm').
            key (Optional[str]): The key within the section. If None, returns the whole section dict.
            default (Any): The default value to return if the section or key is not found.

        Returns:
            Any: The configuration value or the default. Returns None if section exists but key is missing and no default is provided.
        """
        if not self._loaded:
            logger.warning("Attempted to get config value before configuration was loaded.")
            return default # Or raise error? For now, return default.

        section_data = self._config_data.get(section)
        if section_data is None:
            return default

        if key is None:
            return section_data # Return whole section

        # Use get() with default for the key within the section
        return section_data.get(key, default)


    def get_section(self, section: str, default: Optional[Dict] = None) -> Dict[str, Any]:
        """
        Gets an entire configuration section as a dictionary.

        Args:
            section (str): The top-level section key.
            default (Optional[Dict]): Default dictionary if section not found. Defaults to {}.

        Returns:
            Dict[str, Any]: The configuration section dictionary or the default.
        """
        if not self._loaded:
            logger.warning("Attempted to get config section before configuration was loaded.")
            return default if default is not None else {}

        return self._config_data.get(section, default if default is not None else {})

    # --- Convenience Methods ---
    def get_audio_config(self) -> Dict[str, Any]: return self.get_section('audio')
    def get_activation_config(self) -> Dict[str, Any]: return self.get_section('activation')
    def get_stt_config(self) -> Dict[str, Any]: return self.get_section('stt')
    def get_tts_config(self) -> Dict[str, Any]: return self.get_section('tts')
    def get_llm_config(self) -> Dict[str, Any]: return self.get_section('llm')
    def get_context_manager_config(self) -> Dict[str, Any]: return self.get_section('context_manager')
    def get_logging_config(self) -> Dict[str, Any]: return self.get_section('logging')

    def save_config(self) -> None:
        """Saves the current in-memory configuration data back to the YAML file."""
        if not self._loaded:
             logger.error("Cannot save configuration - configuration was never loaded.")
             return # Or raise error?

        try:
            # Ensure necessary directories exist based on current config data
            log_dir = Path(self.get_logging_config().get("log_directory", "logs"))
            log_dir.mkdir(parents=True, exist_ok=True)
            context_path = Path(self.get_context_manager_config().get("storage_path", "data/context.json"))
            context_path.parent.mkdir(parents=True, exist_ok=True)

            with open(self._config_path, 'w', encoding='utf-8') as f:
                yaml.dump(self._config_data, f, default_flow_style=False, sort_keys=False, indent=2)
            logger.info(f"Configuration saved successfully to {self._config_path.resolve()}")
        except Exception as e:
            logger.error(f"Failed to save configuration to {self._config_path.resolve()}: {e}", exc_info=True)
            # Optionally raise an error to be handled by the UI save action
            # raise ConfigError(f"Failed to save config: {e}") from e

    def update_value(self, section: str, key: str, value: Any) -> None:
        """
        Updates a specific configuration value in memory.
        Call save_config() to persist changes.
        """
        if not self._loaded:
             logger.error("Cannot update configuration value - configuration was never loaded.")
             return # Or raise error?

        if section not in self._config_data:
            self._config_data[section] = {}

        if isinstance(self._config_data.get(section), dict):
            self._config_data[section][key] = value
            logger.debug(f"Updated config (in memory): [{section}].{key} = {value}")
        else:
            # This case should ideally not happen if config structure is maintained
            logger.error(f"Cannot update key '{key}': Section '[{section}]' exists but is not a dictionary.")
            # Consider raising an error or trying to fix the structure
            # For now, log the error. If it was None, initialize it.
            if self._config_data.get(section) is None:
                 self._config_data[section] = {key: value}
                 logger.warning(f"Initialized section '[{section}]' as dict to update key '{key}'.")


    @property
    def is_loaded(self) -> bool:
        """Returns True if the configuration has been successfully loaded."""
        return self._loaded

    @property
    def config_data(self) -> Dict[str, Any]:
        """Returns a read-only copy of the internal config dictionary."""
        # Return a copy to prevent accidental direct modification
        return self._config_data.copy()

================================================
FILE: modules/context_manager.py
================================================
# modules/context_manager.py

import json
from pathlib import Path
from typing import List, Dict, Any

from .config import Config

class ContextManagerError(Exception):
    pass

class ContextManager:
    """
    Manages conversation history for STT → LLM → TTS.
    Persists to disk, enforces a max‐context size, and
    provides messages in the OpenAI chat format.
    """

    def __init__(self, config: Config):
        cfg = config.get_section("context_manager", {})
        self.storage_path: Path = Path(cfg.get("storage_path", "data/conversation_state.json"))
        self.strategy: str    = cfg.get("condensation_strategy", "truncate")
        self.threshold: float = float(cfg.get("condensation_threshold", 0.85))
        # max_context_tokens is in llm config
        llm_cfg = config.get_llm_config()
        self.max_tokens: int = int(llm_cfg.get("max_context_tokens", 4096))

        # load existing history or start fresh
        self.messages: List[Dict[str, str]] = []
        self._load()

    def _load(self):
        if self.storage_path.exists():
            try:
                data = json.loads(self.storage_path.read_text(encoding="utf-8"))
                if isinstance(data, list):
                    self.messages = data
            except Exception as e:
                raise ContextManagerError(f"Failed to load context: {e}")

    def _save(self):
        try:
            self.storage_path.parent.mkdir(parents=True, exist_ok=True)
            self.storage_path.write_text(json.dumps(self.messages, ensure_ascii=False, indent=2), encoding="utf-8")
        except Exception as e:
            raise ContextManagerError(f"Failed to save context: {e}")

    def add_message(self, role: str, content: str):
        """
        Add either "user" or "assistant" messages.
        """
        if role not in ("user", "assistant"):
            raise ContextManagerError(f"Invalid role: {role}")
        self.messages.append({"role": role, "content": content})
        self._condense_if_needed()
        self._save()

    def get_context_for_llm(self) -> List[Dict[str, str]]:
        """
        Returns a copy of messages suitable to pass directly
        as the `messages` parameter to OpenAI's chat API.
        """
        return list(self.messages)

    def _condense_if_needed(self):
        """
        If total tokens exceed threshold * max_tokens,
        drop oldest messages (simple truncate strategy).
        """
        # --- naive token estimate: 1 token ≈ 4 chars ---
        total_chars = sum(len(m["content"]) for m in self.messages)
        estimated_tokens = total_chars / 4

        limit = self.threshold * self.max_tokens
        if estimated_tokens <= limit:
            return

        # drop oldest until under threshold
        while self.messages and estimated_tokens > limit:
            dropped = self.messages.pop(0)
            total_chars -= len(dropped["content"])
            estimated_tokens = total_chars / 4
        # you could implement more advanced condensation here


================================================
FILE: modules/llm_manager.py
================================================
# modules/llm_manager.py

import logging
import os
import queue
import threading
import asyncio
from typing import Optional, List, Dict, Any

# Import OpenAI library (v1.0+)
try:
    from openai import (
        OpenAI,
        AsyncOpenAI,
        APIError,
        APIConnectionError,
        APITimeoutError,
        RateLimitError,
        InternalServerError,
    )
    OPENAI_AVAILABLE = True
except ImportError:
    OPENAI_AVAILABLE = False

    class OpenAI: pass
    class AsyncOpenAI: pass

    class APIError(Exception): pass
    class APIConnectionError(APIError): pass
    class APITimeoutError(APIConnectionError): pass

    class RateLimitError(APIError):
        status_code = 429
        message = "Rate limit exceeded."

    class InternalServerError(APIError):
        status_code = 500
        message = "Internal server error."

from .config import Config
from .context_manager import ContextManager

logger = logging.getLogger(__name__)

class LLMManagerError(Exception):
    """Custom exception for LLMManager specific errors."""
    pass

class LLMManager:
    """
    Handles communication with an OpenAI-compatible LLM API endpoint,
    including context management for conversation history.
    """

    def __init__(self, config: Config, gui_queue: queue.Queue):
        logger.info("Initializing LLMManager...")
        if not OPENAI_AVAILABLE:
            raise LLMManagerError("openai library not installed. Run: uv add openai tiktoken")

        self.config = config
        self.gui_queue = gui_queue
        self.llm_config = config.get_llm_config()

        # ContextManager will be linked later
        self.context_manager: Optional[ContextManager] = None

        # API settings
        self.api_base_url: Optional[str] = self.llm_config.get("api_base_url")
        self.api_key_env_var: str = self.llm_config.get("api_key_env_var", "NONE")
        self.model_name: Optional[str] = self.llm_config.get("model_name")

        self.default_temperature: float = self.llm_config.get("default_temperature", 0.7)
        self.default_max_tokens: int = self.llm_config.get("default_max_tokens", 512)

        # Read API key if specified
        self.api_key: Optional[str] = None
        if self.api_key_env_var and self.api_key_env_var.upper() != "NONE":
            self.api_key = os.environ.get(self.api_key_env_var)
            if not self.api_key:
                msg = f"LLM API key env var '{self.api_key_env_var}' not found."
                logger.warning(msg)
                self.gui_queue.put({"type": "log", "payload": msg, "severity": "warning"})

        if not self.api_base_url:
            raise LLMManagerError("LLM 'api_base_url' missing in config.")
        if not self.model_name:
            raise LLMManagerError("LLM 'model_name' missing in config.")

        # Initialize AsyncOpenAI client
        try:
            self.client = AsyncOpenAI(
                base_url=self.api_base_url,
                api_key=self.api_key or "dummy-key",
                timeout=120.0,
                max_retries=1
            )
            logger.info(f"AsyncOpenAI client initialized at {self.api_base_url}")
        except Exception as e:
            logger.critical(f"Failed to init OpenAI client: {e}", exc_info=True)
            raise LLMManagerError(f"OpenAI client initialization failed: {e}") from e

        self.is_processing = False
        logger.info("LLMManager initialized successfully.")

    def set_context_manager(self, context_manager: ContextManager) -> None:
        """
        Link the shared ContextManager for storing and retrieving conversation history.
        """
        self.context_manager = context_manager
        logger.info("ContextManager linked to LLMManager.")

    def _get_system_prompt(self) -> Dict[str, str]:
        """
        Returns a system prompt for every conversation (role: system).
        """
        return {"role": "system", "content": "You are a helpful AI assistant."}

    async def get_response_stream(self, user_input: str):
        """
        Sends a streaming chat-completion request to the LLM,
        including system prompt + history + new user message,
        and streams back deltas to the GUI.
        """
        if self.is_processing:
            logger.warning("LLM busy.")
            self.gui_queue.put({"type": "log", "payload": "LLM busy, please wait.", "severity": "warning"})
            return

        if not self.client:
            logger.error("LLM client not ready.")
            self._signal_error("LLM Client Not Ready")
            return

        if not self.context_manager:
            raise LLMManagerError("ContextManager not set in LLMManager.")

        self.is_processing = True
        self.gui_queue.put({"type": "status", "payload": "Thinking..."})

        full_response_text = ""
        error_occurred = False
        error_message = ""

        try:
            # Build messages list
            messages: List[Dict[str, Any]] = []
            sys_msg = self._get_system_prompt()
            messages.append(sys_msg)

            history = self.context_manager.get_messages_for_llm()
            messages.extend(history)

            messages.append({"role": "user", "content": user_input})
            logger.debug(f"Sending {len(messages)} messages to LLM")

            # Streaming API call
            stream = await self.client.chat.completions.create(
                model=self.model_name,
                messages=messages,
                temperature=self.default_temperature,
                max_tokens=self.default_max_tokens,
                stream=True
            )

            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    full_response_text += delta
                    self.gui_queue.put({
                        "type": "llm_chunk",
                        "payload": {"delta": delta}
                    })

            if not full_response_text.strip():
                logger.warning("Empty response from LLM.")
                self.gui_queue.put({"type": "log", "payload": "LLM returned empty response.", "severity": "warning"})

        except APIConnectionError as e:
            error_occurred = True
            error_message = f"Connection error: {e}"
            logger.error(error_message)
        except APITimeoutError as e:
            error_occurred = True
            error_message = f"Timeout: {e}"
            logger.error(error_message)
        except RateLimitError as e:
            error_occurred = True
            error_message = f"Rate limit: {e.message}"
            logger.error(error_message)
        except InternalServerError as e:
            error_occurred = True
            error_message = f"Server error ({e.status_code}): {e.message}"
            logger.error(error_message, exc_info=True)
        except APIError as e:
            error_occurred = True
            error_message = f"API error ({getattr(e, 'status_code', 'N/A')}): {getattr(e, 'message', str(e))}"
            logger.error(error_message, exc_info=True)
        except Exception as e:
            error_occurred = True
            error_message = f"Unexpected LLM error: {e}"
            logger.error(error_message, exc_info=True)
        finally:
            # Persist assistant reply on success
            if not error_occurred and full_response_text:
                try:
                    self.context_manager.add_message("assistant", full_response_text)
                except Exception as e:
                    logger.warning(f"Could not save assistant reply: {e}")

            # Send final result
            self.gui_queue.put({
                "type": "llm_result",
                "payload": {
                    "text": full_response_text if not error_occurred else None,
                    "error": error_occurred,
                    "error_message": error_message if error_occurred else None
                }
            })

            # If error, update status
            if error_occurred:
                self._signal_error("LLM Failed", error_message)
            else:
                self.gui_queue.put({"type": "status", "payload": "Done."})

            self.is_processing = False
            logger.debug("LLM processing finished.")

    def run_llm_in_background(self, user_input: str):
        """
        Launches the async get_response_stream in a thread so it doesn't block.
        """
        if self.is_processing:
            logger.warning("LLM already processing; request ignored.")
            return

        thread = threading.Thread(
            target=self._run_llm_thread_target,
            args=(user_input,),
            daemon=True,
            name="LLMThread"
        )
        logger.info("Starting LLM background thread.")
        thread.start()

    def _run_llm_thread_target(self, user_input: str):
        """
        Thread target that sets up its own asyncio loop and calls the async method.
        """
        try:
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
            loop.run_until_complete(self.get_response_stream(user_input))
        except Exception as e:
            logger.error(f"Error in LLM background thread: {e}", exc_info=True)
            self._signal_error("LLM Task Failed", str(e))
            self.is_processing = False

    def _signal_error(self, status_message: str, log_message: Optional[str] = None):
        """
        Helper to push an error status and optional log message into the GUI queue.
        """
        self.gui_queue.put({"type": "status", "payload": f"ERROR: {status_message}"})
        if log_message:
            self.gui_queue.put({"type": "log", "payload": log_message, "severity": "error"})


================================================
FILE: modules/stt_manager.py
================================================
# modules/stt_manager.py

import logging
from pathlib import Path
import time
from typing import Optional, Dict, Any, Tuple

try:
    from faster_whisper import WhisperModel
    FASTER_WHISPER_AVAILABLE = True
except ImportError:
    FASTER_WHISPER_AVAILABLE = False
    # Define dummy class if library not installed
    class WhisperModel:
        def __init__(self, *args, **kwargs): pass
        def transcribe(self, *args, **kwargs): return ([], None)

from .config import Config

logger = logging.getLogger(__name__)

# Custom Exception
class STTManagerError(Exception):
    """Custom exception for STTManager errors."""
    pass

class STTManager:
    """
    Handles Speech-to-Text transcription using the Faster Whisper library.
    """

    def __init__(self, config: Config):
        """
        Initializes the STTManager and loads the Faster Whisper model.

        Args:
            config: The application configuration object.

        Raises:
            STTManagerError: If Faster Whisper is not available or model loading fails.
        """
        logger.info("Initializing STTManager...")
        if not FASTER_WHISPER_AVAILABLE:
            raise STTManagerError("faster-whisper library is not installed. Please run: uv add faster-whisper")

        self.config = config
        self.stt_config = config.get_stt_config()

        # Model configuration
        self.model_size: str = self.stt_config.get("model_size", "base.en")
        self.device: str = self.stt_config.get("device", "cpu")
        self.compute_type: str = self.stt_config.get("compute_type", "int8") # Default to int8 for CPU

        # Adjust compute type if CUDA is selected
        if self.device == "cuda":
            # Default to float16 on CUDA if not specified
            self.compute_type = self.stt_config.get("compute_type", "float16")

        # VAD configuration
        self.vad_filter: bool = self.stt_config.get("vad_filter", True)
        # Ensure vad_parameters is a dict, default to empty if not specified or invalid type
        _vad_params = self.stt_config.get("vad_parameters", {})
        self.vad_parameters: Dict[str, Any] = _vad_params if isinstance(_vad_params, dict) else {}

        # Transcription options
        self.beam_size = 5 # Common default, make configurable later if needed

        self.model: Optional[WhisperModel] = None
        self.is_processing: bool = False

        self._load_model()
        logger.info("STTManager initialized successfully.")

    def _load_model(self):
        """Loads the Faster Whisper model."""
        logger.info(f"Loading Faster Whisper model: {self.model_size} (Device: {self.device}, Compute: {self.compute_type})")
        try:
            start_time = time.time()
            self.model = WhisperModel(
                self.model_size,
                device=self.device,
                compute_type=self.compute_type
            )
            load_time = time.time() - start_time
            logger.info(f"Whisper model loaded successfully in {load_time:.2f} seconds.")
        except Exception as e:
            logger.critical(f"Failed to load Whisper model '{self.model_size}': {e}", exc_info=True)
            # Provide specific advice for common errors
            if "cuda" in str(e).lower() or "cublas" in str(e).lower() or "cudnn" in str(e).lower():
                 error_msg = f"Failed to load STT model on GPU. Ensure CUDA toolkit, cuDNN are installed correctly and compatible with PyTorch ({self.device}/{self.compute_type}). Error: {e}"
            elif "not found" in str(e).lower() and self.model_size in str(e).lower():
                error_msg = f"Failed to load STT model: Model size '{self.model_size}' not found. Check spelling or download the model. Error: {e}"
            else:
                error_msg = f"Failed to load STT model '{self.model_size}'. Error: {e}"
            raise STTManagerError(error_msg) from e

    def transcribe(self, audio_path: str) -> Optional[str]:
        """
        Transcribes the given audio file path to text.

        Args:
            audio_path: Path to the audio file (e.g., WAV).

        Returns:
            The transcribed text as a string, or None if transcription fails.
        """
        if self.is_processing:
            logger.warning("Transcription requested but already processing another file.")
            return None
        if not self.model:
            logger.error("Cannot transcribe: Whisper model not loaded.")
            return None

        audio_file = Path(audio_path)
        if not audio_file.exists() or not audio_file.is_file():
            logger.error(f"Cannot transcribe: Audio file not found at {audio_path}")
            return None

        self.is_processing = True
        logger.info(f"Starting transcription for: {audio_path}")
        start_time = time.time()
        transcribed_text: Optional[str] = None

        try:
            segments, info = self.model.transcribe(
                str(audio_file), # Pass path as string
                beam_size=self.beam_size,
                vad_filter=self.vad_filter,
                vad_parameters=self.vad_parameters,
                # language="en" # Optional: force language if needed
            )

            # The transcription happens here as we iterate
            segment_texts = [segment.text for segment in segments]
            full_text = " ".join(segment_texts).strip()

            end_time = time.time()
            duration = end_time - start_time
            logger.info(f"Transcription finished in {duration:.2f}s.")

            if info:
                 logger.debug(f"Detected language: {info.language} (Prob: {info.language_probability:.2f})")
                 # logger.debug(f"VAD info (if enabled): Duration=~{info.duration_after_vad:.2f}s")

            if not full_text:
                logger.warning("Transcription resulted in empty text.")
                transcribed_text = "" # Return empty string instead of None for clarity
            else:
                logger.info(f"Transcription result: '{full_text[:100]}...'")
                transcribed_text = full_text

        except Exception as e:
            logger.error(f"Error during transcription for {audio_path}: {e}", exc_info=True)
            transcribed_text = None # Indicate failure
        finally:
            self.is_processing = False
            logger.debug("STT processing flag set to False.")

        return transcribed_text

================================================
FILE: modules/system_manager.py
================================================
# modules/system_manager.py
import sys
import os
import platform
import subprocess
import socket
import logging
import logging.handlers # Import handlers explicitly
import importlib.util
from typing import Optional, Tuple, List, Dict, Any
from pathlib import Path
import shutil # For checking command existence and moving files
from datetime import datetime # For timestamping logs
from urllib.parse import urlparse # For LLM check

# Assume Config is in the same directory or accessible via PYTHONPATH
from .config import Config, ConfigError

# Setup logger for this module
logger = logging.getLogger(__name__)

# --- Safe torch import for checks ---
try:
    import torch
    TORCH_AVAILABLE = True
    logger.debug("PyTorch imported successfully for system checks.")
except ImportError:
    TORCH_AVAILABLE = False
    logger.debug("PyTorch not found during initial import for system checks.")
    # Define dummy types if torch is not installed for checks to work
    class _cuda:
        @staticmethod
        def is_available() -> bool: return False
        @staticmethod
        def device_count() -> int: return 0
        @staticmethod
        def get_device_name(idx: int) -> str: return "N/A"
    class _version: cuda = None
    torch = type('torch', (), {'cuda': _cuda, 'version': _version})()
# --- End safe torch import ---

class LoggingSetupError(Exception):
    """Custom exception for failures during logging setup."""
    pass

class RequirementError(Exception):
    """Custom exception for critical requirement failures."""
    pass

class SystemManager:
    """
    Handles system-level checks, requirements verification, logging setup (including rotation),
    and logging system information for the MiraiAssist application.
    """

    MIN_PYTHON_VERSION: Tuple[int, int] = (3, 9) # Example: Require Python 3.9+

    def __init__(self, config: Config):
        """
        Initializes the SystemManager. Requires a loaded Config instance.

        Args:
            config (Config): The application configuration object (must be loaded).

        Raises:
            ValueError: If the provided config object is not loaded.
        """
        if not config.is_loaded:
            # Cannot proceed without loaded config
            raise ValueError("Config object must be loaded before initializing SystemManager.")
        self.config = config
        self._system_info_logged = False
        self.console_handler: Optional[logging.StreamHandler] = None # Store console handler if needed

    # --- Logging Setup (with Rotation) ---
    def _rotate_existing_log(self, log_file_path: Path) -> None:
        """Renames an existing log file with a timestamp if it exists."""
        if log_file_path.exists() and log_file_path.stat().st_size > 0:
            try:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                # Construct backup name correctly using stem and suffix
                backup_filename = f"{log_file_path.stem}_{timestamp}{log_file_path.suffix}"
                backup_path = log_file_path.with_name(backup_filename)
                # Ensure backup name doesn't somehow conflict (highly unlikely with timestamp)
                if backup_path.exists():
                    logger.warning(f"Backup log file {backup_path} already exists. Skipping rotation for {log_file_path.name}.")
                    return
                shutil.move(str(log_file_path), str(backup_path)) # Use shutil.move
                logger.debug(f"Rotated existing log file '{log_file_path.name}' to '{backup_path.name}'")
            except Exception as e:
                logger.warning(f"Could not rotate log file {log_file_path}: {e}", exc_info=False)

    def setup_logging(self) -> None:
        """
        Configures application logging based on config.yaml, including startup rotation.

        Sets up console (TUI), application file, and error file handlers. Old log files
        are timestamped on startup. RotatingFileHandler manages size during runtime.

        Raises:
            LoggingSetupError: If essential logging setup fails.
        """
        log_config = self.config.get_logging_config()
        if not log_config:
             logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
             logger.warning("Logging configuration section missing. Using basic console logging.")
             return

        # --- Get Config Values ---
        file_logging_enabled = log_config.get('file_logging_enabled', True)
        log_directory_str = log_config.get('log_directory', 'logs')
        app_log_file_str = log_config.get('application_log_file', 'mirai_assist.log')
        err_log_file_str = log_config.get('error_log_file', 'errors.log')
        app_log_level_str = log_config.get('application_log_level', 'DEBUG').upper()
        err_log_level_str = log_config.get('error_log_level', 'WARNING').upper()
        tui_level_str = log_config.get('tui_level', 'INFO').upper()
        log_format_str = log_config.get('format', '%(asctime)s - %(name)s [%(levelname)s] - %(message)s')
        log_date_format = log_config.get('date_format', '%Y-%m-%d %H:%M:%S')

        try:
            log_directory = Path(log_directory_str).resolve() # Use absolute paths
            app_log_file = log_directory / app_log_file_str
            err_log_file = log_directory / err_log_file_str
        except Exception as e:
             print(f"FATAL ERROR: Invalid logging path configuration: {e}", file=sys.stderr)
             raise LoggingSetupError(f"Invalid logging path configuration: {e}") from e

        # --- Basic Config ---
        logging_level = logging.DEBUG # Root logger level
        formatter = logging.Formatter(log_format_str, datefmt=log_date_format)
        root_logger = logging.getLogger()
        root_logger.setLevel(logging_level)

        # --- Clear Existing Handlers ---
        for handler in root_logger.handlers[:]:
            root_logger.removeHandler(handler)
            handler.close()

        # --- Console/TUI Handler (Setup first) ---
        try:
            tui_level = getattr(logging, tui_level_str, logging.INFO)
            self.console_handler = logging.StreamHandler(sys.stdout)
            self.console_handler.setLevel(tui_level)
            self.console_handler.setFormatter(formatter)
            root_logger.addHandler(self.console_handler)
            print(f"INFO: Console/TUI logging handler configured at level {tui_level_str}.") # Initial print
            root_logger.info("Console/TUI logging handler configured.") # Log via handler
        except Exception as e:
            print(f"FATAL ERROR: Failed to set up console logging: {e}", file=sys.stderr)
            raise LoggingSetupError(f"Failed to set up console logging: {e}") from e

        # --- File Handlers (if enabled) ---
        if file_logging_enabled:
            root_logger.info(f"Attempting file logging setup in directory: {log_directory}")
            try:
                log_directory.mkdir(parents=True, exist_ok=True)
                root_logger.debug(f"Log directory ensured: {log_directory}")

                # <<<--- ROTATION (on startup) --- >>>
                self._rotate_existing_log(app_log_file)
                self._rotate_existing_log(err_log_file)
                # <<<----------------------------->>>

                # Application Log File Handler (Rotating based on size during runtime)
                app_log_level = getattr(logging, app_log_level_str, logging.DEBUG)
                root_logger.debug(f"Setting up application log: {app_log_file} (Level: {app_log_level_str})")
                # Keep backupCount low (e.g., 1) if startup rotation handles history
                app_handler = logging.handlers.RotatingFileHandler(
                    app_log_file, maxBytes=5*1024*1024, backupCount=1, encoding='utf-8'
                )
                app_handler.setLevel(app_log_level)
                app_handler.setFormatter(formatter)
                root_logger.addHandler(app_handler)
                root_logger.debug("Application file handler added.")

                # Error Log File Handler (Rotating based on size during runtime)
                err_log_level = getattr(logging, err_log_level_str, logging.WARNING)
                root_logger.debug(f"Setting up error log: {err_log_file} (Level: {err_log_level_str})")
                err_handler = logging.handlers.RotatingFileHandler(
                    err_log_file, maxBytes=5*1024*1024, backupCount=1, encoding='utf-8'
                )
                err_handler.setLevel(err_log_level)
                err_handler.setFormatter(formatter)
                root_logger.addHandler(err_handler)
                root_logger.debug("Error file handler added.")

                root_logger.info(f"File logging setup successful (with rotation).")

            except Exception as e:
                root_logger.critical(f"Failed to set up file logging handlers in {log_directory}: {e}", exc_info=True)
                raise LoggingSetupError(f"Failed to create file logging handlers: {e}") from e
        else:
            root_logger.info("File logging is disabled by configuration.")

        root_logger.info(f"Logging setup complete.")

    # --- System Information ---
    def log_system_info(self) -> None:
        """Logs detailed information about the system environment."""
        if self._system_info_logged:
            return
        logger.info("--- System Information ---")
        try:
            logger.info(f"OS:           {platform.system()} {platform.release()} ({platform.machine()})")
            logger.info(f"Python Ver:   {sys.version.split()[0]}")
            logger.info(f"Python Path:  {sys.executable}")
            logger.info(f"Project Root: {Path.cwd().resolve()}")

            # Log UV info
            if self._is_command_available("uv"):
                try:
                    result = subprocess.run(["uv", "--version"], capture_output=True, text=True, check=True, timeout=5)
                    logger.info(f"UV Version:   {result.stdout.strip()}")
                except Exception as e:
                    logger.warning(f"Could not get UV version: {e}", exc_info=False)
            else:
                logger.warning("UV command not found in PATH.")

            # Log PyTorch & CUDA
            logger.info(f"PyTorch Inst: {'Yes' if TORCH_AVAILABLE else 'No'}")
            if TORCH_AVAILABLE:
                logger.info(f"PyTorch Ver:  {torch.__version__}")
                cuda_available = torch.cuda.is_available()
                logger.info(f"CUDA Avail:   {'Yes' if cuda_available else 'No'}")
                if cuda_available:
                    torch_cuda_ver = getattr(torch.version, 'cuda', 'N/A')
                    logger.info(f"CUDA Ver:     {torch_cuda_ver}")
                    try:
                        num_gpus = torch.cuda.device_count()
                        logger.info(f"GPU Count:    {num_gpus}")
                        for i in range(num_gpus):
                             logger.info(f"  GPU {i}:      {torch.cuda.get_device_name(i)}")
                    except Exception as e:
                        logger.warning(f"Could not get detailed GPU info: {e}", exc_info=False)

            # Log espeak-ng availability (Windows only)
            if platform.system() == "Windows":
                if self._check_espeak(log_details=False):
                    logger.info("espeak-ng:    Detected")
                else:
                    logger.warning("espeak-ng:    Not detected or not working (Required for Kokoro TTS)")

        except Exception as e:
            logger.error(f"Error gathering system information: {e}", exc_info=True)
        finally:
            logger.info("--------------------------")
            self._system_info_logged = True


    # --- Requirement Checks (Includes Module Checks) ---
    def _check_module_import(self, module_name: str, package_name: Optional[str] = None, purpose: str = "") -> bool:
        """Tries to import a module, logs outcome, returns True if successful."""
        check_package = package_name or module_name
        purpose_str = f" ({purpose})" if purpose else ""
        logger.debug(f"Checking import for module: {module_name}{purpose_str}...")
        try:
            importlib.import_module(module_name)
            logger.debug(f" > Import OK: {module_name}")
            return True
        except ImportError:
            logger.error(f" > Import FAILED: Module '{module_name}' not found{purpose_str}. Install '{check_package}' (`uv add {check_package}`).")
            return False
        except Exception as e:
            logger.error(f" > Import ERROR: Unexpected error importing module '{module_name}'{purpose_str}: {e}", exc_info=True)
            return False

    def check_requirements(self) -> bool:
        """
        Performs essential system and dependency checks before full app launch.

        Raises:
            RequirementError: If a critical requirement is not met. Contains a user-friendly message.

        Returns:
            bool: True if all critical checks pass.
        """
        logger.info("Performing system and dependency requirement checks...")
        critical_failures: List[str] = []
        warnings: List[str] = [] # Keep warnings for non-critical issues

        # --- Section 1: Basic System Checks ---
        logger.debug("--- Checking Basic System ---")
        if not self._check_python_version():
            critical_failures.append(f"Python version {sys.version_info.major}.{sys.version_info.minor} below requirement >= {'.'.join(map(str, self.MIN_PYTHON_VERSION))}.")
        if not self.config.is_loaded:
             critical_failures.append("Configuration could not be loaded.")
        if platform.system() == "Windows" and not self._check_espeak():
            critical_failures.append("espeak-ng not found/working (required for TTS). Install from official releases.")

        # --- Section 2: Core AI/ML Libraries ---
        logger.debug("--- Checking Core AI Libraries ---")
        if not self._check_module_import("torch", purpose="Core ML"):
             critical_failures.append("PyTorch is not installed.")
        # Only check CUDA if torch installed and config requires it
        elif TORCH_AVAILABLE: # Check TORCH_AVAILABLE guard again
            stt_config = self.config.get_stt_config()
            needs_cuda = stt_config.get('device', 'cpu').lower() == 'cuda'
            if needs_cuda:
                 logger.info("Checking CUDA availability (required by STT config)...")
                 if not self._check_cuda():
                     critical_failures.append("STT requires CUDA, but PyTorch cannot detect GPU.")
                 else:
                     logger.info("CUDA check passed (required by STT config).")
            else:
                 logger.info("CUDA acceleration not configured for STT.")

        # --- Section 3: Specific Functionality Modules ---
        logger.debug("--- Checking Functionality Modules ---")
        # LLM
        if not self._check_module_import("openai", purpose="LLM Client"):
             critical_failures.append("openai library not installed.")
        if not self._check_module_import("tiktoken", purpose="LLM Tokenizer"):
             # Maybe make this a warning if fallback token counting is planned? For now, critical.
             critical_failures.append("tiktoken library not installed.")
        if not self._check_llm_endpoint(): # Check endpoint connectivity
            llm_url = self.config.get_llm_config().get('api_base_url', 'N/A')
            critical_failures.append(f"LLM API endpoint unreachable ({llm_url}).")

        # # Hotword
        # if not self._check_module_import("pvporcupine", purpose="Hotword"):
        #      critical_failures.append("pvporcupine library not installed.")
        # # Check for API key environment variable
        # pv_key_var = self.config.get_activation_config().get("porcupine_access_key_env_var")
        # if not pv_key_var or not os.environ.get(pv_key_var):
        #      # Make this critical if hotword is enabled, otherwise warning
        #      if self.config.get_activation_config().get("hotword_enabled"):
        #          critical_failures.append(f"Hotword enabled but access key env var ('{pv_key_var}') not set.")
        #      else:
        #           warnings.append(f"Porcupine key env var ('{pv_key_var}') not set. Hotword detection cannot be enabled.")

        # STT
        if not self._check_module_import("faster_whisper", purpose="STT"):
            critical_failures.append("faster-whisper library not installed.")

        # TTS
        if not self._check_module_import("kokoro", purpose="TTS"):
            critical_failures.append("kokoro library not installed.")
        if not self._check_module_import("soundfile", purpose="Audio File I/O"):
            critical_failures.append("soundfile library not installed.")

        # Audio I/O
        if not self._check_module_import("pyaudio", purpose="Audio Streams"):
            critical_failures.append("PyAudio library not installed. Check OS-specific install notes if needed.")
        if not self._check_module_import("numpy", purpose="Numerical Processing"):
            critical_failures.append("numpy library not installed.")


        # --- Report Results ---
        for warn in warnings:
            logger.warning(f"Requirement Warning: {warn}")

        if critical_failures:
            logger.critical("--- Critical System Requirements Not Met ---")
            for i, fail in enumerate(critical_failures):
                logger.critical(f"  FAIL {i+1}: {fail}")
            logger.critical("--------------------------------------------")
            # Combine failures into a summary message for the GUI
            summary = f"Failed {len(critical_failures)} requirement(s). Check logs."
            if len(critical_failures) == 1:
                 summary = f"Requirement Failed: {critical_failures[0]}"
            elif len(critical_failures) > 1:
                 summary = f"Requirements Failed: {critical_failures[0]} (and others)"

            raise RequirementError(summary) # Use specific exception
        else:
            logger.info("All critical system and dependency checks passed.")

        return True

    # --- Helper Check Methods ---

    def _check_python_version(self) -> bool:
        """Checks if the current Python version meets the minimum requirement."""
        return sys.version_info >= self.MIN_PYTHON_VERSION

    def _is_command_available(self, command: str) -> bool:
        """Checks if a command is available in the system's PATH."""
        is_available = shutil.which(command) is not None
        logger.debug(f"Checking command '{command}': {'Found' if is_available else 'Not Found'}")
        return is_available

    def _check_cuda(self) -> bool:
        """Checks if CUDA is available through PyTorch."""
        if not TORCH_AVAILABLE: return False
        is_available = torch.cuda.is_available()
        logger.debug(f"torch.cuda.is_available(): {is_available}")
        return is_available

    def _check_espeak(self, log_details=True) -> bool:
        """Checks if espeak-ng is installed and runnable."""
        command = "espeak-ng"
        if not self._is_command_available(command):
            if log_details: logger.warning(f"'{command}' command not found in PATH. Install from https://github.com/espeak-ng/espeak-ng/releases")
            return False
        try:
            # Use --version which is standard and exits quickly
            result = subprocess.run([command, "--version"], capture_output=True, text=True, check=True, timeout=5)
            if log_details:
                # Extract only the version number line if possible
                version_line = result.stdout.splitlines()[0] if result.stdout else "N/A"
                logger.debug(f"Found espeak-ng version: {version_line}")
            return True
        except subprocess.TimeoutExpired:
             if log_details: logger.error(f"Running '{command} --version' timed out.")
             return False
        except subprocess.CalledProcessError as e:
             if log_details: logger.error(f"'{command} --version' returned non-zero exit status {e.returncode}.")
             return False
        except FileNotFoundError:
             if log_details: logger.error(f"'{command}' command FileNotFoundError (redundant check).")
             return False
        except Exception as e:
            if log_details: logger.error(f"Error running '{command} --version': {e}", exc_info=False)
            return False

    def _check_llm_endpoint(self) -> bool:
        """Checks if the configured LLM API endpoint host/port is reachable via socket."""
        llm_config = self.config.get_llm_config()
        base_url = llm_config.get('api_base_url')
        if not base_url:
            logger.error("LLM check failed: 'api_base_url' is not defined.")
            return False

        hostname = None # Initialize hostname for logging in case of parsing error
        port = None
        try:
            parsed_url = urlparse(base_url)
            hostname = parsed_url.hostname
            port = parsed_url.port

            if not hostname:
                raise ValueError("Could not extract hostname from URL.")

            # Determine default port if not specified
            if port is None:
                port = 443 if parsed_url.scheme == 'https' else 80
                logger.debug(f"No port specified in URL, assuming default {port} for scheme '{parsed_url.scheme}'.")

            timeout_seconds = 3
            logger.debug(f"Attempting socket connection to LLM host {hostname}:{port} (from {base_url})...")

            with socket.create_connection((hostname, port), timeout=timeout_seconds):
                logger.debug(f"Socket connection successful to {hostname}:{port}.")
                return True # Basic connectivity confirmed

        except ImportError: # Should not happen as urlparse is standard lib
            logger.error("Failed to import urllib.parse for LLM endpoint check.")
            return False
        except ValueError as e:
             logger.error(f"Invalid LLM 'api_base_url' format: '{base_url}'. Error: {e}")
             return False
        except socket.gaierror as e:
            logger.error(f"Hostname resolution failed for LLM endpoint '{hostname or base_url}': {e}")
            return False
        except socket.timeout:
            logger.error(f"Connection timed out reaching LLM endpoint {hostname}:{port}.")
            return False
        except ConnectionRefusedError:
            logger.error(f"Connection refused by LLM server at {hostname}:{port}. Is it running?")
            return False
        except OSError as e:
            logger.error(f"Network OS error reaching LLM endpoint {hostname}:{port}: {e}")
            return False
        except Exception as e:
            logger.error(f"Unexpected error checking LLM endpoint {base_url}: {e}", exc_info=True)
            return False

# End of modules/system_manager.py

================================================
FILE: modules/tts_manager.py
================================================
# modules/tts_manager.py

import threading
import tempfile
import soundfile as sf
from pathlib import Path
from typing import Any, Dict, Optional

from kokoro import KPipeline

from .config import Config
import logging

logger = logging.getLogger(__name__)

class TTSManagerError(Exception):
    pass

class TTSManager:
    """
    Uses kokoro.KPipeline to turn assistant text into audio,
    writes to a temp WAV file, then hands off to AudioManager
    for playback.
    """

    def __init__(self, config: Config, gui_queue: Any, audio_manager: Any):
        tts_cfg = config.get_section("tts", {})
        self.lang_code: str = tts_cfg.get("lang_code", "a")
        self.voice:     str = tts_cfg.get("voice", "af_heart")
        self.speed:     float= float(tts_cfg.get("speed", 1.0))
        self.audio_base: str = tts_cfg.get("output_filename_base", "temp_output")

        self.pipeline = KPipeline(lang_code=self.lang_code)
        self.gui_queue = gui_queue
        self.audio_manager = audio_manager

    def speak_text(self, text: str):
        """
        Runs the TTS pipeline in a background thread,
        saves each chunk to a WAV and then plays it.
        """
        threading.Thread(
            target=self._run_and_play,
            args=(text,),
            daemon=True,
            name="TTSThread"
        ).start()

    def _run_and_play(self, text: str):
        try:
            # kokoro generator yields (graphemes, phonemes, audio ndarray)
            for idx, (gs, ps, audio) in enumerate(
                self.pipeline(text, voice=self.voice, speed=self.speed)
            ):
                # write to a temp file
                fd, wav_path = tempfile.mkstemp(prefix=f"{self.audio_base}_{idx}_", suffix=".wav")
                Path(wav_path).unlink()  # we just need the path
                sf.write(wav_path, audio, samplerate=24000)
                logger.debug(f"TTS wrote chunk #{idx} to {wav_path}")

                # hand off to AudioManager for playback (non‐blocking)
                self.audio_manager.play_audio_file(wav_path)
        except Exception as e:
            logger.error(f"TTS failure: {e}", exc_info=True)
            self.gui_queue.put({"type":"log", "payload":f"TTS Error: {e}", "severity":"error"})


================================================
FILE: pyproject.toml
================================================
[project]
name = "mirai_assist"
version = "0.1.0"
requires-python = ">=3.12.10"
description = "A modular virtual assistant using local AI."

dependencies = [
    # core
    "pyyaml",
    "spacy",
    "openai",
    "pvporcupine",
    "faster-whisper",
    "kokoro",
    "pyaudio",
    "soundfile",
    "numpy",
    "scipy",
    "pillow",
    "torch==2.7.0+cu128",
    "torchvision==0.22.0+cu128",
    "torchaudio==2.7.0+cu128",
    "customtkinter",
    "tiktoken",
]

# ─────────  uv index configuration  ─────────
# 1️⃣  keep PyPI as the **first/default** index
[[tool.uv.index]]
name = "pypi"
url  = "https://pypi.org/simple"

# 2️⃣  add the PyTorch CUDA-12.8 mirror
[[tool.uv.index]]
name = "pytorch-cu128"
url  = "https://download.pytorch.org/whl/cu128"

# 3️⃣  tell uv to fetch only these three from the mirror
[tool.uv.sources]
torch       = { index = "pytorch-cu128" }
torchvision = { index = "pytorch-cu128" }
torchaudio  = { index = "pytorch-cu128" }

-----------------------------------------

Here is all the documentation for my project.

# faster-whisper will server as our STT - It includes built-in VAD.

## Requirements

* Python 3.9 or greater

Unlike openai-whisper, FFmpeg does **not** need to be installed on the system. The audio is decoded with the Python library [PyAV](https://github.com/PyAV-Org/PyAV) which bundles the FFmpeg libraries in its package.

### GPU

GPU execution requires the following NVIDIA libraries to be installed:

* [cuBLAS for CUDA 12](https://developer.nvidia.com/cublas)
* [cuDNN 9 for CUDA 12](https://developer.nvidia.com/cudnn)

**Note**: The latest versions of `ctranslate2` only support CUDA 12 and cuDNN 9. For CUDA 11 and cuDNN 8, the current workaround is downgrading to the `3.24.0` version of `ctranslate2`, for CUDA 12 and cuDNN 8, downgrade to the `4.4.0` version of `ctranslate2`, (This can be done with `pip install --force-reinstall ctranslate2==4.4.0` or specifying the version in a `requirements.txt`).

There are multiple ways to install the NVIDIA libraries mentioned above. The recommended way is described in the official NVIDIA documentation, but we also suggest other installation methods below. 

<details>
<summary>Other installation methods (click to expand)</summary>


**Note:** For all these methods below, keep in mind the above note regarding CUDA versions. Depending on your setup, you may need to install the _CUDA 11_ versions of libraries that correspond to the CUDA 12 libraries listed in the instructions below.

#### Download the libraries from Purfview's repository (Windows & Linux)

Purfview's [whisper-standalone-win](https://github.com/Purfview/whisper-standalone-win) provides the required NVIDIA libraries for Windows & Linux in a [single archive](https://github.com/Purfview/whisper-standalone-win/releases/tag/libs). Decompress the archive and place the libraries in a directory included in the `PATH`.

</details>

## Installation

The module can be installed from [PyPI](https://pypi.org/project/faster-whisper/):

```bash
pip install faster-whisper
```

<details>
<summary>Other installation methods (click to expand)</summary>

### Install the master branch

```bash
pip install --force-reinstall "faster-whisper @ https://github.com/SYSTRAN/faster-whisper/archive/refs/heads/master.tar.gz"
```

### Install a specific commit

```bash
pip install --force-reinstall "faster-whisper @ https://github.com/SYSTRAN/faster-whisper/archive/a4f1cc8f11433e454c3934442b5e1a4ed5e865c3.tar.gz"
```

</details>

## Usage

### Faster-whisper

```python
from faster_whisper import WhisperModel

model_size = "large-v3"

# Run on GPU with FP16
model = WhisperModel(model_size, device="cuda", compute_type="float16")

# or run on GPU with INT8
# model = WhisperModel(model_size, device="cuda", compute_type="int8_float16")
# or run on CPU with INT8
# model = WhisperModel(model_size, device="cpu", compute_type="int8")

segments, info = model.transcribe("audio.mp3", beam_size=5)

print("Detected language '%s' with probability %f" % (info.language, info.language_probability))

for segment in segments:
    print("[%.2fs -> %.2fs] %s" % (segment.start, segment.end, segment.text))
```

**Warning:** `segments` is a *generator* so the transcription only starts when you iterate over it. The transcription can be run to completion by gathering the segments in a list or a `for` loop:

```python
segments, _ = model.transcribe("audio.mp3")
segments = list(segments)  # The transcription will actually run here.
```

### Batched Transcription
The following code snippet illustrates how to run batched transcription on an example audio file. `BatchedInferencePipeline.transcribe` is a drop-in replacement for `WhisperModel.transcribe`

```python
from faster_whisper import WhisperModel, BatchedInferencePipeline

model = WhisperModel("turbo", device="cuda", compute_type="float16")
batched_model = BatchedInferencePipeline(model=model)
segments, info = batched_model.transcribe("audio.mp3", batch_size=16)

for segment in segments:
    print("[%.2fs -> %.2fs] %s" % (segment.start, segment.end, segment.text))
```

### Faster Distil-Whisper

The Distil-Whisper checkpoints are compatible with the Faster-Whisper package. In particular, the latest [distil-large-v3](https://huggingface.co/distil-whisper/distil-large-v3)
checkpoint is intrinsically designed to work with the Faster-Whisper transcription algorithm. The following code snippet 
demonstrates how to run inference with distil-large-v3 on a specified audio file:

```python
from faster_whisper import WhisperModel

model_size = "distil-large-v3"

model = WhisperModel(model_size, device="cuda", compute_type="float16")
segments, info = model.transcribe("audio.mp3", beam_size=5, language="en", condition_on_previous_text=False)

for segment in segments:
    print("[%.2fs -> %.2fs] %s" % (segment.start, segment.end, segment.text))
```

For more information about the distil-large-v3 model, refer to the original [model card](https://huggingface.co/distil-whisper/distil-large-v3).

### Word-level timestamps

```python
segments, _ = model.transcribe("audio.mp3", word_timestamps=True)

for segment in segments:
    for word in segment.words:
        print("[%.2fs -> %.2fs] %s" % (word.start, word.end, word.word))
```

### VAD filter

The library integrates the [Silero VAD](https://github.com/snakers4/silero-vad) model to filter out parts of the audio without speech:

```python
segments, _ = model.transcribe("audio.mp3", vad_filter=True)
```

The default behavior is conservative and only removes silence longer than 2 seconds. See the available VAD parameters and default values in the [source code](https://github.com/SYSTRAN/faster-whisper/blob/master/faster_whisper/vad.py). They can be customized with the dictionary argument `vad_parameters`:

```python
segments, _ = model.transcribe(
    "audio.mp3",
    vad_filter=True,
    vad_parameters=dict(min_silence_duration_ms=500),
)
```
Vad filter is enabled by default for batched transcription.

### Logging

The library logging level can be configured like this:

```python
import logging

logging.basicConfig()
logging.getLogger("faster_whisper").setLevel(logging.DEBUG)
```

## Model conversion

When loading a model from its size such as `WhisperModel("large-v3")`, the corresponding CTranslate2 model is automatically downloaded from the [Hugging Face Hub](https://huggingface.co/Systran).

We also provide a script to convert any Whisper models compatible with the Transformers library. They could be the original OpenAI models or user fine-tuned models.

For example the command below converts the [original "large-v3" Whisper model](https://huggingface.co/openai/whisper-large-v3) and saves the weights in FP16:

```bash
pip install transformers[torch]>=4.23

ct2-transformers-converter --model openai/whisper-large-v3 --output_dir whisper-large-v3-ct2
--copy_files tokenizer.json preprocessor_config.json --quantization float16
```

* The option `--model` accepts a model name on the Hub or a path to a model directory.
* If the option `--copy_files tokenizer.json` is not used, the tokenizer configuration is automatically downloaded when the model is loaded later.

Models can also be converted from the code. See the [conversion API](https://opennmt.net/CTranslate2/python/ctranslate2.converters.TransformersConverter.html).

### Load a converted model

1. Directly load the model from a local directory:
```python
model = faster_whisper.WhisperModel("whisper-large-v3-ct2")
```

2. [Upload your model to the Hugging Face Hub](https://huggingface.co/docs/transformers/model_sharing#upload-with-the-web-interface) and load it from its name:
```python
model = faster_whisper.WhisperModel("username/whisper-large-v3-ct2")

--------------------------------------------------------

# kokoro-tts will be our TTS backend.

# kokoro

An inference library for [Kokoro-82M](https://huggingface.co/hexgrad/Kokoro-82M). You can [`pip install kokoro`](https://pypi.org/project/kokoro/).

> **Kokoro** is an open-weight TTS model with 82 million parameters. Despite its lightweight architecture, it delivers comparable quality to larger models while being significantly faster and more cost-efficient. With Apache-licensed weights, Kokoro can be deployed anywhere from production environments to personal projects.

### Usage
You can run this basic cell on [Google Colab](https://colab.research.google.com/). [Listen to samples](https://huggingface.co/hexgrad/Kokoro-82M/blob/main/SAMPLES.md).
```py
!pip install -q kokoro>=0.9.4 soundfile
!apt-get -qq -y install espeak-ng > /dev/null 2>&1
from kokoro import KPipeline
from IPython.display import display, Audio
import soundfile as sf
import torch
pipeline = KPipeline(lang_code='a')
text = '''
[Kokoro](/kˈOkəɹO/) is an open-weight TTS model with 82 million parameters. Despite its lightweight architecture, it delivers comparable quality to larger models while being significantly faster and more cost-efficient. With Apache-licensed weights, [Kokoro](/kˈOkəɹO/) can be deployed anywhere from production environments to personal projects.
'''
generator = pipeline(text, voice='af_heart')
for i, (gs, ps, audio) in enumerate(generator):
    print(i, gs, ps)
    display(Audio(data=audio, rate=24000, autoplay=i==0))
    sf.write(f'{i}.wav', audio, 24000)
```
Under the hood, `kokoro` uses [`misaki`](https://pypi.org/project/misaki/), a G2P library at https://github.com/hexgrad/misaki

### Advanced Usage
You can run this advanced cell on [Google Colab](https://colab.research.google.com/).
```py
# 1️⃣ Install kokoro
!pip install -q kokoro>=0.9.4 soundfile
# 2️⃣ Install espeak, used for English OOD fallback and some non-English languages
!apt-get -qq -y install espeak-ng > /dev/null 2>&1

# 3️⃣ Initalize a pipeline
from kokoro import KPipeline
from IPython.display import display, Audio
import soundfile as sf
import torch
# 🇺🇸 'a' => American English, 🇬🇧 'b' => British English
# 🇪🇸 'e' => Spanish es
# 🇫🇷 'f' => French fr-fr
# 🇮🇳 'h' => Hindi hi
# 🇮🇹 'i' => Italian it
# 🇯🇵 'j' => Japanese: pip install misaki[ja]
# 🇧🇷 'p' => Brazilian Portuguese pt-br
# 🇨🇳 'z' => Mandarin Chinese: pip install misaki[zh]
pipeline = KPipeline(lang_code='a') # <= make sure lang_code matches voice, reference above.

# This text is for demonstration purposes only, unseen during training
text = '''
The sky above the port was the color of television, tuned to a dead channel.
"It's not like I'm using," Case heard someone say, as he shouldered his way through the crowd around the door of the Chat. "It's like my body's developed this massive drug deficiency."
It was a Sprawl voice and a Sprawl joke. The Chatsubo was a bar for professional expatriates; you could drink there for a week and never hear two words in Japanese.

These were to have an enormous impact, not only because they were associated with Constantine, but also because, as in so many other areas, the decisions taken by Constantine (or in his name) were to have great significance for centuries to come. One of the main issues was the shape that Christian churches were to take, since there was not, apparently, a tradition of monumental church buildings when Constantine decided to help the Christian church build a series of truly spectacular structures. The main form that these churches took was that of the basilica, a multipurpose rectangular structure, based ultimately on the earlier Greek stoa, which could be found in most of the great cities of the empire. Christianity, unlike classical polytheism, needed a large interior space for the celebration of its religious services, and the basilica aptly filled that need. We naturally do not know the degree to which the emperor was involved in the design of new churches, but it is tempting to connect this with the secular basilica that Constantine completed in the Roman forum (the so-called Basilica of Maxentius) and the one he probably built in Trier, in connection with his residence in the city at a time when he was still caesar.

[Kokoro](/kˈOkəɹO/) is an open-weight TTS model with 82 million parameters. Despite its lightweight architecture, it delivers comparable quality to larger models while being significantly faster and more cost-efficient. With Apache-licensed weights, [Kokoro](/kˈOkəɹO/) can be deployed anywhere from production environments to personal projects.
'''
# text = '「もしおれがただ偶然、そしてこうしようというつもりでなくここに立っているのなら、ちょっとばかり絶望するところだな」と、そんなことが彼の頭に思い浮かんだ。'
# text = '中國人民不信邪也不怕邪，不惹事也不怕事，任何外國不要指望我們會拿自己的核心利益做交易，不要指望我們會吞下損害我國主權、安全、發展利益的苦果！'
# text = 'Los partidos políticos tradicionales compiten con los populismos y los movimientos asamblearios.'
# text = 'Le dromadaire resplendissant déambulait tranquillement dans les méandres en mastiquant de petites feuilles vernissées.'
# text = 'ट्रांसपोर्टरों की हड़ताल लगातार पांचवें दिन जारी, दिसंबर से इलेक्ट्रॉनिक टोल कलेक्शनल सिस्टम'
# text = "Allora cominciava l'insonnia, o un dormiveglia peggiore dell'insonnia, che talvolta assumeva i caratteri dell'incubo."
# text = 'Elabora relatórios de acompanhamento cronológico para as diferentes unidades do Departamento que propõem contratos.'

# 4️⃣ Generate, display, and save audio files in a loop.
generator = pipeline(
    text, voice='af_heart', # <= change voice here
    speed=1, split_pattern=r'\n+'
)
# Alternatively, load voice tensor directly:
# voice_tensor = torch.load('path/to/voice.pt', weights_only=True)
# generator = pipeline(
#     text, voice=voice_tensor,
#     speed=1, split_pattern=r'\n+'
# )

for i, (gs, ps, audio) in enumerate(generator):
    print(i)  # i => index
    print(gs) # gs => graphemes/text
    print(ps) # ps => phonemes
    display(Audio(data=audio, rate=24000, autoplay=i==0))
    sf.write(f'{i}.wav', audio, 24000) # save each audio file
```

### Windows Installation
To install espeak-ng on Windows:
1. Go to [espeak-ng releases](https://github.com/espeak-ng/espeak-ng/releases)
2. Click on **Latest release** 
3. Download the appropriate `*.msi` file (e.g. **espeak-ng-20191129-b702b03-x64.msi**)
4. Run the downloaded installer

For advanced configuration and usage on Windows, see the [official espeak-ng Windows guide](https://github.com/espeak-ng/espeak-ng/blob/master/docs/guide.md)

---------------------------------------------------------------

# uv for Python. We should use uv across our project to keep a unified, self-contained Python for the entire project.

### Projects

uv manages project dependencies and environments, with support for lockfiles, workspaces, and more,
similar to `rye` or `poetry`:

```console
$ uv init example
Initialized project `example` at `/home/user/example`

$ cd example

$ uv add ruff
Creating virtual environment at: .venv
Resolved 2 packages in 170ms
   Built example @ file:///home/user/example
Prepared 2 packages in 627ms
Installed 2 packages in 1ms
 + example==0.1.0 (from file:///home/user/example)
 + ruff==0.5.0

$ uv run ruff check
All checks passed!

$ uv lock
Resolved 2 packages in 0.33ms

$ uv sync
Resolved 2 packages in 0.70ms
Audited 1 package in 0.02ms
```

See the [project documentation](https://docs.astral.sh/uv/guides/projects/) to get started.

uv also supports building and publishing projects, even if they're not managed with uv. See the
[publish guide](https://docs.astral.sh/uv/guides/publish/) to learn more.

### Scripts

uv manages dependencies and environments for single-file scripts.

Create a new script and add inline metadata declaring its dependencies:

```console
$ echo 'import requests; print(requests.get("https://astral.sh"))' > example.py

$ uv add --script example.py requests
Updated `example.py`
```

Then, run the script in an isolated virtual environment:

```console
$ uv run example.py
Reading inline script metadata from: example.py
Installed 5 packages in 12ms
<Response [200]>
```

See the [scripts documentation](https://docs.astral.sh/uv/guides/scripts/) to get started.

### Tools

uv executes and installs command-line tools provided by Python packages, similar to `pipx`.

Run a tool in an ephemeral environment using `uvx` (an alias for `uv tool run`):

```console
$ uvx pycowsay 'hello world!'
Resolved 1 package in 167ms
Installed 1 package in 9ms

Install a tool with `uv tool install`:

```console
$ uv tool install ruff
Resolved 1 package in 6ms
Installed 1 package in 2ms
 + ruff==0.5.0
Installed 1 executable: ruff

$ ruff --version
ruff 0.5.0
```

See the [tools documentation](https://docs.astral.sh/uv/guides/tools/) to get started.

### Python versions

uv installs Python and allows quickly switching between versions.

Install multiple Python versions:

```console
$ uv python install 3.10 3.11 3.12
Searching for Python versions matching: Python 3.10
Searching for Python versions matching: Python 3.11
Searching for Python versions matching: Python 3.12
Installed 3 versions in 3.42s
 + cpython-3.10.14-macos-aarch64-none
 + cpython-3.11.9-macos-aarch64-none
 + cpython-3.12.4-macos-aarch64-none
```

Download Python versions as needed:

```console
$ uv venv --python 3.12.0
Using Python 3.12.0
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate

$ uv run --python pypy@3.8 -- python --version
Python 3.8.16 (a9dbdca6fc3286b0addd2240f11d97d8e8de187a, Dec 29 2022, 11:45:30)
[PyPy 7.3.11 with GCC Apple LLVM 13.1.6 (clang-1316.0.21.2.5)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>>
```

Use a specific Python version in the current directory:

```console
$ uv python pin 3.11
Pinned `.python-version` to `3.11`
```

See the [Python installation documentation](https://docs.astral.sh/uv/guides/install-python/) to get
started.

### The pip interface

uv provides a drop-in replacement for common `pip`, `pip-tools`, and `virtualenv` commands.

uv extends their interfaces with advanced features, such as dependency version overrides,
platform-independent resolutions, reproducible resolutions, alternative resolution strategies, and
more.

Migrate to uv without changing your existing workflows — and experience a 10-100x speedup — with the
`uv pip` interface.

Compile requirements into a platform-independent requirements file:

```console
$ uv pip compile docs/requirements.in \
   --universal \
   --output-file docs/requirements.txt
Resolved 43 packages in 12ms
```

Create a virtual environment:

```console
$ uv venv
Using Python 3.12.3
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate
```

Install the locked requirements:

```console
$ uv pip sync docs/requirements.txt
Resolved 43 packages in 11ms
Installed 43 packages in 208ms
 + babel==2.15.0
 + black==24.4.2
 + certifi==2024.7.4
 ...
```
