Chapter 8: Audio Recording and Playback Copy

ssh ubuntu@192.168.0.162

sudo apt-get update
sudo apt-get install -y \
  python3-venv \
  python3-dev \
  portaudio19-dev \
  libsdl2-dev \
  libsdl2-mixer-dev

# 1. Back up the original configuration
sudo cp /etc/apt/sources.list /etc/apt/sources.list.bak
# 2. Switch to a domestic source (using Tsinghua as an example)
sudo tee /etc/apt/sources.list > /dev/null <<'EOF'
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-updates main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-backports main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-security main restricted universe multiverse
EOF
# Note: If using an ARM architecture, use the following source paths
sudo tee /etc/apt/sources.list > /dev/null <<'EOF'
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-updates main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-backports main restricted universe multiverse
deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu-ports/ jammy-security main restricted universe multiverse
EOF
# 3. Update the index
sudo apt-get update
# 4. Install the dependency packages
sudo apt-get install -y \
python3-venv \
python3-dev \
portaudio19-dev \
libsdl2-dev \
libsdl2-mixer-dev

cd /home/ubuntu/audio_example
# Create a virtual environment named .venv in this directory
python3 -m venv .venv

# Activate (run this every time you open a new terminal)
source .venv/bin/activate
# Confirm that pip from the venv is being used
which pip
# Should be similar to: .../audio_example/.venv/bin/python
# Upgrade pip (optional but recommended)
pip install -U pip setuptools wheel
# Install the packages required by the scripts in this directory
pip install numpy pyaudio pygame
# Optional: Users in China can speed up installation by temporarily specifying a mirror source
pip install numpy pyaudio pygame -i https://pypi.tuna.tsinghua.edu.cn/simple
# If using the workspace repository, you can install via requirements.txt
pip install -r requirements.txt
# Optional: Users in China can speed up installation by temporarily specifying a mirror source
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

arecord -f cd test.wav

aplay test.wav

# Set the system volume to 50%
pactl set-sink-volume @DEFAULT_SINK@ 50%

import argparse
import re
import subprocess
import sys

def get_system_volume() -> int | None:
    """Get the current default volume percentage of the system using pactl"""
    try:
        # Run the pactl command to query the default output device (Sink) status
        result = subprocess.run(
            ["pactl", "get-sink-volume", "@DEFAULT_SINK@"],
            capture_output=True,
            text=True,
            check=True,
        )
        # Example pactl output: "Volume: front-left: 32768 /  50% / -18.06 dB, ..."
        # Use a regex to extract the first percentage number
        match = re.search(r"(\d+)%", result.stdout)
        if match:
            volume = int(match.group(1))
            return volume
        else:
            print("⚠️ Unable to parse the system volume output format")
            return None
    except subprocess.CalledProcessError as e:
        print(f"❌ Failed to get volume: {e}")
        return None
    except FileNotFoundError:
        print("❌ pactl command not found. Please check the PulseAudio / PipeWire tools")
        return None

def set_system_volume(volume: int) -> None:
    """Set the system master volume using pactl (0 - 100)"""
    try:
        subprocess.run(
            ["pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{volume}%"],
            check=True,
        )
        print(f"🔊 System master volume successfully set to: {volume}%")
    except subprocess.CalledProcessError as e:
        print(f"❌ Failed to set volume: {e}")
    except FileNotFoundError:
        print("❌ pactl command not found. Please check the PulseAudio / PipeWire tools")

def main():
    parser = argparse.ArgumentParser(
        description="Ubuntu system master volume control tool (supports querying and dynamic setting)"
    )
    # 1. Positional argument for setting the volume
    parser.add_argument(
        "volume",
        nargs="?",  # '?' means the argument is optional
        type=int,
        help="Target volume (integer from 0-100). If not provided and -g is not specified, the current volume is printed by default",
    )
    # 2. Flag option for querying the volume
    parser.add_argument(
        "-g",
        "--get",
        action="store_true",
        help="Only get the current system master volume",
    )
    args = parser.parse_args()
    # Logic: respond to --get first, or default to querying the volume when no volume argument is provided
    if args.get or args.volume is None:
        current_vol = get_system_volume()
        if current_vol is not None:
            print(f"🔈 Current system master volume: {current_vol}%")
        return
    # If a volume argument is provided, validate the range and set the volume
    if not (0 <= args.volume <= 100):
        print(f"⚠️ Error: Volume must be between 0 and 100! Current input: {args.volume}")
        sys.exit(1)
    set_system_volume(args.volume)

if __name__ == "__main__":
    main()

# Query the system volume
python3 system_volume_example.py
# Set the system volume; valid range is 0-100. The following sets it to 50
python3 system_volume_example.py 50

"""Complete audio capture + recording saving example
Asynchronous microphone recording based on PyAudio; a wav audio file is automatically generated when recording ends
"""
import argparse
import queue
import numpy as np
import pyaudio
import wave
import time

class Recorder:
    def __init__(self, config):
        """Initialize audio recorder
        Args:
            config (dict): configuration dictionary
        """
        # Save the configuration parameters
        self.config = config
        # Thread-safe queue: caches audio frames captured in the background
        self.recorder_queue = queue.Queue()
        # Recording device index; select a specific microphone
        self.recorder_device_index = config.get("recorder_device_index", None)
        # Audio sampling rate, default 16000 Hz (commonly used for speech recognition)
        self.recorder_rate = config.get("recorder_audio_rate", 16000)
        # Number of audio frames per acquisition
        self.recorder_chunk_size = config.get("recorder_chunk_size", 1024)
        # Total number of hardware channels of the microphone
        self.recorder_channels = config.get("recorder_channels", 1)
        # List of channel indices to extract
        self.recorder_pickup_channels = config["recorder_pickup_channels"]
        # Playback state flag: set to True while playing audio to mute the mic and prevent feedback/howling
        self.is_playing = False
        # Initialize the PyAudio instance
        self.p = pyaudio.PyAudio()
        # Open the microphone input stream and register the asynchronous callback for acquisition
        self.stream = self.p.open(
            format=pyaudio.paInt16,
            channels=self.recorder_channels,
            rate=self.recorder_rate,
            input=True,
            input_device_index=self.recorder_device_index,
            frames_per_buffer=self.recorder_chunk_size,
            stream_callback=self.stream_callback,
        )

    def stream_callback(self, in_data, frame_count, time_info, status):
        """Background callback function automatically triggered when the audio buffer is full (runs in an independent thread)"""
        # If audio is playing, skip recording to prevent howling/echo
        if self.is_playing:
            return (None, pyaudio.paContinue)
        try:
            # Convert the raw binary byte stream into a 16-bit integer array
            audio_data = np.frombuffer(in_data, dtype=np.int16)
            # Reshape: (frames per acquisition, total channels)
            audio_reshaped = audio_data.reshape(
                self.recorder_chunk_size, self.recorder_channels
            )
            # Normalize: int16 range [-32768, 32767] to float [-1.0, 1.0]
            audio_float = audio_reshaped.astype(np.float32) / 32768.0
            # Extract the channels specified in the config
            audio_picked = audio_float[:, self.recorder_pickup_channels]
            # Non-blocking write to the queue; drop the current frame if the queue is full to avoid blocking the sound card
            self.recorder_queue.put(audio_picked, block=False)
        except queue.Full:
            # Queue full: drop the frame directly without raising an exception
            pass
        # Tell the sound card to continue capturing the next audio segment
        return (None, pyaudio.paContinue)

    def get(self):
        """Called from the main thread: get one frame of floating-point audio data from the queue"""
        try:
            # Wait at most 0.5 seconds to get audio
            data = self.recorder_queue.get(timeout=0.5)
            return data
        except queue.Empty:
            # No audio: return a zero-filled (silence) array as a fallback
            return np.zeros(self.recorder_chunk_size, dtype=np.float32)

    def clear_queue(self):
        """Clear all historically cached audio in the queue"""
        while not self.recorder_queue.empty():
            try:
                self.recorder_queue.get_nowait()
            except queue.Empty:
                break

    def stop_recording(self):
        """Stop the audio capture stream"""
        self.stream.stop_stream()
        self.stream.close()

    def close(self):
        """Release all audio hardware resources"""
        self.stop_recording()
        self.p.terminate()


def save_wav(audio_frames, sample_rate, channels, save_path):
    """
    Save the captured floating-point audio array as a wav file
    :param audio_frames: list of all audio frames
    :param sample_rate: sampling rate
    :param channels: number of channels to save
    :param save_path: audio save path
    """
    # Concatenate all audio segments
    all_audio = np.concatenate(audio_frames, axis=0)
    # Restore the floating-point data to the original 16-bit integer
    audio_int16 = (all_audio * 32768).astype(np.int16)
    # Create the wav file writer object
    wf = wave.open(save_path, 'wb')
    wf.setnchannels(channels)
    wf.setsampwidth(2)  # 16-bit audio fixedly occupies 2 bytes
    wf.setframerate(sample_rate)
    # Write the binary audio data
    wf.writeframes(audio_int16.tobytes())
    wf.close()
    print(f"✅ Recording saved successfully, file path: {save_path}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Record microphone audio and save it as wav")
    parser.add_argument(
        "-o", "--output",
        default="./my_record.wav",
        help="Recording save file path (default: ./my_record.wav)",
    )
    parser.add_argument(
        "-d", "--duration",
        type=float,
        default=5,
        help="Recording duration in seconds (default: 5)",
    )
    args = parser.parse_args()
    # ===================== Recording configuration =====================
    recorder_config = {
        "recorder_device_index": None,        # Microphone device index; None uses the system default microphone
        "recorder_audio_rate": 16000,          # Sampling rate 16000 Hz
        "recorder_chunk_size": 1024,           # Number of frames per acquisition
        "recorder_channels": 1,                # Total hardware channels: mono microphone
        "recorder_pickup_channels": [0]        # Select channel 0 for recording
    }
    record_seconds = args.duration
    save_file_path = args.output
    # 1. Initialize the recorder
    recorder = Recorder(recorder_config)
    print(f"🎙️ Recording started, will record for {record_seconds} seconds...")
    # Cache all captured audio frames
    audio_buffer = []
    start_time = time.time()
    # 2. Loop to capture audio
    while time.time() - start_time < record_seconds:
        frame_data = recorder.get()
        audio_buffer.append(frame_data)
    # 3. Recording finished, release hardware resources
    recorder.close()
    print("Recording finished, saving audio...")
    # 4. Save the audio to a local wav file
    save_wav(
        audio_frames=audio_buffer,
        sample_rate=recorder_config["recorder_audio_rate"],
        channels=len(recorder_config["recorder_pickup_channels"]),
        save_path=save_file_path
    )

# -o specifies the output recording file name; -d specifies the recording duration in seconds
python3 record_example.py -o my_record.wav -d 10

import argparse
from pygame import mixer

def main():
    parser = argparse.ArgumentParser(description="Play a local audio file")
    parser.add_argument("audio_file", help="Path to the audio file to play")
    args = parser.parse_args()
    audio_file = args.audio_file
    # 1. Initialize the mixer module
    mixer.init()
    try:
        # 2. Load the audio file
        mixer.music.load(audio_file)
        print(f"Audio file loaded successfully: {audio_file}")
    except Exception as e:
        print(f"Failed to load audio. Please check the file path: {e}")
        return
    # Default volume (range 0.0 to 1.0)
    current_volume = 0.5
    mixer.music.set_volume(current_volume)
    # 3. Start playback
    mixer.music.play()
    print("\n--- Audio playback control commands ---")
    print(" [p] Pause")
    print(" [r] Resume playback")
    print(" [+] Increase volume (+10%)")
    print(" [-] Decrease volume (-10%)")
    print(" [s] Stop playback")
    print(" [q] Quit the program")
    # Interactive control loop
    while True:
        cmd = input("\nEnter a command: ").strip().lower()
        if cmd == "p":
            mixer.music.pause()
            print("⏸️  Paused")
        elif cmd == "r":
            mixer.music.unpause()
            print("▶️  Resumed playback")
        elif cmd == "+":
            current_volume = min(1.0, current_volume + 0.1)
            mixer.music.set_volume(current_volume)
            print(f"🔊 Volume increased: {int(current_volume * 100)}%")
        elif cmd == "-":
            current_volume = max(0.0, current_volume - 0.1)
            mixer.music.set_volume(current_volume)
            print(f"🔉 Volume decreased: {int(current_volume * 100)}%")
        elif cmd == "s":
            mixer.music.stop()
            print("⏹️  Stopped playback")
        elif cmd == "q":
            mixer.music.stop()
            mixer.quit()
            print("👋 Exited the program")
            break
        else:
            print("⚠️ Unknown command. Please enter p, r, +, -, s or q")

if __name__ == "__main__":
    main()

python3 play_example.py mp3_example.mp3