Benchmark

BenchmarkABC

Bases: ABC

Interface that defines the operation and control of benchmarks.

Source code in asrbench\abc_benchmark.py

class BenchmarkABC(ABC):
    """Interface that defines the operation and control of benchmarks."""

    @property
    @abstractmethod
    def transcribers(self) -> Dict[str, Transcriber]:
        """Gets all the transcribers of the class."""
        raise NotImplementedError("Implement transcribers property.")

    @abstractmethod
    def run(self) -> str:
        """Run the transcription with each transcriber for each audio in each
        dataset of the class."""
        raise NotImplementedError("Implement run method.")

    @abstractmethod
    def run_with_transcriber(self, name: str) -> str:
        """Runs the benchmark only with the chosen transcriber."""
        raise NotImplementedError("Implement run with transcriber method.")

    def add_transcriber(self, name: str, transcriber: Transcriber) -> None:
        """Add a transcriber to the class."""
        if not isinstance(transcriber, Transcriber):
            raise ValueError(
                f"Transcriber {name} is not instance of Transcriber.",
            )

        self.transcribers[name] = transcriber

    def remove_transcriber(self, name: str) -> None:
        """Removes the transcriber from the class."""
        if name not in self.transcribers:
            raise KeyError(f"Transcriber {name} does not exists.")

        self.transcribers.pop(name)

transcribers: Dict[str, Transcriber] abstractmethod property

Gets all the transcribers of the class.

add_transcriber(name, transcriber)

Add a transcriber to the class.

Source code in asrbench\abc_benchmark.py

def add_transcriber(self, name: str, transcriber: Transcriber) -> None:
    """Add a transcriber to the class."""
    if not isinstance(transcriber, Transcriber):
        raise ValueError(
            f"Transcriber {name} is not instance of Transcriber.",
        )

    self.transcribers[name] = transcriber

remove_transcriber(name)

Removes the transcriber from the class.

Source code in asrbench\abc_benchmark.py

def remove_transcriber(self, name: str) -> None:
    """Removes the transcriber from the class."""
    if name not in self.transcribers:
        raise KeyError(f"Transcriber {name} does not exists.")

    self.transcribers.pop(name)

run() abstractmethod

Run the transcription with each transcriber for each audio in each dataset of the class.

Source code in asrbench\abc_benchmark.py

@abstractmethod
def run(self) -> str:
    """Run the transcription with each transcriber for each audio in each
    dataset of the class."""
    raise NotImplementedError("Implement run method.")

run_with_transcriber(name) abstractmethod

Runs the benchmark only with the chosen transcriber.

Source code in asrbench\abc_benchmark.py

@abstractmethod
def run_with_transcriber(self, name: str) -> str:
    """Runs the benchmark only with the chosen transcriber."""
    raise NotImplementedError("Implement run with transcriber method.")

BenchmarkContext

Controls the context and progress of Benchmark execution.

Parameters:

Name	Type	Description	Default
datasets	List[Dataset]	List of datasets in the Benchmark class.	required
observer	Observer	Observer class to display progress to user.	required

Source code in asrbench\benchmark_ctx.py

class BenchmarkContext:
    """Controls the context and progress of Benchmark execution.

    Arguments:
        datasets: List of datasets in the Benchmark class.
        observer: Observer class to display progress to user.
    """
    def __init__(self, datasets: List[Dataset], observer: Observer) -> None:
        self.__datasets: List[Dataset] = datasets
        self._total_pairs: int = 0
        self._current_pair: int = 0
        self._start: float = 0.0
        self._progress: float = 0.0
        self._observer: Observer = observer
        self.__dataset = None

    @property
    def dataset(self) -> Dataset:
        """Get the dataset currently running."""
        if self.__dataset is None:
            raise ValueError(
                "Context dataset is none, verify benchmark run methods.",
            )

        return self.__dataset

    def set_dataset(self, idx: int) -> None:
        """Defines the dataset to be executed."""
        self.__dataset: Dataset = self.__datasets[idx]
        self._total_pairs = len(self.__dataset.pairs)

    def start_progress(self) -> None:
        """Progress timer starts."""
        self._start: float = time.time()
        self._observer.start_progress()

    def reset_progress(self) -> None:
        """Restarts all execution progress."""
        self._current_pair: int = 0
        self._progress: float = 0.0

    def update_progress(self, transcriber_name: str) -> None:
        """Updates progress status."""
        self._current_pair += 1
        self._observer.update_progress(
            self._calculate_progress(),
            self._get_progress_msg(transcriber_name),
        )

        if self._current_pair == self._total_pairs:
            self._observer.finish()

    def _calculate_progress(self) -> float:
        self._progress = self._current_pair / self._total_pairs
        return self._progress

    def _get_progress_msg(self, transcriber: str) -> str:
        dataset: str = self.__dataset.name
        return f"Processing pairs from {dataset} with {transcriber}."

dataset: Dataset property

Get the dataset currently running.

reset_progress()

Restarts all execution progress.

Source code in asrbench\benchmark_ctx.py

def reset_progress(self) -> None:
    """Restarts all execution progress."""
    self._current_pair: int = 0
    self._progress: float = 0.0

set_dataset(idx)

Defines the dataset to be executed.

Source code in asrbench\benchmark_ctx.py

def set_dataset(self, idx: int) -> None:
    """Defines the dataset to be executed."""
    self.__dataset: Dataset = self.__datasets[idx]
    self._total_pairs = len(self.__dataset.pairs)

start_progress()

Progress timer starts.

Source code in asrbench\benchmark_ctx.py

def start_progress(self) -> None:
    """Progress timer starts."""
    self._start: float = time.time()
    self._observer.start_progress()

update_progress(transcriber_name)

Updates progress status.

Source code in asrbench\benchmark_ctx.py

def update_progress(self, transcriber_name: str) -> None:
    """Updates progress status."""
    self._current_pair += 1
    self._observer.update_progress(
        self._calculate_progress(),
        self._get_progress_msg(transcriber_name),
    )

    if self._current_pair == self._total_pairs:
        self._observer.finish()