make this code better

[1rohitkhanal]

"""
Jarvis AI Assistant
A voice-enabled assistant with TTS and STT capabilities.
pip install SpeechRecognition pyttsx3 pyaudio
python jarvis.py # full interactive mode
python jarvis.py --tts-test # TTS only, no mic needed
"""

from future import annotations

import logging
import os
import subprocess
import sys
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

import speech_recognition as sr

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

Logging

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

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger("jarvis")

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

Configuration

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

class Platform(str, Enum):
WINDOWS = "win32"
DARWIN = "darwin"
LINUX = "linux"

@DataClass
class VoiceConfig:
volume: int = 75 # 0–100
rate: int = 175 # words per minute
timeout: int = 5 # listen timeout (seconds)
phrase_limit: int = 10 # max phrase duration (seconds)
energy_threshold: int = 300
dynamic_energy: bool = True
wake_word: str = "jarvis"

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

Text-to-Speech

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

class TTSEngine:
"""Cross-platform text-to-speech engine."""

def __init__(self, config: VoiceConfig) -> None:
    self.config = config
    self._platform = sys.platform
    self._engine = self._load_engine()

def _load_engine(self):
    """Try pyttsx3 first; fall back to platform-native TTS."""
    try:
        import pyttsx3  # optional dependency
        engine = pyttsx3.init()
        engine.setProperty("volume", self.config.volume / 100)
        engine.setProperty("rate", self.config.rate)
        logger.info("TTS: using pyttsx3")
        return engine
    except ImportError:
        logger.info("pyttsx3 not found — using platform-native TTS")
        return None

def speak(self, text: str) -> None:
    """Speak the given text, with sanitisation and error handling."""
    if not text or not text.strip():
        logger.warning("speak() called with empty text — skipping")
        return

    safe_text = text.replace("'", "\\'").replace('"', '\\"')
    logger.info("Speaking: %s", text)

    if self._engine is not None:
        self._speak_pyttsx3(safe_text)
    else:
        self._speak_native(safe_text)

def _speak_pyttsx3(self, text: str) -> None:
    try:
        self._engine.say(text)
        self._engine.runAndWait()
    except Exception as exc:
        logger.error("pyttsx3 error: %s", exc)
        self._speak_native(text)  # fallback

def _speak_native(self, text: str) -> None:
    if self._platform == Platform.WINDOWS:
        self._speak_windows(text)
    elif self._platform == Platform.DARWIN:
        self._speak_macos(text)
    else:
        self._speak_linux(text)

def _speak_windows(self, text: str) -> None:
    ps_script = (
        "Add-Type -AssemblyName System.Speech; "
        "$s = New-Object System.Speech.Synthesis.SpeechSynthesizer; "
        f"$s.Volume = {self.config.volume}; "
        f"$s.Rate = {max(-10, min(10, (self.config.rate - 175) // 25))}; "
        f"$s.Speak('{text}')"
    )
    self._run(["powershell", "-Command", ps_script])

def _speak_macos(self, text: str) -> None:
    self._run(["say", "-r", str(self.config.rate), text])

def _speak_linux(self, text: str) -> None:
    if self._command_exists("espeak"):
        self._run(["espeak", "-s", str(self.config.rate), text])
    elif self._command_exists("festival"):
        proc = subprocess.run(
            ["festival", "--tts"],
            input=text,
            capture_output=True,
            text=True,
            check=False,
        )
        if proc.returncode != 0:
            logger.error("festival error: %s", proc.stderr)
    else:
        logger.error("No TTS engine found. Install espeak: sudo apt install espeak")

@staticmethod
def _run(cmd: list[str]) -> None:
    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
    if result.returncode != 0:
        logger.error("TTS subprocess error: %s", result.stderr)

@staticmethod
def _command_exists(cmd: str) -> bool:
    return (
        subprocess.run(
            ["which", cmd], capture_output=True, check=False
        ).returncode == 0
    )

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

Speech-to-Text

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

class STTEngine:
"""Microphone-based speech recognition using Google Web Speech API."""

def __init__(self, config: VoiceConfig) -> None:
    self.config = config
    self.recognizer = sr.Recognizer()
    self.recognizer.energy_threshold = config.energy_threshold
    self.recognizer.dynamic_energy_threshold = config.dynamic_energy

def listen(self) -> Optional[str]:
    """
    Listen for a single utterance and return the transcript, or None on failure.
    """
    with sr.Microphone() as source:
        logger.info("Listening…")
        self.recognizer.adjust_for_ambient_noise(source, duration=0.5)
        try:
            audio = self.recognizer.listen(
                source,
                timeout=self.config.timeout,
                phrase_time_limit=self.config.phrase_limit,
            )
        except sr.WaitTimeoutError:
            logger.debug("Listen timeout — no speech detected")
            return None

    return self._recognise(audio)

def _recognise(self, audio: sr.AudioData) -> Optional[str]:
    try:
        text = self.recognizer.recognize_google(audio)
        logger.info("Heard: %s", text)
        return text.lower().strip()
    except sr.UnknownValueError:
        logger.debug("Could not understand audio")
    except sr.RequestError as exc:
        logger.error("Recognition service unavailable: %s", exc)
    return None

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

Command dispatcher

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

@DataClass
class CommandRegistry:
"""Simple keyword → handler mapping."""

_handlers: dict = field(default_factory=dict)

def register(self, keyword: str):
    """Decorator: @registry.register('weather')"""
    def decorator(fn):
        self._handlers[keyword.lower()] = fn
        return fn
    return decorator

def dispatch(self, text: str, assistant: "Jarvis") -> Optional[str]:
    for keyword, handler in self._handlers.items():
        if keyword in text:
            return handler(text, assistant)
    return None

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

Jarvis

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

class Jarvis:
"""
Main assistant class.

Usage:
    jarvis = Jarvis()
    jarvis.run()          # interactive loop
    jarvis.say("Hello")   # one-shot TTS
"""

def __init__(self, config: Optional[VoiceConfig] = None) -> None:
    self.config = config or VoiceConfig()
    self.tts = TTSEngine(self.config)
    self.stt = STTEngine(self.config)
    self.registry = CommandRegistry()
    self._running = False
    self._register_builtin_commands()

# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def say(self, text: str) -> None:
    self.tts.speak(text)

def run(self) -> None:
    """Enter the main listen → respond loop."""
    self.say("Jarvis online. Waiting for your command.")
    self._running = True
    try:
        while self._running:
            self._tick()
    except KeyboardInterrupt:
        logger.info("Interrupted by user")
    finally:
        self.say("Shutting down. Goodbye.")
        logger.info("Jarvis stopped")

# ------------------------------------------------------------------
# Internal
# ------------------------------------------------------------------
def _tick(self) -> None:
    text = self.stt.listen()
    if text is None:
        return

    if self.config.wake_word and self.config.wake_word not in text:
        logger.debug("Wake word not detected, ignoring: %s", text)
        return

    response = self.registry.dispatch(text, self)
    if response:
        self.say(response)
    else:
        self.say("I'm not sure how to handle that yet.")

def _register_builtin_commands(self) -> None:
    registry = self.registry

    @registry.register("stop")
    def cmd_stop(text: str, bot: Jarvis) -> str:
        bot._running = False
        return "Stopping."

    @registry.register("hello")
    def cmd_hello(text: str, bot: Jarvis) -> str:
        return "Hello! How can I help you?"

    @registry.register("time")
    def cmd_time(text: str, bot: Jarvis) -> str:
        from datetime import datetime
        return f"The time is {datetime.now().strftime('%I:%M %p')}."

    @registry.register("date")
    def cmd_date(text: str, bot: Jarvis) -> str:
        from datetime import date
        return f"Today is {date.today().strftime('%A, %B %d %Y')}."

    @registry.register("who are you")
    def cmd_identity(text: str, bot: Jarvis) -> str:
        return "I am Jarvis, your personal AI assistant."

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

Entry point

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

def main() -> None:
config = VoiceConfig(
volume=80,
rate=175,
timeout=5,
wake_word="jarvis",
)
jarvis = Jarvis(config)

# Quick smoke-test without mic
if "--tts-test" in sys.argv:
    jarvis.say("Text-to-speech test successful.")
    return

jarvis.run()

if name == "main":
main()