(output_dir)
| 1148 | |
| 1149 | # TODO make the mp3 to wave a funciton and not duplicated |
| 1150 | def audio_integrity(output_dir): |
| 1151 | def check_wav_properties(f, filename): |
| 1152 | try: |
| 1153 | with wave.open(f, "rb") as wav_file: |
| 1154 | integrity_dir = os.path.join(results_folder, "audio_integrity") |
| 1155 | os.makedirs(integrity_dir, exist_ok=True) |
| 1156 | out_file = os.path.join(integrity_dir, f"{os.path.splitext(filename)[0]}.txt") |
| 1157 | with open(out_file, "w") as log: |
| 1158 | log.write(f"Channels: {wav_file.getnchannels()}\n") |
| 1159 | log.write(f"Frame rate: {wav_file.getframerate()}\n") |
| 1160 | log.write(f"Sample width: {wav_file.getsampwidth()}\n") |
| 1161 | log.write(f"Frames: {wav_file.getnframes()}\n") |
| 1162 | prGreen(f"[AUDIO INTEGRITY] Checked {filename}") |
| 1163 | except Exception as e: |
| 1164 | prRed(f"[AUDIO INTEGRITY] Failed to read {filename}: {e}") |
| 1165 | |
| 1166 | wav_dir = os.path.join(output_dir, "wav") |
| 1167 | if os.path.isdir(wav_dir): |
| 1168 | files = [f for f in os.listdir(wav_dir) if os.path.isfile(os.path.join(wav_dir, f))] |
| 1169 | with ThreadPoolExecutor(max_workers=os.cpu_count() - 1) as executor: |
| 1170 | for filename in tqdm(files, desc="Audio integrity test: "): |
| 1171 | f = os.path.join(wav_dir, filename) |
| 1172 | executor.submit(check_wav_properties, f, filename) |
| 1173 | prGreen("AUDIO INTEGRITY TEST DONE") |
| 1174 | |
| 1175 | |
| 1176 | def audio_dectection(output_dir): |
no test coverage detected