utils_audio.py - main - bioxakep/VoiceTrans

utils_audio.py
270 строк11 KB
bioxakep
audio params func refactor
13 авг 2025, 23:42
13 авг 2025, 23:426e1234a
from dataclasses import dataclass

import math
import os.path
from numpy import mean
from pydub import AudioSegment

from loggers import get_logger
from config import app_config
from collections import defaultdict

from utils_common import humanize_time

utils_logger = get_logger(__name__, level=app_config.log_level, file_name="utils.log")


@dataclass
class CoefData:
    parts_count: int
    max_duration: int


class AudioPart:
    __slots__ = ("start", "end")

    def __init__(self, start_time: int, end_time: int):
        self.start = start_time
        self.end = end_time

    @property
    def duration(self):
        return self.end - self.start

    def __repr__(self):
        return "%s(%s, %s)" % (self.__class__.__name__, self.start, self.end)

    def __str__(self):
        return "%s: %s - %s" % (self.__class__.__name__, self.start, self.end)


def analyze_freq_of_amplitude(audio: AudioSegment, chunk_size: int = 200):
    dbfs_freq = defaultdict(int)
    max_dbfs_freq = defaultdict(int)
    for i in range(0, len(audio), chunk_size):
        audio_chunk = audio[i : i + chunk_size]
        dbfs_freq[str(10 * (int(audio_chunk.dBFS) // 10))] += 1
        max_dbfs_freq[str(10 * (int(audio_chunk.max_dBFS) // 10))] += 1

    with open("analysis_freq_of_dbfs.csv", "w") as f:
        f.write("dbfs;freq\n")
        for k, v in dbfs_freq.items():
            f.write("%s;%s\n" % (k, v))
    with open("analysis_freq_of_max_dbfs.csv", "w") as f:
        f.write("max_dbfs;freq\n")
        for k, v in max_dbfs_freq.items():
            f.write("%s;%s\n" % (k, v))
    return dbfs_freq, max_dbfs_freq


def get_voices_parts_old(
    audio: AudioSegment,
    min_speech_seconds: int = 3,
    threshold_coef: int = 7,  # of 10
) -> list[AudioPart] | None:
    count = 0
    min_dbfs = 0
    max_dbfs = -200
    chunk_size = 100
    window_size = 20
    dbfs_set = set()

    voice_parts = list()
    for s in range(0, len(audio), chunk_size):
        end = s + chunk_size if s + chunk_size < len(audio) else len(audio)
        audio_chunk = audio[s:end]
        if audio_chunk.max_dBFS > -math.inf and audio_chunk.dBFS > -math.inf:
            dbfs_set.add(5 * (int(audio_chunk.max_dBFS) // 5))
            dbfs_set.add(5 * (int(audio_chunk.dBFS) // 5))
        max_dbfs = max_dbfs if max_dbfs > audio_chunk.max_dBFS else audio_chunk.max_dBFS
        if audio_chunk.dBFS > -math.inf:
            min_dbfs = min_dbfs if min_dbfs < audio_chunk.dBFS else audio_chunk.dBFS
        count += 1
    if len(dbfs_set) == 0:
        return None
    sorted_dbfs_set = list(sorted(dbfs_set))
    window = []
    start_voice = 0
    speech_threshold = min_dbfs + threshold_coef * abs(max_dbfs - min_dbfs) // 10
    silence_threshold = min_dbfs + threshold_coef * abs(max_dbfs - min_dbfs) // 10
    utils_logger.info(
        f"Диапазон громкости: [%.2f : %.2f], порог начала: %.2f, порог конца: %.2f",
        min_dbfs,
        max_dbfs,
        speech_threshold,
        silence_threshold,
    )
    part_count = 0
    chunk_count = 0
    for s in range(0, len(audio), chunk_size):
        chunk_count += 1
        end = s + chunk_size if s + chunk_size < len(audio) else len(audio)
        audio_chunk = audio[s:end]
        if audio_chunk.max_dBFS > speech_threshold and start_voice == 0:
            start_voice = s - chunk_size // 2 if s - chunk_size // 2 > 0 else 0
        if start_voice > 0:
            window.append(
                audio_chunk.max_dBFS
                if audio_chunk.max_dBFS != -math.inf
                else min_dbfs - 1
            )
            if len(window) > window_size:
                window.pop(0)
        if len(window) == 0:
            continue
        if mean(window) < silence_threshold:
            part_end = (
                s + chunk_size // 2 if s + chunk_size // 2 < len(audio) else len(audio)
            )
            if end - start_voice < (min_speech_seconds * 1000):
                continue
            part_count += 1
            voice_parts.append(AudioPart(start_voice, part_end))
            start_voice = 0
            window.clear()
    if start_voice > 0:
        voice_parts.append(AudioPart(start_voice, len(audio)))
    if start_voice == 0 and part_count == 0 and audio.duration_seconds < 30:
        voice_parts.append(AudioPart(0, len(audio)))
    return voice_parts


def get_audio_params(audio_file_path: str):
    audio = AudioSegment.from_file(audio_file_path)
    bit_rate = audio.frame_rate * audio.sample_width * audio.channels * 8
    if audio.channels == 1:
        return audio.duration_seconds, False, False
    left_channel, right_channel = audio.split_to_mono()
    audio_dir: str = os.path.dirname(audio_file_path)
    file_name: str = os.path.basename(audio_file_path)
    file_name_template: str = ".".join(file_name.split(".")[:-1])
    left_path: str = os.path.join(audio_dir, file_name_template + "_left_channel.wav")
    right_path: str = os.path.join(audio_dir, file_name_template + "_right_channel.wav")
    left_channel.export(left_path, format="wav", bitrate=bit_rate)
    right_channel.export(right_path, format="wav", bitrate=bit_rate)
    return audio.duration_seconds, left_path, right_path


def get_speech_parts(
    audio: AudioSegment,
    min_duration: int = 2,
    max_duration: int = 25,
) -> list[AudioPart] | None:
    """
    Алгоритм определения порога громкости для разделения аудиофайла на речевые отрезки
    :param audio: Объект класса AudioSegment
    :param min_duration: Минимальная длительность речи в секундах
    :param max_duration: Максимальная длительность речи в секундах
    :return: Список объектов класса AudioPart
    """
    utils_logger.info(
        "Выделяем речевые отрезки из файла длительностью %s",
        humanize_time(audio.duration_seconds, clocks=False),
    )
    utils_logger.info(
        "Заданы минимальная и максимальная длительность речи: %d и %d секунды",
        min_duration,
        max_duration,
    )
    chunk_size = 100
    window_size = 20
    min_dbfs, max_dbfs = 0, -200
    chunks = len(audio) // chunk_size
    for i in range(chunks):
        curr_dbfs = audio[i * chunk_size : (i + 1) * chunk_size].dBFS
        if curr_dbfs == -math.inf:
            continue
        min_dbfs = min_dbfs if min_dbfs < curr_dbfs else curr_dbfs
        max_dbfs = max_dbfs if max_dbfs > curr_dbfs else curr_dbfs
    utils_logger.info(
        "Минимальная громкость: %.2f дБ, Максимальная громкость: %.2f дБ",
        min_dbfs,
        max_dbfs,
    )
    coef_parts = defaultdict(list)
    range_size = int(8 / 0.2)
    for j in range(range_size):
        k = 9.0 - (0.2 * j)
        k = round(k, 2)
        mean_dbfs = min_dbfs + k * (max_dbfs - min_dbfs) // 10
        start_voice = 0
        curr_audio_parts = []
        curr_max_duration = 0
        window = []
        for i in range(chunks):
            curr_dbfs = audio[i * chunk_size : (i + 1) * chunk_size].dBFS  # Float value
            curr_dbfs = curr_dbfs if curr_dbfs != -math.inf else min_dbfs
            window.append(curr_dbfs)
            if len(window) > window_size:
                window.pop(0)
            if len(window) < window_size:
                continue
            # if curr_dbfs > mean_dbfs and start_voice == 0:
            if mean(window) > mean_dbfs and start_voice == 0:
                start_voice = i - window_size
            if start_voice > 0:
                if mean(window) > mean_dbfs:
                    continue
                end_voice = i + 1
                if chunk_size * (end_voice - start_voice) <= (min_duration * 1000):
                    continue
                audio_part = AudioPart(
                    start_time=start_voice * chunk_size - chunk_size // 2,
                    end_time=end_voice * chunk_size + chunk_size // 2,
                )
                if audio_part.duration > curr_max_duration:
                    curr_max_duration = audio_part.duration
                curr_audio_parts.append(audio_part)
                start_voice = 0
                window.clear()
        if start_voice > 0 and mean(window) > mean_dbfs:
            end_of_part = len(audio)
            if len(audio) - start_voice * chunk_size > max_duration * 1000:
                end_of_part = start_voice * chunk_size + max_duration * 1000
                ignore_end = (len(audio) - end_of_part) / 1000
                utils_logger.info(f"Игнорируем кусочек длиной {ignore_end} секунд")
            last_part = AudioPart(
                start_time=start_voice * chunk_size - chunk_size // 2,
                end_time=end_of_part,
            )
            curr_audio_parts.append(last_part)

        # Анализ текущего коэффициента
        parts_count = len(curr_audio_parts)
        if parts_count == 0:
            continue
        if curr_max_duration < max_duration * 1000:
            coef_parts.update({k: curr_audio_parts})
    # Анализ с целью выбора наиболее эффективного порога громкости
    max_possible_duration = max(
        [max(part.duration for part in parts) for parts in coef_parts.values()]
    )
    best_coefficients = [
        k
        for k, parts in coef_parts.items()
        if max_possible_duration == max([p.duration for p in parts])
    ]
    if len(best_coefficients) == 0:
        return None
    min_parts_count = min([len(coef_parts.get(k)) for k in best_coefficients])
    for k, parts in coef_parts.items():
        if len(parts) == min_parts_count:
            utils_logger.info(
                "В результате анализа выбран порог громкости %.2f дБ",
                min_dbfs + k * (max_dbfs - min_dbfs) // 10,
            )
            utils_logger.info("Выделено отрезков речи: %s" % len(parts))
            return parts
    return None


if __name__ == "__main__":
    file_path: str = "/Users/bioxakep/IdeaProjects/VoiceTrans/assets/audio/Dialog.wav"
    audio_segment = AudioSegment.from_file(file_path)
    if audio_segment.channels > 1:
        left, right = audio_segment.split_to_mono()
        audio_segment = left
    voice_parts = get_speech_parts(audio_segment)
    if voice_parts is None:
        print("Речь не была выделена из файла")
    max_part_dur = max([voice_part.duration for voice_part in voice_parts])