Code
# Install necessary libraries
!pip install --quiet PyMuPDF pillow
!pip install --quiet google-cloud-vision
%pip install --upgrade --quiet google-cloud-texttospeech
1 📝 Empowering Blind Students: Handwritten Note to Audio Converter
1.1 Introduction
This Colab notebook provides an AI-powered solution to convert handwritten notes into accessible audio formats. Designed primarily for blind students, it leverages Google Cloud’s Vision AI for Optical Character Recognition (OCR), Google’s Gemini Pro for intelligent text correction and formatting, and Google Cloud’s Text-to-Speech (TTS) for natural-sounding audio synthesis.
The goal is to transform scanned handwritten pages into structured, grammatically correct text and then into segmented audio files, making educational content more accessible.
1.2 How to Use
- Upload Files: Upload your PDF of handwritten notes and your Google Cloud service account JSON key to your Colab environment. You can drag and drop them into the ‘Files’ tab on the left sidebar.
- Get Gemini API Key: Obtain a Gemini API key from Google AI Studio (aistudio.google.com/build/api-key).
- Configure Paths and Keys: Update the
PDF_FILE_PATH,GOOGLE_CLOUD_SERVICE_ACCOUNT_KEY_PATH, andGEMINI_API_KEYvariables in theConfigurationsection below. - Run All Cells: Execute all cells in the notebook by going to
Runtime > Run all. - Download Outputs: Once the process is complete, you will find the generated audio segments (
.mp3files) and the cleaned text file (.txt) in theaudio_segmentsfolder in the Colab ‘Files’ tab. You can download them to your local machine.
Code
import os
import fitz # PyMuPDF for PDF handling
from PIL import Image
import io
import re
import math
from pydub import AudioSegment
from google.colab import userdata
# Google Cloud API imports
from google.cloud import vision
from google.cloud import texttospeech
import google.generativeai as genaiCode
# --- Configuration ---
# IMPORTANT: Update these paths with the actual names of your uploaded files
TITLE='Geography pages 1 to 11'
PDF_FILE_PATH = '<input file path here>'
# Ensure your Google Cloud service account JSON key file is uploaded to your Colab files
# and set this path to its location (e.g., 'service_account.json').
GOOGLE_CLOUD_SERVICE_ACCOUNT_KEY_PATH = '<google cloud service account file name here>'
# Gemini API Key: Store this securely in Colab Secrets (Left Panel -> Key icon) under the name 'GEMINI_API_KEY'
GEMINI_API_KEY = userdata.get('GEMINI_API_KEY')
OUTPUT_IMAGE_FOLDER = "extracted_images"
OUTPUT_AUDIO_FOLDER = "audio_segments"
OUTPUT_TEXT_FILE = "cleaned_text.txt" # Added output text file name
TTS_VOICE_NAME = "en-IN-Chirp3-HD-Aoede" # Set to a Chirp3-HD voice from en-IN
# Explore others for naturalness: https://cloud.google.com/text-to-speech/docs/voices
SPEAKING_RATE = 1.0 # Adjust speaking rate (0.25 to 4.0, 1.0 is default)
AUDIO_SEGMENT_DURATION_MIN = 12 # Desired duration for each audio segment
# Set environment variable for Google Cloud authentication (Vision & TTS)
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = GOOGLE_CLOUD_SERVICE_ACCOUNT_KEY_PATH
# Configure Gemini API
genai.configure(api_key=GEMINI_API_KEY)1.3 Workflow Steps
This notebook orchestrates a multi-step AI workflow:
- PDF to Images: Each page of the input PDF is converted into a high-resolution image.
- Handwritten Text Recognition (OCR): Google Cloud Vision AI processes these images to extract handwritten text, even from challenging inputs.
- Intelligent Text Correction and Formatting: The extracted text, often containing errors from OCR or original handwriting, is sent to Google’s Gemini Pro model. Gemini corrects grammar and spelling, cleans up inconsistencies, and formats the text into a structured Markdown document with headings and lists.
- Text Cleanup for Readability: Further processing converts Markdown headings to bold plain text and bullet points to numbered lists, optimizing the text for both readability and subsequent audio synthesis.
- Text-to-Speech Synthesis: The cleaned and corrected text is then converted into natural-sounding speech using Google Cloud Text-to-Speech. The
Chirp3-HDvoice is used for high-quality output, and bold headings are converted into subtle pauses for better listening comprehension. - Audio Segmentation: The synthesized audio is split into manageable segments (e.g., 12-minute chunks) for easier navigation and listening, especially for longer documents.
Code
class BlindStudentAidAgent:
def __init__(self, pdf_path, image_output_folder, audio_output_folder,
tts_voice, speaking_rate, segment_duration_min):
self.pdf_path = pdf_path
self.image_output_folder = image_output_folder
self.audio_output_folder = audio_output_folder
self.tts_voice = tts_voice
self.speaking_rate = speaking_rate
self.segment_duration_min = segment_duration_min
# Initialize Google Cloud clients
self.vision_client = vision.ImageAnnotatorClient()
self.tts_client = texttospeech.TextToSpeechClient()
self.gemini_model = genai.GenerativeModel('gemini-1.5-flash') # Or 'gemini-1.5-pro' for higher quality/cost
# Create output directories if they don't exist
if not os.path.exists(self.image_output_folder):
os.makedirs(self.image_output_folder)
if not os.path.exists(self.audio_output_folder):
os.makedirs(self.audio_output_folder)
def extract_images_from_pdf(self):
"""
Extracts each page of a PDF as a high-resolution image.
"""
print(f"Extracting images from {self.pdf_path}...")
doc = fitz.open(self.pdf_path)
image_paths = []
for i in range(len(doc)):
page = doc.load_page(i)
# Increase resolution significantly for better OCR on handwritten text
pix = page.get_pixmap(matrix=fitz.Matrix(4, 4)) # Increased matrix for even higher res
image_path = os.path.join(self.image_output_folder, f"page_{i+1}.png")
pix.save(image_path)
image_paths.append(image_path)
doc.close()
print(f"Extracted {len(image_paths)} images to {self.image_output_folder}.")
return image_paths
def recognize_handwritten_text(self, image_path):
"""
Recognizes handwritten text from an image using Google Cloud Vision API.
"""
with io.open(image_path, 'rb') as image_file:
content = image_file.read()
image = vision.Image(content=content)
# Use document_text_detection for dense text like a page of notes
response = self.vision_client.document_text_detection(image=image)
full_text = response.full_text_annotation.text
return full_text
def correct_text_with_gemini(self, text):
"""
Corrects grammar and spelling, cleans up text, and formats with Markdown headings
using Gemini API.
"""
print("Correcting, cleaning, and formatting with Markdown using Gemini API...")
prompt = f"""
The following text was extracted from handwritten notes. It may contain
typographical errors, spelling mistakes, and grammatical issues.
Your task is to:
1. **Correct all spelling and grammar errors.**
2. **Ensure proper sentence structure and punctuation.**
3. **Identify logical sections, topics, and sub-topics within the text.**
4. **Format the corrected text using Markdown headings.**
* Use `#` for main sections/chapters.
* Use `##` for major subheadings.
* Use `###` for minor subheadings.
* Do not use headings beyond `###` unless absolutely necessary for complex structures.
5. **Maintain the original meaning and factual content strictly.**
6. **Do NOT add any new information, summarize, or paraphrase beyond correction and structuring.**
7. **Ensure the output is solely the markdown-formatted text, with no conversational filler.**
Text to correct and format:
{text}
"""
try:
response = self.gemini_model.generate_content(prompt)
corrected_and_formatted_text = response.text.strip()
print("Gemini API correction and Markdown formatting complete.")
return corrected_and_formatted_text
except Exception as e:
print(f"Error during Gemini API call for correction and formatting: {e}")
print("Returning original text for now. Please check your API key, model, and quota.")
return text # Fallback to original text if API call fails
def clean_and_format_text(self, text):
"""
Performs additional cleanup and formats the text for plain text output with bold headings
and numbered lists.
"""
print("Performing final text cleanup and formatting for plain text with bold headings and numbered lists...")
# Replace common variations of newlines with a consistent newline
cleaned_text = text.replace('\r\n', '\n').replace('\r', '\n')
# Remove any leading/trailing whitespace from the entire text
cleaned_text = cleaned_text.strip()
lines = cleaned_text.split('\n')
formatted_lines = []
list_counter = 0
in_list = False
for line in lines:
stripped_line = line.strip()
# Convert Markdown headings to bold plain text
if stripped_line.startswith("### "):
heading_text = stripped_line[4:]
formatted_lines.append(f"**{heading_text}**\n") # Add newline after heading
in_list = False # Reset list state after a heading
elif stripped_line.startswith("## "):
heading_text = stripped_line[3:]
formatted_lines.append(f"**{heading_text}**\n") # Add newline after heading
in_list = False # Reset list state after a heading
elif stripped_line.startswith("# "):
heading_text = stripped_line[2:]
formatted_lines.append(f"**{heading_text}**\n") # Add newline after heading
in_list = False # Reset list state after a heading
# Convert bullet points (*, -, +) to numbered list items
elif stripped_line.startswith(('* ', '- ', '+ ')):
if not in_list:
list_counter = 1 # Start new numbered list
in_list = True
list_item_text = stripped_line[2:] # Remove bullet and space
formatted_lines.append(f"{list_counter}. {list_item_text}")
list_counter += 1
elif stripped_line.startswith(('*. ', '-. ', '+. ')): # Handle bullets followed by a period
if not in_list:
list_counter = 1 # Start new numbered list
in_list = True
list_item_text = stripped_line[3:] # Remove bullet, period and space
formatted_lines.append(f"{list_counter}. {list_item_text}")
list_counter += 1
# Handle numbered lists already present
elif re.match(r'^\d+\.\s', stripped_line):
if not in_list:
in_list = True # Recognize existing numbered list
# Keep the existing numbered list format
formatted_lines.append(stripped_line)
# Reset counter if it's the start of a new numbered list (e.g., starts with 1.)
if stripped_line.startswith('1.'):
list_counter = 2 # Next item should be 2
else:
# Try to infer the next number, or just increment if possible
try:
current_number = int(stripped_line.split('.')[0])
list_counter = current_number + 1
except ValueError:
list_counter += 1 # Fallback to simple increment
else:
# If the line is not a heading or a list item and we were in a list, end the list
if in_list and stripped_line: # Only end list if the line is not empty
in_list = False
list_counter = 0 # Reset counter
formatted_lines.append("") # Add an empty line to separate list from next paragraph
formatted_lines.append(line)
cleaned_text = "\n".join(formatted_lines)
# Ensure no more than two consecutive newlines (for paragraph breaks)
cleaned_text = re.sub(r'\n\n+', '\n\n', cleaned_text)
print("Text cleanup and formatting complete.")
return cleaned_text
def split_text_into_chunks(self, text, max_bytes):
"""
Splits text into chunks that are under the maximum byte limit.
Splits by sentences or paragraphs to maintain readability.
"""
print(f"Splitting text into chunks under {max_bytes} bytes...")
chunks = []
current_chunk = ""
# Split by sentences (using common sentence-ending punctuation) or paragraphs (double newline)
# This regex split attempts to keep the delimiter with the sentence/paragraph
sentences = re.split(r'([.!?]\s+|\n\n)', text)
for sentence in sentences:
# Check if adding the next sentence/paragraph exceeds the limit
# Encode the potential next chunk to check its byte size
if len((current_chunk + sentence).encode('utf-8')) > max_bytes and current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence.strip()
else:
current_chunk += sentence
# Add the last chunk if it's not empty
if current_chunk:
chunks.append(current_chunk.strip())
print(f"Text split into {len(chunks)} chunks.")
return chunks
def synthesize_text_to_audio(self, text, output_filename, max_retries=3):
"""
Synthesizes text to an audio file using Google Cloud Text-to-Speech with retry logic.
Replaces bold headings with "[pause]".
"""
print(f"Synthesizing audio to {output_filename}...")
# Replace bold headings with "[pause]" as per documentation for Chirp3-HD
# We will use "[pause]" followed by the heading text
text_with_pauses = re.sub(r'\*\*(.*?)\*\*', r'[pause] \1', text)
# Use the 'markup' field for input with pause tags
synthesis_input = texttospeech.SynthesisInput(markup=text_with_pauses)
voice = texttospeech.VoiceSelectionParams(
language_code="en-IN",
name=self.tts_voice,
ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL
)
audio_config = texttospeech.AudioConfig(
audio_encoding=texttospeech.AudioEncoding.MP3,
speaking_rate=self.speaking_rate
)
for attempt in range(max_retries):
try:
response = self.tts_client.synthesize_speech(
input=synthesis_input, voice=voice, audio_config=audio_config
)
with open(output_filename, "wb") as out:
out.write(response.audio_content)
print(f'Audio content written to file "{output_filename}"')
return output_filename
except Exception as e:
print(f"Attempt {attempt + 1} failed during Text-to-Speech synthesis: {e}")
if attempt < max_retries - 1:
print(f"Retrying in 2 seconds...")
import time
time.sleep(2)
else:
print(f"Max retries ({max_retries}) reached. Returning None.")
return None
def concatenate_audio_files(self, audio_files, output_filename):
"""
Concatenates a list of audio files into a single audio file.
Only concatenates if there are files to combine.
"""
print(f"Concatenating {len(audio_files)} audio files into {output_filename}...")
if not audio_files:
print("No audio files to concatenate.")
return None
combined = AudioSegment.empty()
for audio_file in audio_files:
try:
segment = AudioSegment.from_file(audio_file)
combined += segment
except Exception as e:
print(f"Error processing audio file {audio_file}: {e}")
# Optionally, decide how to handle corrupted files (skip, log, etc.)
if not combined.empty():
combined.export(output_filename, format="mp3")
print(f"Concatenation complete. Saved to {output_filename}.")
return output_filename
else:
print("Combined audio is empty after processing files.")
return None
def split_audio_by_duration(self, audio_path):
"""
Splits an audio file into segments of a specified duration.
"""
print(f"Splitting audio file {audio_path} into {self.segment_duration_min}-minute segments...")
audio = AudioSegment.from_file(audio_path)
total_length_ms = len(audio)
segment_length_ms = self.segment_duration_min * 60 * 1000
num_segments = math.ceil(total_length_ms / segment_length_ms)
segment_paths = []
for i in range(num_segments):
start_ms = i * segment_length_ms
end_ms = min((i + 1) * segment_length_ms, total_length_ms)
segment = audio[start_ms:end_ms]
segment_filename = os.path.join(self.audio_output_folder, f"notes_part_{i+1}.mp3")
segment.export(segment_filename, format="mp3")
segment_paths.append(segment_filename)
print(f"Exported {segment_filename}")
print("Audio splitting complete.")
return segment_paths
def run_workflow(self):
"""
Orchestrates the entire agentic workflow.
"""
print("--- Starting AI Agent Workflow for Blind Students ---")
# Step 1: Extract images from PDF
image_files = self.extract_images_from_pdf()
if not image_files:
print("No images extracted. Check PDF file path and content.")
return
# Step 2: Recognize handwritten text from images
all_extracted_text = []
for i, img_path in enumerate(image_files):
print(f"Recognizing text from {img_path} ({i+1}/{len(image_files)})...")
try:
text = self.recognize_handwritten_text(img_path)
all_extracted_text.append(text)
except Exception as e:
print(f"Error recognizing text from {img_path}: {e}")
all_extracted_text.append(f"[Error extracting text from {os.path.basename(img_path)}]") # Placeholder
full_document_text = "\n".join(all_extracted_text)
print("\n--- Raw Extracted Text (First 500 characters) ---")
print(full_document_text[:500] + "..." if len(full_document_text) > 500 else full_document_text)
# Step 3: Correct grammar and spelling using Gemini
corrected_text = self.correct_text_with_gemini(full_document_text)
print("\n--- Corrected Text (First 500 characters) ---")
print(corrected_text[:500] + "..." if len(corrected_text) > 500 else corrected_text)
# Step 4: Cleanup and format text
final_cleaned_text = self.clean_and_format_text(corrected_text)
print("\n--- Final Cleaned & Formatted Text (First 500 characters) ---")
print(final_cleaned_text[:500] + "..." if len(final_cleaned_text) > 500 else final_cleaned_text)
# Add the title as a heading to the cleaned text
text_with_title = f"**{TITLE}**\n\n{final_cleaned_text}"
# Save the cleaned text with title to a file
text_output_path = os.path.join(self.audio_output_folder, OUTPUT_TEXT_FILE)
with open(text_output_path, "w") as f:
f.write(text_with_title)
print(f"\nCleaned text with title saved to {text_output_path}")
# Step 5: Split text into chunks and synthesize audio for each chunk
text_chunks = self.split_text_into_chunks(text_with_title, 4900) # Use a buffer under 5000 bytes
temp_audio_files = []
for i, chunk in enumerate(text_chunks):
temp_audio_filename = os.path.join(self.audio_output_folder, f"temp_chunk_{i+1}.mp3")
# Call the updated synthesize_text_to_audio method
synthesized_file = self.synthesize_text_to_audio(chunk, temp_audio_filename)
if synthesized_file:
temp_audio_files.append(synthesized_file)
else:
print(f"Skipping concatenation for chunk {i+1} due to synthesis error after multiple retries.")
# Step 6: Concatenate temporary audio files
full_audio_filename = os.path.join(self.audio_output_folder, "full_notes_audio.mp3")
concatenated_audio_path = self.concatenate_audio_files(temp_audio_files, full_audio_filename)
# Step 7: Split audio into segments
segment_paths = []
if concatenated_audio_path:
segment_paths = self.split_audio_by_duration(concatenated_audio_path)
# Clean up temporary audio files
print("Cleaning up temporary audio files...")
for temp_file in temp_audio_files:
try:
os.remove(temp_file)
print(f"Removed {temp_file}")
except OSError as e:
print(f"Error removing temporary file {temp_file}: {e}")
print("Temporary file cleanup complete.")
print("\n--- Workflow complete! ---")
print(f"Audio segments are saved in the '{self.audio_output_folder}' folder.")
print(f"Cleaned text is saved to '{text_output_path}'.")
print("You can download them from the Colab Files sidebar.")
# Return segment paths if successful, otherwise None or handle error appropriately
return segment_paths
# --- Main Execution Block ---
if __name__ == "__main__":
print("Please ensure your PDF file (e.g., 'handwritten_notes.pdf'),")
print("your Google Cloud service account JSON key (e.g., 'service_account.json'),")
print("and your Gemini API Key are uploaded/set up in Colab.")
print("\nRemember to replace 'your_handwritten_notes.pdf', 'your-service-account-key.json',")
print("and 'YOUR_GEMINI_API_KEY' with your actual file names and API key.")
print("If you get authentication errors, double-check your API keys and service account roles.")
# Create dummy files for demonstration if not uploaded
if not os.path.exists(PDF_FILE_PATH):
print(f"Warning: {PDF_FILE_PATH} not found. Creating a dummy PDF for demonstration.")
with fitz.open() as doc:
page = doc.new_page()
page.insert_text((72, 72), "This is a sample handwritten note. It has some mispelled words.")
page.insert_text((72, 100), "This sentence has a grammer error. And also some extra spaces.")
doc.save(PDF_FILE_PATH)
# if GEMINI_API_KEY == 'YOUR_GEMINI_API_KEY':
# print("\nERROR: Please replace 'YOUR_GEMINI_API_KEY' with your actual Gemini API key.")
# print("You can get one from Google AI Studio (aistudio.google.com/build/api-key).")
# # Note: Removed exit() to allow workflow to proceed with dummy files if API key is not set,
# # but Gemini correction will fail gracefully. For full functionality, key is needed.
# Instantiate and run the agent
agent = BlindStudentAidAgent(
pdf_path=PDF_FILE_PATH,
image_output_folder=OUTPUT_IMAGE_FOLDER,
audio_output_folder=OUTPUT_AUDIO_FOLDER,
tts_voice=TTS_VOICE_NAME,
speaking_rate=SPEAKING_RATE,
segment_duration_min=AUDIO_SEGMENT_DURATION_MIN
)
agent.run_workflow()