# MIT License # # Copyright (c) 2020-2025 CNRS # Copyright (c) 2025- pyannoteAI # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. from typing import Dict, Optional, Sequence, Text, Tuple, Union import numpy as np from pyannote.audio.core.task import Problem, Resolution, Specifications from pyannote.audio.tasks.segmentation.mixins import SegmentationTask from pyannote.core import Segment, SlidingWindowFeature from pyannote.database import Protocol from torch_audiomentations.core.transforms_interface import BaseWaveformTransform from torchmetrics import Metric class VoiceActivityDetection(SegmentationTask): """Voice activity detection Voice activity detection (or VAD) is the task of detecting speech regions in a given audio recording. It is addressed as a binary (0 or 1) sequence labeling task. A frame is marked as "speech" (1) as soon as at least one speaker is active. Parameters ---------- protocol : Protocol pyannote.database protocol cache : str, optional As (meta-)data preparation might take a very long time for large datasets, it can be cached to disk for later (and faster!) re-use. When `cache` does not exist, `Task.prepare_data()` generates training and validation metadata from `protocol` and save them to disk. When `cache` exists, `Task.prepare_data()` is skipped and (meta)-data are loaded from disk. Defaults to a temporary path. duration : float, optional Chunks duration. Defaults to 2s. warm_up : float or (float, float), optional Use that many seconds on the left- and rightmost parts of each chunk to warm up the model. While the model does process those left- and right-most parts, only the remaining central part of each chunk is used for computing the loss during training, and for aggregating scores during inference. Defaults to 0. (i.e. no warm-up). balance: Sequence[Text], optional When provided, training samples are sampled uniformly with respect to these keys. For instance, setting `balance` to ["database","subset"] will make sure that each database & subset combination will be equally represented in the training samples. weight: str, optional When provided, use this key to as frame-wise weight in loss function. batch_size : int, optional Number of training samples per batch. Defaults to 32. num_workers : int, optional Number of workers used for generating training samples. Defaults to multiprocessing.cpu_count() // 2. pin_memory : bool, optional If True, data loaders will copy tensors into CUDA pinned memory before returning them. See pytorch documentation for more details. Defaults to False. augmentation : BaseWaveformTransform, optional torch_audiomentations waveform transform, used by dataloader during training. metric : optional Validation metric(s). Can be anything supported by torchmetrics.MetricCollection. Defaults to AUROC (area under the ROC curve). """ def __init__( self, protocol: Protocol, cache: Optional[Union[str, None]] = None, duration: float = 2.0, warm_up: Union[float, Tuple[float, float]] = 0.0, balance: Optional[Sequence[Text]] = None, weight: Optional[Text] = None, batch_size: int = 32, num_workers: Optional[int] = None, pin_memory: bool = False, augmentation: Optional[BaseWaveformTransform] = None, metric: Union[Metric, Sequence[Metric], Dict[str, Metric]] = None, ): super().__init__( protocol, duration=duration, warm_up=warm_up, batch_size=batch_size, num_workers=num_workers, pin_memory=pin_memory, augmentation=augmentation, metric=metric, cache=cache, ) self.balance = balance self.weight = weight self.specifications = Specifications( problem=Problem.BINARY_CLASSIFICATION, resolution=Resolution.FRAME, duration=self.duration, min_duration=self.min_duration, warm_up=self.warm_up, classes=[ "speech", ], ) def prepare_chunk(self, file_id: int, start_time: float, duration: float): """Prepare chunk for voice activity detection Parameters ---------- file_id : int File index start_time : float Chunk start time duration : float Chunk duration. Returns ------- sample : dict Dictionary containing the chunk data with the following keys: - `X`: waveform - `y`: target as a SlidingWindowFeature instance - `meta`: - `database`: database index - `file`: file index """ file = self.get_file(file_id) chunk = Segment(start_time, start_time + duration) sample = dict() sample["X"], _ = self.model.audio.crop(file, chunk) # gather all annotations of current file annotations = self.prepared_data["annotations-segments"][ self.prepared_data["annotations-segments"]["file_id"] == file_id ] # gather all annotations with non-empty intersection with current chunk chunk_annotations = annotations[ (annotations["start"] < chunk.end) & (annotations["end"] > chunk.start) ] # discretize chunk annotations at model output resolution step = self.model.receptive_field.step half = 0.5 * self.model.receptive_field.duration start = np.maximum(chunk_annotations["start"], chunk.start) - chunk.start - half start_idx = np.maximum(0, np.round(start / step)).astype(int) end = np.minimum(chunk_annotations["end"], chunk.end) - chunk.start - half end_idx = np.round(end / step).astype(int) # frame-level targets num_frames = self.model.num_frames( round(duration * self.model.hparams.sample_rate) ) y = np.zeros((num_frames, 1), dtype=np.uint8) for start, end in zip(start_idx, end_idx): y[start : end + 1, 0] = 1 sample["y"] = SlidingWindowFeature( y, self.model.receptive_field, labels=["speech"] ) metadata = self.prepared_data["audio-metadata"][file_id] sample["meta"] = {key: metadata[key] for key in metadata.dtype.names} sample["meta"]["file"] = file_id return sample