Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import os
- import json
- from transformers import pipeline
- import datetime
- import signal
- from moviepy.editor import AudioFileClip
- chunk_length = 30
- stride_length = (5, 5)
- kwargs = {"language": "english"}
- model = "whisper-large-v3-turbo"
- print("Initializing pipeline...")
- pipe = pipeline("automatic-speech-recognition", model="C:\\Git\\" + model, chunk_length_s=chunk_length, stride_length_s=stride_length, return_timestamps=True, generate_kwargs=kwargs)
- # The original full text can roughly be recovered by doing "".join(chunk["text"] for chunk in output["chunks"]).
- print("Pipeline initialized.")
- file_extensions = tuple([".wav", ".mp3", ".wma", ".mp4"])
- def get_json_file_path(audio_file_path):
- file_extension = os.path.splitext(audio_file_path)[1]
- if file_extension in file_extensions:
- return audio_file_path.replace(file_extension, ".json")
- def signal_handler(sig, frame):
- global running
- if (running):
- print("Interrupt registered. Finishing current task.")
- running = False
- else:
- print("Interrupt interrupted, will continue processing.")
- running = True
- def seconds_to_hms(seconds):
- hours = (seconds // 3600)
- minutes = (seconds // 60) % 60
- seconds = seconds % 60
- return f"{hours}:{minutes}:{seconds}"
- def process_audio_file(audio_file_path):
- """Processes an audio file and saves the result as a JSON file."""
- try:
- # Check if the JSON file already exists
- json_file_path = get_json_file_path(audio_file_path)
- if (json_file_path is None) or os.path.exists(json_file_path):
- return
- audio_clip = AudioFileClip(audio_file_path)
- duration_seconds = int(audio_clip.duration)
- print(f"Processing {audio_file_path}. Duration: "+seconds_to_hms(duration_seconds))
- start_time = datetime.datetime.now()
- print("Started at time: " + start_time.strftime("%I:%M:%S %p"))
- # Process the audio file
- result = pipe(audio_file_path)
- end_time = datetime.datetime.now()
- time_taken = end_time-start_time
- print("Elapsed time: " + seconds_to_hms(time_taken.seconds))
- result["model"] = model
- result["parameters"] = "chunk_length_s=30, stride_length_s=(5, 5)"
- result["source_path"] = audio_file_path
- directory_path = os.path.dirname(audio_file_path)
- last_directory = os.path.basename(directory_path)
- try:
- date = datetime.datetime.fromisoformat(last_directory)
- result["date"] = last_directory
- except ValueError:
- pass
- # Save the result as a JSON file
- with open(json_file_path, "w") as f:
- json.dump(result, f, indent=4)
- print(f"Saved result to {json_file_path}")
- except Exception as e:
- print(f"Error processing {audio_file_path}: {e}")
- def process_directory(directory_path):
- """Processes all audio files in a directory and its subdirectories."""
- global running
- if not running:
- return
- for root, dirs, files in os.walk(directory_path):
- for file in files:
- file = file.lower()
- if file.endswith(file_extensions):
- audio_file_path = os.path.join(root, file)
- process_audio_file(audio_file_path)
- if not running:
- print("Interrupted.")
- return
- running = True
- signal.signal(signal.SIGINT, signal_handler)
- process_directory("D:\\Recordings")
Advertisement
Add Comment
Please, Sign In to add comment