Understanding Audio Data

Audio is a continuous signal that represents variations in air pressure over time. When we work with audio in computers, we need to convert this continuous signal into a discrete representation through a process called sampling.

Working with Audio Files in Python

Python offers several libraries for working with audio data. We'll focus on librosa, a powerful library for audio and music analysis, and soundfile for reading and writing audio files.

Loading and Playing Audio

import librosa
import librosa.display
import matplotlib.pyplot as plt
import numpy as np
import soundfile as sf
from IPython.display import Audio

# Load an audio file
file_path = 'path/to/your/audio_file.wav'
audio, sample_rate = librosa.load(file_path, sr=None)  # sr=None preserves the original sample rate

# Display basic information
print(f"Audio shape: {audio.shape}")
print(f"Sample rate: {sample_rate} Hz")
print(f"Duration: {len(audio) / sample_rate:.2f} seconds")

# Play the audio (in Jupyter notebooks)
Audio(data=audio, rate=sample_rate)

# Visualize the waveform
plt.figure(figsize=(12, 4))
librosa.display.waveshow(audio, sr=sample_rate)
plt.title('Audio Waveform')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')
plt.tight_layout()
plt.show()

Recording Audio

To record audio from a microphone, we can use the pyaudio library:

import pyaudio
import wave
import numpy as np

def record_audio(filename, duration=5, sample_rate=16000, channels=1):
    """
    Record audio from the microphone and save to a file.
    
    Parameters:
    - filename: Output file name (WAV format)
    - duration: Recording duration in seconds
    - sample_rate: Sampling rate in Hz
    - channels: Number of audio channels
    """
    # Initialize PyAudio
    p = pyaudio.PyAudio()
    
    # Open stream
    stream = p.open(format=pyaudio.paInt16,
                    channels=channels,
                    rate=sample_rate,
                    input=True,
                    frames_per_buffer=1024)
    
    print(f"Recording for {duration} seconds...")
    
    frames = []
    
    # Record audio in chunks
    for i in range(0, int(sample_rate / 1024 * duration)):
        data = stream.read(1024)
        frames.append(data)
    
    print("Recording finished.")
    
    # Stop and close the stream
    stream.stop_stream()
    stream.close()
    p.terminate()
    
    # Save the recorded audio to a WAV file
    wf = wave.open(filename, 'wb')
    wf.setnchannels(channels)
    wf.setsampwidth(p.get_sample_size(pyaudio.paInt16))
    wf.setframerate(sample_rate)
    wf.writeframes(b''.join(frames))
    wf.close()
    
    print(f"Audio saved to {filename}")

# Example usage
record_audio('recorded_audio.wav', duration=5)

Audio Feature Extraction

Audio features are numerical representations that capture different aspects of audio signals. These features are essential for tasks like speech recognition, music classification, and audio analysis.

Time-Domain Features

import librosa
import numpy as np
import matplotlib.pyplot as plt

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Calculate energy
energy = np.sum(audio**2) / len(audio)
print(f"Energy: {energy:.6f}")

# Calculate zero-crossing rate
zero_crossings = librosa.feature.zero_crossing_rate(audio)[0]
print(f"Zero-crossing rate: {np.mean(zero_crossings):.6f}")

# Calculate root mean square energy (RMS)
rms = librosa.feature.rms(y=audio)[0]
print(f"RMS energy: {np.mean(rms):.6f}")

# Visualize RMS energy over time
plt.figure(figsize=(12, 8))

plt.subplot(2, 1, 1)
librosa.display.waveshow(audio, sr=sample_rate)
plt.title('Audio Waveform')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')

plt.subplot(2, 1, 2)
frames = np.arange(len(rms))
t = librosa.frames_to_time(frames, sr=sample_rate)
plt.plot(t, rms)
plt.title('RMS Energy Over Time')
plt.xlabel('Time (s)')
plt.ylabel('RMS Energy')

plt.tight_layout()
plt.show()

Frequency-Domain Features

The frequency domain provides insights into the spectral content of audio signals. The Short-Time Fourier Transform (STFT) is a common technique to analyze how frequency content changes over time.

import librosa
import librosa.display
import numpy as np
import matplotlib.pyplot as plt

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Compute the Short-Time Fourier Transform (STFT)
stft = librosa.stft(audio)
magnitude = np.abs(stft)  # Magnitude of the STFT
phase = np.angle(stft)    # Phase of the STFT

# Convert to decibels (dB)
magnitude_db = librosa.amplitude_to_db(magnitude, ref=np.max)

# Visualize the spectrogram
plt.figure(figsize=(12, 8))

plt.subplot(2, 1, 1)
librosa.display.waveshow(audio, sr=sample_rate)
plt.title('Audio Waveform')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')

plt.subplot(2, 1, 2)
librosa.display.specshow(magnitude_db, sr=sample_rate, x_axis='time', y_axis='hz')
plt.colorbar(format='%+2.0f dB')
plt.title('Spectrogram')
plt.xlabel('Time (s)')
plt.ylabel('Frequency (Hz)')

plt.tight_layout()
plt.show()

# Extract spectral features
spectral_centroid = librosa.feature.spectral_centroid(y=audio, sr=sample_rate)[0]
spectral_bandwidth = librosa.feature.spectral_bandwidth(y=audio, sr=sample_rate)[0]
spectral_rolloff = librosa.feature.spectral_rolloff(y=audio, sr=sample_rate)[0]

print(f"Spectral Centroid (mean): {np.mean(spectral_centroid):.2f} Hz")
print(f"Spectral Bandwidth (mean): {np.mean(spectral_bandwidth):.2f} Hz")
print(f"Spectral Rolloff (mean): {np.mean(spectral_rolloff):.2f} Hz")

Mel-Frequency Cepstral Coefficients (MFCCs)

MFCCs are one of the most widely used features in speech and audio processing. They capture the short-term power spectrum of a sound and are particularly useful for speech recognition.

import librosa
import librosa.display
import numpy as np
import matplotlib.pyplot as plt

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Extract MFCCs
mfccs = librosa.feature.mfcc(y=audio, sr=sample_rate, n_mfcc=13)

# Visualize MFCCs
plt.figure(figsize=(12, 4))
librosa.display.specshow(mfccs, sr=sample_rate, x_axis='time')
plt.colorbar(format='%+2.0f')
plt.title('MFCCs')
plt.xlabel('Time (s)')
plt.ylabel('MFCC Coefficients')
plt.tight_layout()
plt.show()

# Calculate statistics of MFCCs
mfcc_means = np.mean(mfccs, axis=1)
mfcc_vars = np.var(mfccs, axis=1)

print("MFCC Means:")
for i, mean in enumerate(mfcc_means):
    print(f"MFCC {i+1}: {mean:.4f}")

Basic Audio Manipulations

Now that we understand how to load and analyze audio, let's explore some basic manipulations we can perform on audio signals.

Changing Volume

import librosa
import soundfile as sf
import numpy as np

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Increase volume (multiply by a factor > 1)
audio_louder = audio * 1.5

# Decrease volume (multiply by a factor < 1)
audio_quieter = audio * 0.5

# Save the modified audio
sf.write('louder_audio.wav', audio_louder, sample_rate)
sf.write('quieter_audio.wav', audio_quieter, sample_rate)

Changing Speed and Pitch

import librosa
import soundfile as sf
import numpy as np
import pyrubberband as pyrb

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Change speed (without changing pitch)
# Factor > 1 speeds up, factor < 1 slows down
speed_factor = 1.5
audio_fast = pyrb.time_stretch(audio, sample_rate, speed_factor)

# Change pitch (without changing speed)
# Positive semitones increase pitch, negative decrease
pitch_shift = 4  # Shift up by 4 semitones
audio_high_pitch = pyrb.pitch_shift(audio, sample_rate, pitch_shift)

# Save the modified audio
sf.write('fast_audio.wav', audio_fast, sample_rate)
sf.write('high_pitch_audio.wav', audio_high_pitch, sample_rate)

# Using librosa for time stretching and pitch shifting
audio_fast = librosa.effects.time_stretch(audio, rate=speed_factor)
audio_high_pitch = librosa.effects.pitch_shift(audio, sr=sample_rate, n_steps=pitch_shift)

Applying Filters

import librosa
import soundfile as sf
import numpy as np
from scipy import signal

# Load audio file
audio, sample_rate = librosa.load('path/to/your/audio_file.wav', sr=None)

# Low-pass filter (keeps frequencies below the cutoff)
cutoff_low = 1000  # 1000 Hz cutoff
b, a = signal.butter(4, cutoff_low, 'low', fs=sample_rate)
audio_low_pass = signal.filtfilt(b, a, audio)

# High-pass filter (keeps frequencies above the cutoff)
cutoff_high = 1000  # 1000 Hz cutoff
b, a = signal.butter(4, cutoff_high, 'high', fs=sample_rate)
audio_high_pass = signal.filtfilt(b, a, audio)

# Band-pass filter (keeps frequencies between the cutoffs)
cutoff_low = 500   # 500 Hz lower cutoff
cutoff_high = 2000  # 2000 Hz upper cutoff
b, a = signal.butter(4, [cutoff_low, cutoff_high], 'band', fs=sample_rate)
audio_band_pass = signal.filtfilt(b, a, audio)

# Save the filtered audio
sf.write('low_pass_audio.wav', audio_low_pass, sample_rate)
sf.write('high_pass_audio.wav', audio_high_pass, sample_rate)
sf.write('band_pass_audio.wav', audio_band_pass, sample_rate)

Noise Reduction

Noise reduction is a common preprocessing step for speech recognition and other audio applications. Here's a simple approach using spectral subtraction:

import librosa
import soundfile as sf
import numpy as np
import matplotlib.pyplot as plt

# Load audio file
audio, sample_rate = librosa.load('path/to/your/noisy_audio.wav', sr=None)

# Assume the first 1 second is noise (adjust as needed)
noise_sample = audio[:int(sample_rate)]

# Compute the noise profile
noise_stft = librosa.stft(noise_sample)
noise_power = np.mean(np.abs(noise_stft)**2, axis=1)
noise_power = noise_power[:, np.newaxis]

# Compute the STFT of the audio
audio_stft = librosa.stft(audio)
audio_power = np.abs(audio_stft)**2

# Perform spectral subtraction
gain = 1 - (noise_power / audio_power)
gain = np.maximum(0, gain)  # Ensure non-negative values
gain = gain**0.5  # Apply square root for magnitude

# Apply the gain to the STFT
audio_stft_denoised = audio_stft * gain

# Convert back to time domain
audio_denoised = librosa.istft(audio_stft_denoised)

# Save the denoised audio
sf.write('denoised_audio.wav', audio_denoised, sample_rate)

# Visualize the results
plt.figure(figsize=(12, 8))

plt.subplot(2, 1, 1)
librosa.display.waveshow(audio, sr=sample_rate)
plt.title('Original Noisy Audio')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')

plt.subplot(2, 1, 2)
librosa.display.waveshow(audio_denoised, sr=sample_rate)
plt.title('Denoised Audio')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')

plt.tight_layout()
plt.show()

Understanding Speech Recognition

Speech recognition systems typically follow a pipeline that includes:

1. Audio Capture: Recording audio from a microphone or loading from a file
2. Preprocessing: Noise reduction, normalization, and feature extraction
3. Feature Extraction: Converting audio into features like MFCCs
4. Acoustic Modeling: Mapping audio features to phonetic units
5. Language Modeling: Determining the most likely sequence of words
6. Decoding: Converting the model output into text

Modern speech recognition systems use deep learning models like recurrent neural networks (RNNs), convolutional neural networks (CNNs), and transformer-based architectures.

Speech Recognition in Python

Python offers several libraries for speech recognition, ranging from simple API wrappers to complete deep learning frameworks.

Using the SpeechRecognition Library

The SpeechRecognition library provides a simple interface to various speech recognition APIs and engines.

import speech_recognition as sr

def recognize_from_file(audio_file_path, language='en-US'):
    """
    Recognize speech from an audio file using Google's speech recognition API.
    
    Parameters:
    - audio_file_path: Path to the audio file
    - language: Language code (default: 'en-US')
    
    Returns:
    - Recognized text
    """
    # Initialize recognizer
    recognizer = sr.Recognizer()
    
    # Load audio file
    with sr.AudioFile(audio_file_path) as source:
        # Record the audio data
        audio_data = recognizer.record(source)
        
        try:
            # Recognize speech using Google Speech Recognition
            text = recognizer.recognize_google(audio_data, language=language)
            print(f"Google Speech Recognition thinks you said: {text}")
            return text
        except sr.UnknownValueError:
            print("Google Speech Recognition could not understand audio")
            return None
        except sr.RequestError as e:
            print(f"Could not request results from Google Speech Recognition service; {e}")
            return None

def recognize_from_microphone(language='en-US', duration=5):
    """
    Recognize speech from the microphone using Google's speech recognition API.
    
    Parameters:
    - language: Language code (default: 'en-US')
    - duration: Recording duration in seconds (default: 5)
    
    Returns:
    - Recognized text
    """
    # Initialize recognizer
    recognizer = sr.Recognizer()
    
    # Use the microphone as source
    with sr.Microphone() as source:
        print("Adjusting for ambient noise...")
        recognizer.adjust_for_ambient_noise(source, duration=1)
        
        print(f"Listening for {duration} seconds...")
        audio_data = recognizer.listen(source, timeout=duration)
        
        try:
            # Recognize speech using Google Speech Recognition
            text = recognizer.recognize_google(audio_data, language=language)
            print(f"Google Speech Recognition thinks you said: {text}")
            return text
        except sr.UnknownValueError:
            print("Google Speech Recognition could not understand audio")
            return None
        except sr.RequestError as e:
            print(f"Could not request results from Google Speech Recognition service; {e}")
            return None

# Example usage
if __name__ == "__main__":
    # Recognize speech from a file
    text = recognize_from_file('path/to/your/audio_file.wav')
    
    # Recognize speech from the microphone
    # text = recognize_from_microphone(duration=5)

Using CMU Sphinx for Offline Recognition

If you need offline speech recognition, you can use CMU Sphinx through the pocketsphinx library:

import speech_recognition as sr

def recognize_offline(audio_file_path):
    """
    Recognize speech from an audio file using CMU Sphinx (offline).
    
    Parameters:
    - audio_file_path: Path to the audio file
    
    Returns:
    - Recognized text
    """
    # Initialize recognizer
    recognizer = sr.Recognizer()
    
    # Load audio file
    with sr.AudioFile(audio_file_path) as source:
        # Record the audio data
        audio_data = recognizer.record(source)
        
        try:
            # Recognize speech using Sphinx
            text = recognizer.recognize_sphinx(audio_data)
            print(f"Sphinx thinks you said: {text}")
            return text
        except sr.UnknownValueError:
            print("Sphinx could not understand audio")
            return None
        except sr.RequestError as e:
            print(f"Sphinx error; {e}")
            return None

# Example usage
text = recognize_offline('path/to/your/audio_file.wav')

Advanced Speech Recognition with Deep Learning

For more advanced speech recognition tasks, you can use deep learning libraries like TensorFlow or PyTorch with pre-trained models.

Using Hugging Face Transformers

The Hugging Face Transformers library provides access to state-of-the-art speech recognition models like Wav2Vec2:

import torch
from transformers import Wav2Vec2ForCTC, Wav2Vec2Processor
import librosa
import numpy as np

def recognize_with_wav2vec2(audio_file_path, model_name="facebook/wav2vec2-base-960h"):
    """
    Recognize speech from an audio file using Wav2Vec2.
    
    Parameters:
    - audio_file_path: Path to the audio file
    - model_name: Name of the pre-trained model
    
    Returns:
    - Recognized text
    """
    # Load the model and processor
    processor = Wav2Vec2Processor.from_pretrained(model_name)
    model = Wav2Vec2ForCTC.from_pretrained(model_name)
    
    # Load audio file
    audio, sample_rate = librosa.load(audio_file_path, sr=16000)
    
    # Process the audio
    inputs = processor(audio=audio, sampling_rate=16000, return_tensors="pt", padding=True)
    
    # Perform inference
    with torch.no_grad():
        logits = model(inputs.input_values, attention_mask=inputs.attention_mask).logits
    
    # Decode the predicted IDs
    predicted_ids = torch.argmax(logits, dim=-1)
    transcription = processor.batch_decode(predicted_ids)
    
    return transcription[0]

# Example usage
text = recognize_with_wav2vec2('path/to/your/audio_file.wav')
print(f"Wav2Vec2 thinks you said: {text}")

Using Whisper

OpenAI's Whisper is a powerful speech recognition model that supports multiple languages and can handle noisy environments:

import whisper

def recognize_with_whisper(audio_file_path, model_size="base"):
    """
    Recognize speech from an audio file using OpenAI's Whisper.
    
    Parameters:
    - audio_file_path: Path to the audio file
    - model_size: Size of the Whisper model ('tiny', 'base', 'small', 'medium', 'large')
    
    Returns:
    - Recognized text
    """
    # Load the model
    model = whisper.load_model(model_size)
    
    # Transcribe the audio
    result = model.transcribe(audio_file_path)
    
    return result["text"]

# Example usage
text = recognize_with_whisper('path/to/your/audio_file.wav', model_size="base")
print(f"Whisper thinks you said: {text}")

Real-Time Speech Recognition

For voice assistants, real-time speech recognition is essential. Here's how to implement it using the SpeechRecognition library:

import speech_recognition as sr
import time

def real_time_speech_recognition(timeout=None, phrase_time_limit=None):
    """
    Perform real-time speech recognition from the microphone.
    
    Parameters:
    - timeout: Maximum number of seconds to wait for speech (None for no timeout)
    - phrase_time_limit: Maximum number of seconds for a phrase (None for no limit)
    
    Returns:
    - Generator yielding recognized text
    """
    # Initialize recognizer
    recognizer = sr.Recognizer()
    
    # Use the microphone as source
    with sr.Microphone() as source:
        print("Adjusting for ambient noise...")
        recognizer.adjust_for_ambient_noise(source, duration=1)
        
        print("Listening... (Press Ctrl+C to stop)")
        
        try:
            while True:
                try:
                    print("Waiting for speech...")
                    audio_data = recognizer.listen(source, timeout=timeout, phrase_time_limit=phrase_time_limit)
                    
                    try:
                        # Recognize speech using Google Speech Recognition
                        text = recognizer.recognize_google(audio_data)
                        print(f"Recognized: {text}")
                        yield text
                    except sr.UnknownValueError:
                        print("Could not understand audio")
                    except sr.RequestError as e:
                        print(f"Request error: {e}")
                except sr.WaitTimeoutError:
                    print("Timeout waiting for speech")
        except KeyboardInterrupt:
            print("Stopped by user")
            return

# Example usage
if __name__ == "__main__":
    for text in real_time_speech_recognition(timeout=5, phrase_time_limit=5):
        # Process the recognized text
        if text.lower() == "stop":
            print("Stopping...")
            break
        
        # Respond to the recognized text
        print(f"You said: {text}")

Voice Activity Detection (VAD)

Voice Activity Detection (VAD) is a technique to detect the presence of speech in an audio signal. It's useful for real-time speech recognition to determine when to start and stop recording:

import numpy as np
import librosa
import pyaudio
import wave
import webrtcvad
import struct
import collections

def vad_collector(sample_rate=16000, frame_duration_ms=30, padding_duration_ms=300, vad=None, frames=None):
    """
    Generator that yields series of consecutive audio frames comprising each utterance.
    
    Parameters:
    - sample_rate: Audio sample rate in Hz
    - frame_duration_ms: Duration of each frame in milliseconds
    - padding_duration_ms: Amount of padding to include before and after each utterance
    - vad: Voice activity detector
    - frames: Audio frames
    
    Returns:
    - Generator yielding utterances as a list of frames
    """
    if vad is None:
        vad = webrtcvad.Vad(3)  # Aggressiveness mode (0-3)
    
    num_padding_frames = int(padding_duration_ms / frame_duration_ms)
    ring_buffer = collections.deque(maxlen=num_padding_frames)
    triggered = False
    
    for frame in frames:
        is_speech = vad.is_speech(frame, sample_rate)
        
        if not triggered:
            ring_buffer.append((frame, is_speech))
            num_voiced = len([f for f, speech in ring_buffer if speech])
            
            if num_voiced > 0.9 * ring_buffer.maxlen:
                triggered = True
                for f, s in ring_buffer:
                    yield f
                ring_buffer.clear()
        else:
            yield frame
            ring_buffer.append((frame, is_speech))
            num_unvoiced = len([f for f, speech in ring_buffer if not speech])
            
            if num_unvoiced > 0.9 * ring_buffer.maxlen:
                triggered = False
                yield None  # Signal the end of an utterance
                ring_buffer.clear()

def record_with_vad(output_file, duration=10, sample_rate=16000, frame_duration_ms=30):
    """
    Record audio with voice activity detection.
    
    Parameters:
    - output_file: Output WAV file
    - duration: Maximum recording duration in seconds
    - sample_rate: Audio sample rate in Hz
    - frame_duration_ms: Duration of each frame in milliseconds
    
    Returns:
    - List of utterances (each utterance is a list of frames)
    """
    # Initialize PyAudio
    p = pyaudio.PyAudio()
    
    # Open stream
    stream = p.open(format=pyaudio.paInt16,
                    channels=1,
                    rate=sample_rate,
                    input=True,
                    frames_per_buffer=int(sample_rate * frame_duration_ms / 1000))
    
    # Initialize VAD
    vad = webrtcvad.Vad(3)  # Aggressiveness mode (0-3)
    
    print(f"Recording for {duration} seconds with VAD...")
    
    frames = []
    utterances = []
    current_utterance = []
    
    # Record audio in chunks
    for i in range(0, int(sample_rate / (sample_rate * frame_duration_ms / 1000) * duration)):
        frame = stream.read(int(sample_rate * frame_duration_ms / 1000))
        frames.append(frame)
    
    # Stop and close the stream
    stream.stop_stream()
    stream.close()
    p.terminate()
    
    # Process frames with VAD
    for frame in vad_collector(sample_rate, frame_duration_ms, 300, vad, frames):
        if frame is None:
            if current_utterance:
                utterances.append(current_utterance)
                current_utterance = []
        else:
            current_utterance.append(frame)
    
    # Add the last utterance if it exists
    if current_utterance:
        utterances.append(current_utterance)
    
    # Save the utterances to a WAV file
    if utterances:
        with wave.open(output_file, 'wb') as wf:
            wf.setnchannels(1)
            wf.setsampwidth(p.get_sample_size(pyaudio.paInt16))
            wf.setframerate(sample_rate)
            for utterance in utterances:
                for frame in utterance:
                    wf.writeframes(frame)
    
    print(f"Recorded {len(utterances)} utterances")
    print(f"Audio saved to {output_file}")
    
    return utterances

# Example usage
if __name__ == "__main__":
    utterances = record_with_vad('vad_recording.wav', duration=10)

Understanding Text-to-Speech Synthesis

Text-to-speech synthesis typically involves several steps:

1. Text Analysis: Parsing and normalizing the input text
2. Phonetic Conversion: Converting text to phonetic representations
3. Prosody Generation: Determining rhythm, stress, and intonation
4. Waveform Generation: Generating the audio waveform

Modern TTS systems use deep learning models to generate natural-sounding speech. These models can be categorized into several types:

• Concatenative TTS: Combines pre-recorded speech segments
• Parametric TTS: Uses statistical models to generate speech parameters
• Neural TTS: Uses neural networks to generate speech directly

Text-to-Speech in Python

Python offers several libraries for text-to-speech synthesis, ranging from simple wrappers to advanced neural TTS systems.

Using pyttsx3 (Offline TTS)

pyttsx3 is a text-to-speech conversion library that works offline. It uses the speech engines available on your system (e.g., SAPI5 on Windows, NSSpeechSynthesizer on macOS, eSpeak on Linux).

import pyttsx3

def text_to_speech(text, voice_id=None, rate=200, volume=1.0, save_to_file=None):
    """
    Convert text to speech using pyttsx3.
    
    Parameters:
    - text: Text to convert to speech
    - voice_id: Voice ID to use (None for default)
    - rate: Speech rate (words per minute)
    - volume: Volume (0.0 to 1.0)
    - save_to_file: Path to save the speech to a file (None to play directly)
    """
    # Initialize the TTS engine
    engine = pyttsx3.init()
    
    # Set properties
    engine.setProperty('rate', rate)
    engine.setProperty('volume', volume)
    
    # Set voice if specified
    if voice_id is not None:
        engine.setProperty('voice', voice_id)
    
    # Get available voices
    voices = engine.getProperty('voices')
    print(f"Available voices: {len(voices)}")
    for i, voice in enumerate(voices):
        print(f"Voice {i}: {voice.id} - {voice.name}")
    
    # Save to file or speak directly
    if save_to_file:
        engine.save_to_file(text, save_to_file)
        engine.runAndWait()
        print(f"Speech saved to {save_to_file}")
    else:
        engine.say(text)
        engine.runAndWait()

# Example usage
if __name__ == "__main__":
    # Speak directly
    text_to_speech("Hello, this is a test of text-to-speech synthesis using pyttsx3.")
    
    # Save to file
    text_to_speech("This is a test of saving speech to a file.", save_to_file="tts_output.mp3")
    
    # Use a different voice (if available)
    # text_to_speech("This is a test with a different voice.", voice_id="HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Speech\\Voices\\Tokens\\TTS_MS_EN-US_ZIRA_11.0")

Using gTTS (Google Text-to-Speech)

gTTS (Google Text-to-Speech) is a Python library and CLI tool that interfaces with Google Translate's text-to-speech API. It requires an internet connection but provides high-quality speech.

from gtts import gTTS
import os
from io import BytesIO
from pydub import AudioSegment
from pydub.playback import play

def text_to_speech_gtts(text, lang='en', slow=False, save_to_file=None, play_audio=True):
    """
    Convert text to speech using Google Text-to-Speech.
    
    Parameters:
    - text: Text to convert to speech
    - lang: Language code (e.g., 'en', 'fr', 'es')
    - slow: Whether to speak slowly
    - save_to_file: Path to save the speech to a file (None to play directly)
    - play_audio: Whether to play the audio (if save_to_file is None)
    """
    # Create gTTS object
    tts = gTTS(text=text, lang=lang, slow=slow)
    
    # Save to file or play directly
    if save_to_file:
        tts.save(save_to_file)
        print(f"Speech saved to {save_to_file}")
    elif play_audio:
        # Save to a BytesIO object
        fp = BytesIO()
        tts.write_to_fp(fp)
        fp.seek(0)
        
        # Convert to AudioSegment and play
        audio = AudioSegment.from_file(fp, format="mp3")
        play(audio)

# Example usage
if __name__ == "__main__":
    # Speak directly
    text_to_speech_gtts("Hello, this is a test of text-to-speech synthesis using Google Text-to-Speech.")
    
    # Save to file
    text_to_speech_gtts("This is a test of saving speech to a file.", save_to_file="gtts_output.mp3")
    
    # Use a different language
    text_to_speech_gtts("Bonjour, comment ça va?", lang='fr')

from gtts.lang import tts_langs
print(tts_langs())

Advanced Text-to-Speech with Deep Learning

For more advanced text-to-speech tasks, you can use deep learning libraries like TensorFlow or PyTorch with pre-trained models.

Using Mozilla TTS

Mozilla TTS is an open-source text-to-speech framework that provides high-quality speech synthesis using deep learning models.

from TTS.utils.manage import ModelManager
from TTS.utils.synthesizer import Synthesizer
import numpy as np
import soundfile as sf

def text_to_speech_mozilla(text, model_name="tts_models/en/ljspeech/tacotron2-DDC", vocoder_name="vocoder_models/en/ljspeech/multiband-melgan", save_to_file=None):
    """
    Convert text to speech using Mozilla TTS.
    
    Parameters:
    - text: Text to convert to speech
    - model_name: Name of the TTS model
    - vocoder_name: Name of the vocoder model
    - save_to_file: Path to save the speech to a file (None to return the audio array)
    
    Returns:
    - Audio array if save_to_file is None, otherwise None
    """
    # Initialize model manager
    model_manager = ModelManager()
    
    # Download models if not already downloaded
    model_path, config_path, model_item = model_manager.download_model(model_name)
    vocoder_path, vocoder_config_path, _ = model_manager.download_model(vocoder_name)
    
    # Initialize synthesizer
    synthesizer = Synthesizer(
        tts_checkpoint=model_path,
        tts_config_path=config_path,
        vocoder_checkpoint=vocoder_path,
        vocoder_config=vocoder_config_path
    )
    
    # Synthesize speech
    outputs = synthesizer.tts(text)
    
    # Save to file or return the audio array
    if save_to_file:
        sf.write(save_to_file, outputs["wav"], outputs["sampling_rate"])
        print(f"Speech saved to {save_to_file}")
        return None
    else:
        return outputs["wav"], outputs["sampling_rate"]

# Example usage
if __name__ == "__main__":
    # Synthesize speech and save to file
    text_to_speech_mozilla("Hello, this is a test of text-to-speech synthesis using Mozilla TTS.", save_to_file="mozilla_tts_output.wav")
    
    # Synthesize speech and get the audio array
    audio, sample_rate = text_to_speech_mozilla("This is another test.")
    print(f"Audio shape: {audio.shape}, Sample rate: {sample_rate}")

Using Hugging Face Transformers

The Hugging Face Transformers library provides access to state-of-the-art text-to-speech models like SpeechT5:

import torch
from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan
import soundfile as sf

def text_to_speech_huggingface(text, speaker_embedding=None, save_to_file=None):
    """
    Convert text to speech using Hugging Face Transformers.
    
    Parameters:
    - text: Text to convert to speech
    - speaker_embedding: Speaker embedding for voice cloning (None for default)
    - save_to_file: Path to save the speech to a file (None to return the audio array)
    
    Returns:
    - Audio array if save_to_file is None, otherwise None
    """
    # Load models and processor
    processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_tts")
    model = SpeechT5ForTextToSpeech.from_pretrained("microsoft/speecht5_tts")
    vocoder = SpeechT5HifiGan.from_pretrained("microsoft/speecht5_hifigan")
    
    # Process text
    inputs = processor(text=text, return_tensors="pt")
    
    # Generate speech
    if speaker_embedding is None:
        # Use a random speaker embedding
        speaker_embedding = torch.randn(1, 512)
    
    speech = model.generate_speech(inputs["input_ids"], speaker_embedding, vocoder=vocoder)
    
    # Save to file or return the audio array
    if save_to_file:
        sf.write(save_to_file, speech.numpy(), 16000)
        print(f"Speech saved to {save_to_file}")
        return None
    else:
        return speech.numpy(), 16000

# Example usage
if __name__ == "__main__":
    # Synthesize speech and save to file
    text_to_speech_huggingface("Hello, this is a test of text-to-speech synthesis using Hugging Face Transformers.", save_to_file="huggingface_tts_output.wav")

from transformers import SpeechT5Processor, SpeechT5ForSpeechToSpeech
import librosa

# Load the model and processor
processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_vc")
model = SpeechT5ForSpeechToSpeech.from_pretrained("microsoft/speecht5_vc")

# Load a reference audio file
audio, sample_rate = librosa.load("reference_audio.wav", sr=16000)

# Extract the speaker embedding
inputs = processor(audio=audio, sampling_rate=sample_rate, return_tensors="pt")
speaker_embedding = model.get_speaker_embeddings(inputs["input_values"])

# Use the speaker embedding for text-to-speech
text_to_speech_huggingface("This is voice cloning with SpeechT5.", speaker_embedding=speaker_embedding, save_to_file="voice_cloning_output.wav")

Customizing Text-to-Speech Output

Text-to-speech systems often provide ways to customize the output, such as changing the voice, speed, pitch, and adding pauses or emphasis.

Using SSML (Speech Synthesis Markup Language)

SSML is a markup language that allows you to control how text is spoken. It's supported by many TTS systems, including Google Cloud Text-to-Speech and Amazon Polly.

from google.cloud import texttospeech
import os

# Set up Google Cloud credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "path/to/your/credentials.json"

def text_to_speech_ssml(ssml_text, language_code="en-US", voice_name="en-US-Wavenet-D", save_to_file="output.mp3"):
    """
    Convert SSML text to speech using Google Cloud Text-to-Speech.
    
    Parameters:
    - ssml_text: SSML text to convert to speech
    - language_code: Language code
    - voice_name: Voice name
    - save_to_file: Path to save the speech to a file
    """
    # Initialize client
    client = texttospeech.TextToSpeechClient()
    
    # Set the input
    synthesis_input = texttospeech.SynthesisInput(ssml=ssml_text)
    
    # Set the voice
    voice = texttospeech.VoiceSelectionParams(
        language_code=language_code,
        name=voice_name
    )
    
    # Set the audio config
    audio_config = texttospeech.AudioConfig(
        audio_encoding=texttospeech.AudioEncoding.MP3
    )
    
    # Perform the synthesis
    response = client.synthesize_speech(
        input=synthesis_input,
        voice=voice,
        audio_config=audio_config
    )
    
    # Save the audio to a file
    with open(save_to_file, "wb") as out:
        out.write(response.audio_content)
        print(f"Audio content written to {save_to_file}")

# Example usage
if __name__ == "__main__":
    ssml_text = """
    
        Here's an example of SSML.
        
        You can add pauses, change the speaking rate,
        adjust the pitch,
        and even add SSML.
        
    
    """
    
    text_to_speech_ssml(ssml_text, save_to_file="ssml_output.mp3")

Voice Assistant Architecture

A typical voice assistant consists of several key components working together:

Natural Language Understanding (NLU)

Natural Language Understanding is a critical component of voice assistants that extracts meaning from the user's speech. It involves identifying the user's intent and extracting relevant entities from their utterances.

Intent Recognition

Intent recognition determines what the user wants to do. For example, "What's the weather like today?" has a weather intent, while "Set an alarm for 7 AM" has an alarm intent.

import spacy
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import LinearSVC
import joblib
import numpy as np

class IntentClassifier:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.vectorizer = TfidfVectorizer(ngram_range=(1, 2), max_features=1000)
        self.classifier = LinearSVC()
        self.intents = []
        
    def train(self, training_data):
        """
        Train the intent classifier.
        
        Parameters:
        - training_data: List of (text, intent) tuples
        """
        texts, intents = zip(*training_data)
        
        # Preprocess texts
        processed_texts = [self._preprocess(text) for text in texts]
        
        # Vectorize texts
        X = self.vectorizer.fit_transform(processed_texts)
        
        # Train classifier
        self.classifier.fit(X, intents)
        
        # Store unique intents
        self.intents = list(set(intents))
        
    def predict(self, text):
        """
        Predict the intent of a text.
        
        Parameters:
        - text: Input text
        
        Returns:
        - Predicted intent
        - Confidence score
        """
        processed_text = self._preprocess(text)
        X = self.vectorizer.transform([processed_text])
        
        # Get prediction
        intent = self.classifier.predict(X)[0]
        
        # Get confidence score
        decision_values = self.classifier.decision_function(X)
        confidence = np.max(decision_values)
        
        return intent, confidence
    
    def _preprocess(self, text):
        """Preprocess text by lemmatizing and removing stopwords"""
        doc = self.nlp(text.lower())
        tokens = [token.lemma_ for token in doc if not token.is_stop and not token.is_punct]
        return " ".join(tokens)
    
    def save(self, filepath):
        """Save the model to a file"""
        model_data = {
            "vectorizer": self.vectorizer,
            "classifier": self.classifier,
            "intents": self.intents
        }
        joblib.dump(model_data, filepath)
    
    def load(self, filepath):
        """Load the model from a file"""
        model_data = joblib.load(filepath)
        self.vectorizer = model_data["vectorizer"]
        self.classifier = model_data["classifier"]
        self.intents = model_data["intents"]

# Example usage
if __name__ == "__main__":
    # Training data: (text, intent) pairs
    training_data = [
        ("What's the weather like today", "weather"),
        ("What's the forecast for tomorrow", "weather"),
        ("Will it rain this weekend", "weather"),
        ("Set an alarm for 7 AM", "set_alarm"),
        ("Wake me up at 6:30 tomorrow", "set_alarm"),
        ("Remind me to call mom at 5 PM", "set_reminder"),
        ("I need to remember to buy milk", "set_reminder"),
        ("Play some music", "play_music"),
        ("I want to listen to jazz", "play_music"),
        ("Tell me a joke", "tell_joke"),
        ("What time is it", "get_time"),
        ("What's the current time", "get_time")
    ]
    
    # Create and train the classifier
    intent_classifier = IntentClassifier()
    intent_classifier.train(training_data)
    
    # Test the classifier
    test_texts = [
        "What's the weather going to be like",
        "Set an alarm for tomorrow morning",
        "I need to remember my appointment",
        "Play some rock music",
        "Tell me something funny"
    ]
    
    for text in test_texts:
        intent, confidence = intent_classifier.predict(text)
        print(f"Text: '{text}'")
        print(f"Predicted intent: {intent} (confidence: {confidence:.2f})")
        print()
    
    # Save the model
    intent_classifier.save("intent_classifier.joblib")

Entity Extraction

Entity extraction identifies specific pieces of information in the user's utterance, such as dates, times, locations, and names. For example, in "Set an alarm for 7 AM," "7 AM" is a time entity.

import spacy
import re
from datetime import datetime, timedelta

class EntityExtractor:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        
        # Regular expressions for common entities
        self.time_pattern = re.compile(r'(\d{1,2})(:\d{2})?\s*(am|pm|AM|PM)?')
        self.date_pattern = re.compile(r'(today|tomorrow|yesterday|next week|next month)')
        
    def extract_entities(self, text):
        """
        Extract entities from text.
        
        Parameters:
        - text: Input text
        
        Returns:
        - Dictionary of extracted entities
        """
        entities = {}
        
        # Process with spaCy
        doc = self.nlp(text)
        
        # Extract named entities
        for ent in doc.ents:
            if ent.label_ not in entities:
                entities[ent.label_] = []
            entities[ent.label_].append(ent.text)
        
        # Extract time entities
        time_matches = self.time_pattern.findall(text)
        if time_matches:
            entities['TIME'] = []
            for match in time_matches:
                hour, minute, period = match
                hour = int(hour)
                minute = int(minute[1:]) if minute else 0
                
                # Handle AM/PM
                if period and period.lower() == 'pm' and hour < 12:
                    hour += 12
                elif period and period.lower() == 'am' and hour == 12:
                    hour = 0
                
                time_str = f"{hour:02d}:{minute:02d}"
                entities['TIME'].append(time_str)
        
        # Extract date entities
        date_matches = self.date_pattern.findall(text)
        if date_matches:
            entities['DATE'] = []
            for match in date_matches:
                if match.lower() == 'today':
                    date = datetime.now().strftime('%Y-%m-%d')
                elif match.lower() == 'tomorrow':
                    date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'yesterday':
                    date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'next week':
                    date = (datetime.now() + timedelta(weeks=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'next month':
                    # Simple approximation for next month
                    date = (datetime.now() + timedelta(days=30)).strftime('%Y-%m-%d')
                
                entities['DATE'].append(date)
        
        return entities

# Example usage
if __name__ == "__main__":
    entity_extractor = EntityExtractor()
    
    test_texts = [
        "Set an alarm for 7 AM tomorrow",
        "Remind me to call John at 3:30 PM today",
        "What's the weather like in New York next week",
        "Schedule a meeting with Sarah for 2 PM next month"
    ]
    
    for text in test_texts:
        entities = entity_extractor.extract_entities(text)
        print(f"Text: '{text}'")
        print(f"Extracted entities: {entities}")
        print()

Dialog Management

Dialog management is responsible for maintaining the conversation flow and context. It determines how the voice assistant should respond to user inputs based on the current state of the conversation.

State-Based Dialog Management

A simple approach to dialog management is to use a state machine, where the conversation transitions between different states based on user inputs.

class DialogState:
    def __init__(self, name, handler):
        self.name = name
        self.handler = handler
        self.transitions = {}
    
    def add_transition(self, intent, next_state):
        """Add a transition to another state based on intent"""
        self.transitions[intent] = next_state
    
    def next_state(self, intent):
        """Get the next state based on intent"""
        return self.transitions.get(intent, self)
    
    def handle(self, intent, entities):
        """Handle the current state"""
        return self.handler(intent, entities)

class DialogManager:
    def __init__(self):
        self.states = {}
        self.current_state = None
        self.context = {}
    
    def add_state(self, state):
        """Add a state to the dialog manager"""
        self.states[state.name] = state
        if self.current_state is None:
            self.current_state = state
    
    def process(self, intent, entities):
        """Process user input and return a response"""
        # Update context with new entities
        for entity_type, values in entities.items():
            self.context[entity_type] = values
        
        # Handle current state
        response = self.current_state.handle(intent, self.context)
        
        # Transition to next state
        self.current_state = self.current_state.next_state(intent)
        
        return response

# Example usage
def greeting_handler(intent, context):
    return "Hello! How can I help you today?"

def weather_handler(intent, context):
    location = context.get('GPE', ['your location'])[0]
    date = context.get('DATE', ['today'])[0]
    return f"The weather in {location} for {date} is sunny with a high of 75°F."

def alarm_handler(intent, context):
    time = context.get('TIME', [''])[0]
    date = context.get('DATE', ['today'])[0]
    if time:
        return f"I've set an alarm for {time} on {date}."
    else:
        return "What time would you like to set the alarm for?"

def reminder_handler(intent, context):
    time = context.get('TIME', [''])[0]
    date = context.get('DATE', ['today'])[0]
    if time:
        return f"I'll remind you at {time} on {date}."
    else:
        return "When would you like to be reminded?"

def fallback_handler(intent, context):
    return "I'm not sure how to help with that. Can you try rephrasing?"

# Create dialog states
greeting_state = DialogState("greeting", greeting_handler)
weather_state = DialogState("weather", weather_handler)
alarm_state = DialogState("alarm", alarm_handler)
reminder_state = DialogState("reminder", reminder_handler)
fallback_state = DialogState("fallback", fallback_handler)

# Set up transitions
greeting_state.add_transition("weather", weather_state)
greeting_state.add_transition("set_alarm", alarm_state)
greeting_state.add_transition("set_reminder", reminder_state)

weather_state.add_transition("set_alarm", alarm_state)
weather_state.add_transition("set_reminder", reminder_state)
weather_state.add_transition("weather", weather_state)

alarm_state.add_transition("set_reminder", reminder_state)
alarm_state.add_transition("weather", weather_state)
alarm_state.add_transition("set_alarm", alarm_state)

reminder_state.add_transition("set_alarm", alarm_state)
reminder_state.add_transition("weather", weather_state)
reminder_state.add_transition("set_reminder", reminder_state)

# Create dialog manager
dialog_manager = DialogManager()
dialog_manager.add_state(greeting_state)
dialog_manager.add_state(weather_state)
dialog_manager.add_state(alarm_state)
dialog_manager.add_state(reminder_state)
dialog_manager.add_state(fallback_state)

# Example conversation
print(dialog_manager.process("greeting", {}))
print(dialog_manager.process("weather", {"GPE": ["New York"]}))
print(dialog_manager.process("set_alarm", {"TIME": ["07:00"], "DATE": ["tomorrow"]}))

Frame-Based Dialog Management

Frame-based dialog management uses "frames" or "slots" to track the information needed to complete a task. The system prompts the user for missing information until all required slots are filled.

class DialogFrame:
    def __init__(self, name, slots=None, handler=None):
        self.name = name
        self.slots = slots or {}
        self.handler = handler
        self.required_slots = set()
    
    def add_slot(self, slot_name, prompt, required=False):
        """Add a slot to the frame"""
        self.slots[slot_name] = {
            "value": None,
            "prompt": prompt
        }
        if required:
            self.required_slots.add(slot_name)
    
    def fill_slot(self, slot_name, value):
        """Fill a slot with a value"""
        if slot_name in self.slots:
            self.slots[slot_name]["value"] = value
            return True
        return False
    
    def is_complete(self):
        """Check if all required slots are filled"""
        for slot_name in self.required_slots:
            if self.slots[slot_name]["value"] is None:
                return False
        return True
    
    def get_missing_slot(self):
        """Get the first missing required slot"""
        for slot_name in self.required_slots:
            if self.slots[slot_name]["value"] is None:
                return slot_name, self.slots[slot_name]["prompt"]
        return None, None
    
    def execute(self):
        """Execute the frame handler with the filled slots"""
        if self.handler and self.is_complete():
            slot_values = {name: slot["value"] for name, slot in self.slots.items()}
            return self.handler(slot_values)
        return None

class FrameDialogManager:
    def __init__(self):
        self.frames = {}
        self.active_frame = None
        self.entity_slot_mapping = {}
    
    def add_frame(self, frame):
        """Add a frame to the dialog manager"""
        self.frames[frame.name] = frame
    
    def map_entity_to_slot(self, entity_type, frame_name, slot_name):
        """Map an entity type to a slot in a frame"""
        if entity_type not in self.entity_slot_mapping:
            self.entity_slot_mapping[entity_type] = []
        self.entity_slot_mapping[entity_type].append((frame_name, slot_name))
    
    def activate_frame(self, frame_name):
        """Activate a frame"""
        if frame_name in self.frames:
            self.active_frame = self.frames[frame_name]
            return True
        return False
    
    def process(self, intent, entities):
        """Process user input and return a response"""
        # Activate frame based on intent if no active frame
        if self.active_frame is None or intent != "continue":
            frame_name = intent.replace("get_", "").replace("set_", "")
            if frame_name in self.frames:
                self.activate_frame(frame_name)
            else:
                return "I'm not sure how to help with that."
        
        # Fill slots based on entities
        for entity_type, values in entities.items():
            if entity_type in self.entity_slot_mapping:
                for frame_name, slot_name in self.entity_slot_mapping[entity_type]:
                    if frame_name == self.active_frame.name:
                        self.active_frame.fill_slot(slot_name, values[0])
        
        # Check if frame is complete
        if self.active_frame.is_complete():
            response = self.active_frame.execute()
            self.active_frame = None
            return response
        else:
            # Prompt for missing slot
            slot_name, prompt = self.active_frame.get_missing_slot()
            return prompt

# Example usage
def weather_handler(slots):
    location = slots.get("location", "your location")
    date = slots.get("date", "today")
    return f"The weather in {location} for {date} is sunny with a high of 75°F."

def alarm_handler(slots):
    time = slots.get("time", "")
    date = slots.get("date", "today")
    return f"I've set an alarm for {time} on {date}."

# Create frames
weather_frame = DialogFrame("weather", handler=weather_handler)
weather_frame.add_slot("location", "Which location would you like the weather for?", required=True)
weather_frame.add_slot("date", "Which date would you like the weather for?", required=True)

alarm_frame = DialogFrame("alarm", handler=alarm_handler)
alarm_frame.add_slot("time", "What time would you like to set the alarm for?", required=True)
alarm_frame.add_slot("date", "Which date would you like to set the alarm for?", required=False)

# Create frame dialog manager
frame_dialog_manager = FrameDialogManager()
frame_dialog_manager.add_frame(weather_frame)
frame_dialog_manager.add_frame(alarm_frame)

# Map entities to slots
frame_dialog_manager.map_entity_to_slot("GPE", "weather", "location")
frame_dialog_manager.map_entity_to_slot("DATE", "weather", "date")
frame_dialog_manager.map_entity_to_slot("DATE", "alarm", "date")
frame_dialog_manager.map_entity_to_slot("TIME", "alarm", "time")

# Example conversation
print(frame_dialog_manager.process("weather", {"GPE": ["New York"]}))
print(frame_dialog_manager.process("continue", {"DATE": ["tomorrow"]}))

print(frame_dialog_manager.process("set_alarm", {"TIME": ["07:00"]}))
print(frame_dialog_manager.process("continue", {"DATE": ["tomorrow"]}))

Complete Voice Assistant Implementation

Now let's put everything together to build a complete voice assistant that can listen for commands, understand them, and respond appropriately.

import speech_recognition as sr
import pyttsx3
import spacy
import re
import datetime
import webbrowser
import random
import time
import threading
import json
import os
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import LinearSVC
import numpy as np

class VoiceAssistant:
    def __init__(self, name="Assistant", wake_word="hey assistant"):
        self.name = name
        self.wake_word = wake_word.lower()
        
        # Initialize speech recognition
        self.recognizer = sr.Recognizer()
        self.recognizer.energy_threshold = 4000
        self.recognizer.dynamic_energy_threshold = True
        
        # Initialize text-to-speech
        self.tts_engine = pyttsx3.init()
        self.tts_engine.setProperty('rate', 180)
        self.tts_engine.setProperty('volume', 0.9)
        
        # Get available voices
        voices = self.tts_engine.getProperty('voices')
        if voices:
            # Try to find a female voice
            female_voice = next((voice for voice in voices if 'female' in voice.name.lower()), None)
            if female_voice:
                self.tts_engine.setProperty('voice', female_voice.id)
        
        # Initialize NLU components
        self.nlp = spacy.load("en_core_web_sm")
        self.intent_classifier = IntentClassifier()
        self.entity_extractor = EntityExtractor()
        
        # Initialize dialog manager
        self.dialog_manager = DialogManager()
        self._setup_dialog_manager()
        
        # Running flag
        self.running = False
        self.listening_for_wake_word = False
        
        # Load training data
        self._load_training_data()
    
    def _load_training_data(self):
        """Load and train the intent classifier with training data"""
        training_data = [
            ("what's the weather like", "weather"),
            ("what's the forecast for today", "weather"),
            ("how's the weather", "weather"),
            ("will it rain today", "weather"),
            ("what's the temperature", "weather"),
            
            ("set an alarm", "set_alarm"),
            ("wake me up at", "set_alarm"),
            ("set a timer for", "set_alarm"),
            ("remind me to wake up", "set_alarm"),
            
            ("remind me to", "set_reminder"),
            ("i need to remember to", "set_reminder"),
            ("don't let me forget to", "set_reminder"),
            ("set a reminder for", "set_reminder"),
            
            ("what time is it", "get_time"),
            ("tell me the time", "get_time"),
            ("what's the current time", "get_time"),
            
            ("what day is it", "get_date"),
            ("what's today's date", "get_date"),
            ("tell me the date", "get_date"),
            
            ("play some music", "play_music"),
            ("i want to listen to", "play_music"),
            ("play", "play_music"),
            
            ("open", "open_app"),
            ("launch", "open_app"),
            ("start", "open_app"),
            
            ("search for", "web_search"),
            ("look up", "web_search"),
            ("find information about", "web_search"),
            
            ("tell me a joke", "tell_joke"),
            ("say something funny", "tell_joke"),
            ("make me laugh", "tell_joke"),
            
            ("who are you", "assistant_info"),
            ("what's your name", "assistant_info"),
            ("tell me about yourself", "assistant_info"),
            
            ("thank you", "gratitude"),
            ("thanks", "gratitude"),
            ("that's helpful", "gratitude"),
            
            ("goodbye", "goodbye"),
            ("bye", "goodbye"),
            ("see you later", "goodbye"),
            ("exit", "goodbye"),
            ("stop", "goodbye")
        ]
        
        self.intent_classifier.train(training_data)
    
    def _setup_dialog_manager(self):
        """Set up the dialog manager with states and handlers"""
        # Define handlers
        def greeting_handler(intent, context):
            responses = [
                f"Hello! I'm {self.name}. How can I help you today?",
                f"Hi there! I'm {self.name}. What can I do for you?",
                f"Greetings! I'm {self.name}. How may I assist you?"
            ]
            return random.choice(responses)
        
        def weather_handler(intent, context):
            location = context.get('GPE', ['your location'])[0]
            date = context.get('DATE', ['today'])[0]
            
            # In a real implementation, you would call a weather API here
            weather_conditions = ["sunny", "partly cloudy", "cloudy", "rainy", "snowy"]
            temp_range = (65, 85) if random.choice(weather_conditions) in ["sunny", "partly cloudy"] else (45, 65)
            temp = random.randint(*temp_range)
            
            condition = random.choice(weather_conditions)
            return f"The weather in {location} for {date} is {condition} with a high of {temp}°F."
        
        def alarm_handler(intent, context):
            time_entity = context.get('TIME', [''])[0]
            date_entity = context.get('DATE', ['today'])[0]
            
            if time_entity:
                # In a real implementation, you would set an actual alarm here
                return f"I've set an alarm for {time_entity} on {date_entity}."
            else:
                return "What time would you like to set the alarm for?"
        
        def reminder_handler(intent, context):
            time_entity = context.get('TIME', [''])[0]
            date_entity = context.get('DATE', ['today'])[0]
            
            # Try to extract what to remind about
            reminder_text = ""
            if 'REMINDER' in context:
                reminder_text = f" to {context['REMINDER'][0]}"
            
            if time_entity:
                # In a real implementation, you would set an actual reminder here
                return f"I'll remind you{reminder_text} at {time_entity} on {date_entity}."
            else:
                return f"When would you like to be reminded{reminder_text}?"
        
        def time_handler(intent, context):
            current_time = datetime.datetime.now().strftime("%I:%M %p")
            return f"The current time is {current_time}."
        
        def date_handler(intent, context):
            current_date = datetime.datetime.now().strftime("%A, %B %d, %Y")
            return f"Today is {current_date}."
        
        def music_handler(intent, context):
            genre = context.get('GENRE', [''])[0]
            artist = context.get('PERSON', [''])[0]
            
            if genre:
                return f"Playing {genre} music for you."
            elif artist:
                return f"Playing music by {artist}."
            else:
                return "Playing some music for you."
        
        def app_handler(intent, context):
            app_name = context.get('APP', [''])[0]
            
            if app_name:
                return f"Opening {app_name} for you."
            else:
                return "Which application would you like to open?"
        
        def search_handler(intent, context):
            query = context.get('QUERY', [''])[0]
            
            if query:
                # In a real implementation, you would open a browser with the search query
                return f"Searching the web for {query}."
            else:
                return "What would you like to search for?"
        
        def joke_handler(intent, context):
            jokes = [
                "Why don't scientists trust atoms? Because they make up everything!",
                "Why did the scarecrow win an award? Because he was outstanding in his field!",
                "What do you call a fake noodle? An impasta!",
                "Why couldn't the bicycle stand up by itself? It was two tired!",
                "What do you call a fish with no eyes? Fsh!"
            ]
            return random.choice(jokes)
        
        def assistant_info_handler(intent, context):
            return f"I'm {self.name}, your voice assistant. I can help you with weather, alarms, reminders, and more."
        
        def gratitude_handler(intent, context):
            responses = [
                "You're welcome!",
                "Happy to help!",
                "Anytime!",
                "My pleasure!"
            ]
            return random.choice(responses)
        
        def goodbye_handler(intent, context):
            responses = [
                "Goodbye!",
                "See you later!",
                "Have a great day!",
                "Bye for now!"
            ]
            self.running = False
            return random.choice(responses)
        
        def fallback_handler(intent, context):
            responses = [
                "I'm not sure how to help with that.",
                "I didn't understand. Could you try rephrasing?",
                "I'm still learning and don't know how to respond to that yet.",
                "I'm not sure what you mean. Can you try asking differently?"
            ]
            return random.choice(responses)
        
        # Create states
        greeting_state = DialogState("greeting", greeting_handler)
        weather_state = DialogState("weather", weather_handler)
        alarm_state = DialogState("set_alarm", alarm_handler)
        reminder_state = DialogState("set_reminder", reminder_handler)
        time_state = DialogState("get_time", time_handler)
        date_state = DialogState("get_date", date_handler)
        music_state = DialogState("play_music", music_handler)
        app_state = DialogState("open_app", app_handler)
        search_state = DialogState("web_search", search_handler)
        joke_state = DialogState("tell_joke", joke_handler)
        assistant_info_state = DialogState("assistant_info", assistant_info_handler)
        gratitude_state = DialogState("gratitude", gratitude_handler)
        goodbye_state = DialogState("goodbye", goodbye_handler)
        fallback_state = DialogState("fallback", fallback_handler)
        
        # Add states to dialog manager
        self.dialog_manager.add_state(greeting_state)
        self.dialog_manager.add_state(weather_state)
        self.dialog_manager.add_state(alarm_state)
        self.dialog_manager.add_state(reminder_state)
        self.dialog_manager.add_state(time_state)
        self.dialog_manager.add_state(date_state)
        self.dialog_manager.add_state(music_state)
        self.dialog_manager.add_state(app_state)
        self.dialog_manager.add_state(search_state)
        self.dialog_manager.add_state(joke_state)
        self.dialog_manager.add_state(assistant_info_state)
        self.dialog_manager.add_state(gratitude_state)
        self.dialog_manager.add_state(goodbye_state)
        self.dialog_manager.add_state(fallback_state)
        
        # Set up transitions (simplified - in a real implementation, you would define more transitions)
        for state in self.dialog_manager.states.values():
            for intent in self.dialog_manager.states:
                if intent in self.dialog_manager.states:
                    state.add_transition(intent, self.dialog_manager.states[intent])
    
    def speak(self, text):
        """Speak the given text"""
        print(f"{self.name}: {text}")
        self.tts_engine.say(text)
        self.tts_engine.runAndWait()
    
    def listen(self, timeout=None, phrase_time_limit=None):
        """Listen for a command"""
        with sr.Microphone() as source:
            print("Listening...")
            self.recognizer.adjust_for_ambient_noise(source, duration=0.5)
            try:
                audio = self.recognizer.listen(source, timeout=timeout, phrase_time_limit=phrase_time_limit)
                try:
                    text = self.recognizer.recognize_google(audio).lower()
                    print(f"You said: {text}")
                    return text
                except sr.UnknownValueError:
                    return None
                except sr.RequestError:
                    self.speak("Sorry, I'm having trouble accessing the recognition service.")
                    return None
            except sr.WaitTimeoutError:
                return None
    
    def listen_for_wake_word(self):
        """Listen for the wake word"""
        self.listening_for_wake_word = True
        
        while self.listening_for_wake_word:
            with sr.Microphone() as source:
                print("Listening for wake word...")
                self.recognizer.adjust_for_ambient_noise(source, duration=0.5)
                try:
                    audio = self.recognizer.listen(source, timeout=10, phrase_time_limit=3)
                    try:
                        text = self.recognizer.recognize_google(audio).lower()
                        print(f"Heard: {text}")
                        
                        if self.wake_word in text:
                            print("Wake word detected!")
                            self.speak(f"Yes, I'm here.")
                            self.process_commands()
                    except sr.UnknownValueError:
                        pass
                    except sr.RequestError:
                        print("Could not request results from Google Speech Recognition service")
                except (sr.WaitTimeoutError, Exception) as e:
                    pass
    
    def process_commands(self):
        """Process voice commands"""
        self.running = True
        
        while self.running:
            command = self.listen(timeout=5, phrase_time_limit=5)
            
            if command:
                # Predict intent
                intent, confidence = self.intent_classifier.predict(command)
                print(f"Intent: {intent} (confidence: {confidence:.2f})")
                
                # Extract entities
                entities = self.entity_extractor.extract_entities(command)
                print(f"Entities: {entities}")
                
                # Extract query for web search
                if intent == "web_search" and 'QUERY' not in entities:
                    query = command.replace("search for", "").replace("look up", "").replace("find information about", "").strip()
                    entities['QUERY'] = [query]
                
                # Extract reminder text
                if intent == "set_reminder" and 'REMINDER' not in entities:
                    reminder_text = command.replace("remind me to", "").replace("i need to remember to", "").replace("don't let me forget to", "").replace("set a reminder for", "").strip()
                    entities['REMINDER'] = [reminder_text]
                
                # Process with dialog manager
                response = self.dialog_manager.process(intent, entities)
                
                # Speak response
                self.speak(response)
                
                # If goodbye intent, exit the loop
                if intent == "goodbye":
                    break
            
            time.sleep(0.1)
    
    def run(self):
        """Run the voice assistant"""
        self.speak(f"Hello, I'm {self.name}. Say '{self.wake_word}' to activate me.")
        
        try:
            self.listen_for_wake_word()
        except KeyboardInterrupt:
            self.speak("Goodbye!")
            self.listening_for_wake_word = False

# Intent classifier class
class IntentClassifier:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.vectorizer = TfidfVectorizer(ngram_range=(1, 2), max_features=1000)
        self.classifier = LinearSVC()
        self.intents = []
        
    def train(self, training_data):
        """Train the intent classifier"""
        texts, intents = zip(*training_data)
        processed_texts = [self._preprocess(text) for text in texts]
        X = self.vectorizer.fit_transform(processed_texts)
        self.classifier.fit(X, intents)
        self.intents = list(set(intents))
        
    def predict(self, text):
        """Predict the intent of a text"""
        processed_text = self._preprocess(text)
        X = self.vectorizer.transform([processed_text])
        intent = self.classifier.predict(X)[0]
        decision_values = self.classifier.decision_function(X)
        confidence = np.max(decision_values)
        return intent, confidence
    
    def _preprocess(self, text):
        """Preprocess text by lemmatizing and removing stopwords"""
        doc = self.nlp(text.lower())
        tokens = [token.lemma_ for token in doc if not token.is_stop and not token.is_punct]
        return " ".join(tokens)

# Entity extractor class
class EntityExtractor:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.time_pattern = re.compile(r'(\d{1,2})(:\d{2})?\s*(am|pm|AM|PM)?')
        self.date_pattern = re.compile(r'(today|tomorrow|yesterday|next week|next month)')
        
    def extract_entities(self, text):
        """Extract entities from text"""
        entities = {}
        
        # Process with spaCy
        doc = self.nlp(text)
        
        # Extract named entities
        for ent in doc.ents:
            if ent.label_ not in entities:
                entities[ent.label_] = []
            entities[ent.label_].append(ent.text)
        
        # Extract time entities
        time_matches = self.time_pattern.findall(text)
        if time_matches:
            entities['TIME'] = []
            for match in time_matches:
                hour, minute, period = match
                hour = int(hour)
                minute = int(minute[1:]) if minute else 0
                
                # Handle AM/PM
                if period and period.lower() == 'pm' and hour < 12:
                    hour += 12
                elif period and period.lower() == 'am' and hour == 12:
                    hour = 0
                
                time_str = f"{hour:02d}:{minute:02d}"
                entities['TIME'].append(time_str)
        
        # Extract date entities
        date_matches = self.date_pattern.findall(text)
        if date_matches:
            entities['DATE'] = []
            for match in date_matches:
                if match.lower() == 'today':
                    date = datetime.datetime.now().strftime('%Y-%m-%d')
                elif match.lower() == 'tomorrow':
                    date = (datetime.datetime.now() + datetime.timedelta(days=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'yesterday':
                    date = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'next week':
                    date = (datetime.datetime.now() + datetime.timedelta(weeks=1)).strftime('%Y-%m-%d')
                elif match.lower() == 'next month':
                    date = (datetime.datetime.now() + datetime.timedelta(days=30)).strftime('%Y-%m-%d')
                
                entities['DATE'].append(date)
        
        return entities

# Dialog state class
class DialogState:
    def __init__(self, name, handler):
        self.name = name
        self.handler = handler
        self.transitions = {}
    
    def add_transition(self, intent, next_state):
        """Add a transition to another state based on intent"""
        self.transitions[intent] = next_state
    
    def next_state(self, intent):
        """Get the next state based on intent"""
        return self.transitions.get(intent, self)
    
    def handle(self, intent, entities):
        """Handle the current state"""
        return self.handler(intent, entities)

# Dialog manager class
class DialogManager:
    def __init__(self):
        self.states = {}
        self.current_state = None
        self.context = {}
    
    def add_state(self, state):
        """Add a state to the dialog manager"""
        self.states[state.name] = state
        if self.current_state is None:
            self.current_state = state
    
    def process(self, intent, entities):
        """Process user input and return a response"""
        # Update context with new entities
        for entity_type, values in entities.items():
            self.context[entity_type] = values
        
        # Find the appropriate state for the intent
        if intent in self.states:
            self.current_state = self.states[intent]
        
        # Handle current state
        response = self.current_state.handle(intent, self.context)
        
        # Transition to next state
        self.current_state = self.current_state.next_state(intent)
        
        return response

# Example usage
if __name__ == "__main__":
    assistant = VoiceAssistant(name="Aria", wake_word="hey aria")
    assistant.run()

This implementation includes all the components we've discussed: speech recognition, text-to-speech, intent classification, entity extraction, and dialog management. The voice assistant can:

• Listen for a wake word to activate
• Recognize various intents like weather, alarms, reminders, time, date, music, web search, etc.
• Extract entities like times, dates, locations, and people
• Maintain context across turns in the conversation
• Respond appropriately to user queries

Extending the Voice Assistant

You can extend this voice assistant in several ways:

1. Add More Intents: Expand the training data to recognize more types of user requests
2. Improve Entity Extraction: Add more sophisticated entity extraction for complex queries
3. Integrate External APIs: Connect to weather services, calendar APIs, music streaming services, etc.
4. Add Contextual Understanding: Improve the dialog manager to handle more complex conversations
5. Implement Personalization: Store user preferences and adapt responses accordingly
6. Add Multi-turn Conversations: Handle follow-up questions and references to previous turns
7. Implement Proactive Features: Add reminders, notifications, and other proactive behaviors