{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import numpy as np\n", "import pyaudio\n", "import wave\n", "import os\n", "import time\n", "import ipywidgets as widgets\n", "from IPython.display import display, clear_output, Audio\n", "import matplotlib.pyplot as plt\n", "from datetime import datetime\n", "import threading\n", "import glob\n", "\n", "# Audio recording parameters\n", "FORMAT = pyaudio.paInt16\n", "CHANNELS = 1\n", "RATE = 16000 # Sample rate (Hz)\n", "CHUNK = 1024 # Buffer size\n", "RECORD_SECONDS = 1 # Each chunk is 1 second\n", "\n", "\n", "# Create directory for storing audio chunks\n", "def create_directory():\n", " timestamp = datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n", " directory = \"audio_input\"\n", " os.makedirs(directory, exist_ok=True)\n", " return directory\n", "\n", "\n", "# Function to save audio chunk\n", "def save_audio_chunk(frames, directory, chunk_number):\n", " filename = os.path.join(directory, f\"chunk_{chunk_number:04d}.wav\")\n", " wf = wave.open(filename, \"wb\")\n", " wf.setnchannels(CHANNELS)\n", " wf.setsampwidth(2) # 2 bytes for paInt16\n", " wf.setframerate(RATE)\n", " wf.writeframes(b\"\".join(frames))\n", " wf.close()\n", " return filename\n", "\n", "\n", "# Function to combine all audio chunks into one file\n", "def combine_audio_chunks(directory):\n", " output_filename = os.path.join(directory, \"full_recording.wav\")\n", " chunk_files = sorted(glob.glob(os.path.join(directory, \"chunk_*.wav\")))\n", "\n", " if not chunk_files:\n", " return None\n", "\n", " # Read the first file to get parameters\n", " with wave.open(chunk_files[0], \"rb\") as wf:\n", " params = wf.getparams()\n", "\n", " # Create output file with same parameters\n", " with wave.open(output_filename, \"wb\") as outfile:\n", " outfile.setparams(params)\n", "\n", " # Write each chunk to the output file\n", " for chunk_file in chunk_files:\n", " with wave.open(chunk_file, \"rb\") as infile:\n", " outfile.writeframes(infile.readframes(infile.getnframes()))\n", "\n", " return output_filename\n", "\n", "\n", "# Audio recording class\n", "class AudioRecorder:\n", " def __init__(self):\n", " self.is_recording = False\n", " self.audio = pyaudio.PyAudio()\n", " self.frames = []\n", " self.chunk_count = 0\n", " self.directory = None\n", " self.stream = None\n", " self.thread = None\n", " self.all_frames = [] # Store all frames for full recording\n", "\n", " def start_recording(self):\n", " if self.is_recording:\n", " return\n", "\n", " self.is_recording = True\n", " self.directory = create_directory()\n", " self.chunk_count = 0\n", " self.frames = []\n", " self.all_frames = [] # Reset all frames\n", "\n", " # Open audio stream\n", " self.stream = self.audio.open(\n", " format=FORMAT,\n", " channels=CHANNELS,\n", " rate=RATE,\n", " input=True,\n", " frames_per_buffer=CHUNK,\n", " )\n", "\n", " # Start recording thread\n", " self.thread = threading.Thread(target=self._record_thread)\n", " self.thread.daemon = True\n", " self.thread.start()\n", "\n", " def _record_thread(self):\n", " try:\n", " while self.is_recording:\n", " # Record for 1 second\n", " self.frames = []\n", " for _ in range(0, int(RATE / CHUNK * RECORD_SECONDS)):\n", " if not self.is_recording:\n", " break\n", " if self.stream is not None:\n", " data = self.stream.read(CHUNK, exception_on_overflow=False)\n", " self.frames.append(data)\n", " self.all_frames.append(data) # Add to full recording\n", "\n", " if self.frames and self.is_recording:\n", " # Save the 1-second chunk\n", " filename = save_audio_chunk(\n", " self.frames, self.directory, self.chunk_count\n", " )\n", " self.chunk_count += 1\n", "\n", " # Update UI with latest chunk info\n", " with output:\n", " clear_output(wait=True)\n", " print(\n", " f\"Recording... Saved chunk {self.chunk_count} to {filename}\"\n", " )\n", "\n", " # Plot waveform of the latest chunk\n", " audio_data = np.frombuffer(\n", " b\"\".join(self.frames), dtype=np.int16\n", " )\n", " plt.figure(figsize=(10, 2))\n", " plt.plot(audio_data)\n", " plt.title(f\"Waveform of chunk {self.chunk_count}\")\n", " plt.xlabel(\"Sample\")\n", " plt.ylabel(\"Amplitude\")\n", " plt.tight_layout()\n", " plt.show()\n", " except Exception as e:\n", " print(f\"Error in recording thread: {e}\")\n", "\n", " def stop_recording(self):\n", " self.is_recording = False\n", " if self.stream:\n", " self.stream.stop_stream()\n", " self.stream.close()\n", " self.stream = None\n", "\n", " if self.thread:\n", " self.thread.join(timeout=2.0)\n", " self.thread = None\n", "\n", " with output:\n", " clear_output(wait=True)\n", " print(\n", " f\"Recording stopped. {self.chunk_count} chunks saved to {self.directory}\"\n", " )\n", "\n", " # Combine all chunks into one file\n", " if self.all_frames:\n", " full_audio_file = combine_audio_chunks(self.directory)\n", " if full_audio_file:\n", " print(f\"Full recording saved to {full_audio_file}\")\n", "\n", " # Display full waveform\n", " full_audio_data = np.frombuffer(\n", " b\"\".join(self.all_frames), dtype=np.int16\n", " )\n", " plt.figure(figsize=(12, 3))\n", " plt.plot(full_audio_data)\n", " plt.title(\"Full Recording Waveform\")\n", " plt.xlabel(\"Sample\")\n", " plt.ylabel(\"Amplitude\")\n", " plt.tight_layout()\n", " plt.show()\n", "\n", " # Play the full recording (input audio)\n", " print(\"Input Audio:\")\n", " display(Audio(full_audio_file))\n", "\n", " # Check if there's a file in audio_output directory and play it\n", " if os.path.exists(\"audio_output\"):\n", " audio_files = glob.glob(os.path.join(\"audio_output\", \"*.wav\"))\n", " if audio_files:\n", " latest_file = max(audio_files, key=os.path.getctime)\n", " print(f\"\\nOutput Audio (processed):\")\n", " print(f\"File: {latest_file}\")\n", " display(Audio(latest_file))\n", " else:\n", " print(\"\\nNo processed output audio files found.\")\n", " else:\n", " print(\"\\nNo audio_output directory found.\")\n", "\n", " def close(self):\n", " self.stop_recording()\n", " self.audio.terminate()\n", "\n", "\n", "# Create UI components\n", "output = widgets.Output()\n", "recorder = AudioRecorder()\n", "\n", "start_button = widgets.Button(\n", " description=\"Start Recording\", button_style=\"success\", icon=\"microphone\"\n", ")\n", "\n", "stop_button = widgets.Button(\n", " description=\"Stop Recording\", button_style=\"danger\", icon=\"stop\"\n", ")\n", "\n", "\n", "# Button callbacks\n", "def on_start_button_clicked(b):\n", " recorder.start_recording()\n", "\n", "\n", "def on_stop_button_clicked(b):\n", " recorder.stop_recording()\n", "\n", "\n", "start_button.on_click(on_start_button_clicked)\n", "stop_button.on_click(on_stop_button_clicked)\n", "\n", "# Display UI\n", "display(widgets.HBox([start_button, stop_button]))\n", "display(output)\n", "\n", "# Ensure cleanup when notebook is closed\n", "import atexit\n", "\n", "atexit.register(lambda: recorder.close())\n", "\n", "with output:\n", " print(\"Audio recorder ready. Click 'Start Recording' to begin.\")\n", " print(f\"Each chunk will be 1 second ({RATE} samples at {RATE}Hz)\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "suno_env_dev", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.15" } }, "nbformat": 4, "nbformat_minor": 2 }