[Playbook] Cochl.Scribe 따라하기

pip install transformers

git clone https://github.com/speechbrain/speechbrain.git cd speechbrain pip install -r requirments.txt pip install --editable .

from speechbrain.pretrained.interfaces import Speech_Emotion_Diarization # classifier 모델 classifier = Speech_Emotion_Diarization.from_hparams( source="speechbrain/emotion-diarization-wavlm-large" ) # 예시 파일을 통한 결과 확인 results = classifier.diarize_file("speechbrain/emotion-diarization-wavlm-large/example.wav") # 결과 보여주기 print(results)

pip install git+https://github.com/m-bain/whisperx.git

import whisperx import gc device = "cuda" audio_file = "audio.mp3" batch_size = 16 # reduce if low on GPU mem compute_type = "float16" # change to "int8" if low on GPU mem (may reduce accuracy) # 1. Transcribe with original whisper (batched) model = whisperx.load_model("large-v2", device, compute_type=compute_type) audio = whisperx.load_audio(audio_file) result = model.transcribe(audio, batch_size=batch_size) print(result["segments"]) # before alignment # delete model if low on GPU resources # import gc; gc.collect(); torch.cuda.empty_cache(); del model # 2. Align whisper output model_a, metadata = whisperx.load_align_model(language_code=result["language"], device=device) result = whisperx.align(result["segments"], model_a, metadata, audio, device, return_char_alignments=False) print(result["segments"]) # after alignment # delete model if low on GPU resources # import gc; gc.collect(); torch.cuda.empty_cache(); del model_a # 3. Assign speaker labels diarize_model = whisperx.DiarizationPipeline(use_auth_token=YOUR_HF_TOKEN, device=device) # add min/max number of speakers if known diarize_segments = diarize_model(audio) # diarize_model(audio, min_speakers=min_speakers, max_speakers=max_speakers) result = whisperx.assign_word_speakers(diarize_segments, result) print(diarize_segments) print(result["segments"]) # segments are now assigned speaker IDs

pip install ffmpeg

def extract_audio(input_file, output_file = './dataset/audio/extracted_audio.mp3'): cmd = [ 'ffmpeg', '-i', input_file, '-vn', output_file ] subprocess.run(cmd, input='y\n', text=True) return output_file

def cochlSense(filename, api_key = './assets/api_key.json'): with open(api_key, 'rb') as fr: apifile = json.load(fr) apikey = apifile['cochl_key'] configuration = sense.Configuration() configuration.api_key['API_Key'] = apikey client = sense.ApiClient(configuration) api = sense_api.AudioSessionApi(client) f = open(filename, "rb") size = os.stat(filename).st_size #create a new audio session session = api.create_session(CreateSession( content_type="audio/" + os.path.splitext(filename)[1][1:], type=AudioType("file"), total_size=size, )) results_list = [] #upload file per 1Mib chunks seq = session.chunk_sequence while True: chunk = f.read(2**20) if not chunk: break encoded = base64.b64encode(chunk).decode("utf-8") print("uploading") uploaded = api.upload_chunk(session_id=session.session_id, chunk_sequence=seq, audio_chunk=AudioChunk(encoded)) seq = uploaded.chunk_sequence # read inference result next_token = "" while True: resp = api.read_status(session_id=session.session_id, next_token=next_token) for result in resp.inference.results: result_dict = result.to_dict() print(json.dumps(result_dict)) results_list.append(result_dict) if "next_token" in resp.inference.page: next_token = resp.inference.page.next_token else: break # list to json file json_path = 'dataset/data/result_tags.json' with open(json_path, 'w', encoding='utf-8') as json_file: json.dump(results_list, json_file, ensure_ascii=False, indent=4) # list to csv keys = results_list[0].keys() csv_path = 'dataset/data/result_tags.csv' with open(csv_path, 'w', newline='', encoding='utf-8') as csv_file: writer = csv.DictWriter(csv_file, fieldnames=keys) writer.writeheader() for row in results_list: writer.writerow(row) logging_print("CochlSense tags obtained.") return csv_path, json_path def speechMood(audio): # If you want to change your extracted audio file, change the audio path below audio_path = './dataset/audio/extracted_audio.mp3' classifier = Speech_Emotion_Diarization.from_hparams( source="speechbrain/emotion-diarization-wavlm-large" ) diary = classifier.diarize_file(audio) logging_print("Speech Mood tags obtained.") return diary["audio_path"] def whisper_result(audio_path, language = 'ko', api_keys = './assets/api_key.json'): args = cli() global result # Get the huggingface user token with open(api_keys, 'rb') as fr: apifile = json.load(fr) my_token = apifile['hf_token'] # whisperx inference options options = dict(language=language, beam_size=5, best_of=5) transcribe_options = dict(task="transcribe", **options) # whisperx reference model inference model = whisperx.load_model("large-v2", device=args.device, compute_type=args.compute_type) audio = whisperx.load_audio(audio_path) # Align whisperx output result = model.transcribe(audio, batch_size=args.batch_size) model_a, metadata = whisperx.load_align_model(language_code=result["language"], device=args.device) result = whisperx.align(result["segments"], model_a, metadata, audio, device=args.device, return_char_alignments=False) # if you want to speaker diarization, uncomment the following lines ''' # Assign speaker labels diarize_model = whisperxx.DiarizationPipeline(use_auth_token=my_token, device=args.device) # add min/max number of speakers if known diarize_segments = diarize_model(audio) result = whisperxx.assign_word_speakers(diarize_segments, result) ''' logging_print("Whisper results obtained.") return result['segments']

{ "tags": [ {"probability": 0.5, "name": "Knock"}, {"probability": 0.5, "name": "Whistle"} ], "start_time": 7.0, "end_time": 8.0 }

{ "tags": [ {"probability": 0.0, "name": "Others"} ], "start_time": 7.0, "end_time": 8.0 }

{ 'start': 11.012, 'end': 12.657, 'text': "어디서든 체험해보세요.", 'words': [ { 'word': "어디서든", 'start': 11.012, 'end': 11.412, 'score': 0.038 }, { 'word': "체험해보세요.", 'start': 11.472, 'end': 12.657, 'score': 0.075 } ] }

{ './dataset/audio/extracted_audio.mp3': [ {'start': 0.0, 'end': 0.02, 'emotion': 'h'}, {'start': 0.02, 'end': 0.06, 'emotion': 'a'}, {'start': 0.06, 'end': 0.1, 'emotion': 'n'} ] }

def combine_results(speech_moods, sense_csv, whispers): # Extract data from speech_moods speech_mood_data = [] for mood in speech_moods: start_time = mood['start'] end_time = mood['end'] emotion = mood['emotion'] speech_mood_data.append({ 'start_time': start_time, 'end_time': end_time, 'emotion': emotion }) # Extract data from sense_csv sense_data = [] sense_df = pd.read_csv(sense_csv) for idx, row in sense_df.iterrows(): start_time = row['start_time'] end_time = row['end_time'] tags = ast.literal_eval(row['tags']) for tag in tags: if 'Others' not in tag['name']: sense_data.append({ 'start_time': start_time, 'end_time': end_time, 'tag_name': tag['name'] }) # Extract data from whispers whispers_data = [] for whisper in whispers: start_time = whisper['start'] end_time = whisper['end'] text = whisper['text'] for sense_tag in sense_data: if start_time >= sense_tag['start_time'] and end_time <= sense_tag['end_time']: # Convert emotion code to text using emotion_to_text function emotion_text = emotion_to_text(sense_tag['emotion']) subtitle_text = f"[{emotion_text} {sense_tag['tag_name']}] <br> {text}" whispers_data.append({ 'start_time': start_time, 'end_time': end_time, 'text': subtitle_text }) break # Creating the combined_results DataFrame combined_results = pd.DataFrame(whispers_data) combined_results.reset_index(inplace=True) combined_results.columns = ['index', 'start_time', 'end_time', 'subtitles'] return combined_results

import os def generate_smi(filename: str, subtitles: List[Tuple[float, float, str]], output_dir: str = "dataset/output/", lang: str = "ko"): output_path = os.path.join(output_dir, filename) with open(output_path, "w") as f: f.write("<SAMI>\n") f.write("<HEAD>\n") f.write("<Title>Sample SMI file</Title>\n") f.write("</HEAD>\n") f.write("<BODY>\n") for start_time, _, text in subtitles: f.write(f"<SYNC Start={int(start_time * 1000)}>\n") f.write(f"<P Class={lang}CC>{text}</P>\n") f.write("</SYNC>\n") f.write("</BODY>\n") f.write("</SAMI>\n") def matching_formats(subtitles, args): if args.output_type == 'all': generate_smi(args.output_path, subtitles) generate_srt(args.output_path, subtitles) generate_txt(args.output_path, subtitles) elif args.output_type == 'smi': return generate_smi(args.output_path, subtitles) elif args.output_type == 'srt': return generate_srt(args.output_path, subtitles) elif args.output_type == 'txt': return generate_txt(args.output_path, subtitles) else: raise ValueError(f"Invalid output type: {args.output_type}")

def generate_smi(filename: str, subtitles: List[Tuple[float, float, str]], output_dir: str = "dataset/output/", lang: str = "ko"): output_path = os.path.join(output_dir, filename) with open(output_path, "w") as f: f.write("<SAMI>\n") f.write("<HEAD>\n") f.write("<Title>Sample SMI file</Title>\n") f.write("</HEAD>\n") f.write("<BODY>\n") for start_time, _, text in subtitles: f.write(f"<SYNC Start={int(start_time * 1000)}>\n") f.write(f"<P Class={lang}CC>{text}</P>\n") f.write("</SYNC>\n") f.write("</BODY>\n") f.write("</SAMI>\n")

def matching_formats(subtitles, args): if args.output_type == 'all': generate_smi(args.output_path, subtitles) generate_srt(args.output_path, subtitles) generate_txt(args.output_path, subtitles) elif args.output_type == 'smi': return generate_smi(args.output_path, subtitles) elif args.output_type == 'srt': return generate_srt(args.output_path, subtitles) elif args.output_type == 'txt': return generate_txt(args.output_path, subtitles) else: raise ValueError(f"Invalid output type: {args.output_type}")

if name == "__main__": try: args = cli() # Extracting audio from the video audio = extract_audio(args.input_path) # Getting WhisperX tags whispers = whisper_result(audio) # Getting SpeechBrain tags speech_moods = speechMood(audio) # Getting CochlSense tags sense_tags, _ = cochlSense(audio) # Transforming the CochlSense tags DataFrame combined_results = combine_results(speech_moods, sense_tags, whispers) # Writing subtitles subtitles = generate_subtitles(combined_results, whispers, speech_moods) # Writing the subtitles in the desired format matching_formats(subtitles, args) if args.visualize: visualizer(subtitles, args) # Visualizing if the argument is provided else: logging_print("Transcript completed. No visualization.") except Exception as e: logging.error(f"An error occurred: {e}") # Logging any unexpected error

$ git clone https://www.github.com/jwyeeh-dev/CochlScribe $ cd CochlScribe $ python cochlscribe.py -iv [input_video] -o [output_video] -ot [output_type]