speechcommands.py 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149
  1. import os
  2. from pathlib import Path
  3. from typing import Optional, Tuple, Union
  4. import torchaudio
  5. from torch import Tensor
  6. from torch.hub import download_url_to_file
  7. from torch.utils.data import Dataset
  8. from torchaudio.datasets.utils import extract_archive
  9. FOLDER_IN_ARCHIVE = "SpeechCommands"
  10. URL = "speech_commands_v0.02"
  11. HASH_DIVIDER = "_nohash_"
  12. EXCEPT_FOLDER = "_background_noise_"
  13. _CHECKSUMS = {
  14. "https://storage.googleapis.com/download.tensorflow.org/data/speech_commands_v0.01.tar.gz": "743935421bb51cccdb6bdd152e04c5c70274e935c82119ad7faeec31780d811d", # noqa: E501
  15. "https://storage.googleapis.com/download.tensorflow.org/data/speech_commands_v0.02.tar.gz": "af14739ee7dc311471de98f5f9d2c9191b18aedfe957f4a6ff791c709868ff58", # noqa: E501
  16. }
  17. def _load_list(root, *filenames):
  18. output = []
  19. for filename in filenames:
  20. filepath = os.path.join(root, filename)
  21. with open(filepath) as fileobj:
  22. output += [os.path.normpath(os.path.join(root, line.strip())) for line in fileobj]
  23. return output
  24. def load_speechcommands_item(filepath: str, path: str) -> Tuple[Tensor, int, str, str, int]:
  25. relpath = os.path.relpath(filepath, path)
  26. label, filename = os.path.split(relpath)
  27. # Besides the officially supported split method for datasets defined by "validation_list.txt"
  28. # and "testing_list.txt" over "speech_commands_v0.0x.tar.gz" archives, an alternative split
  29. # method referred to in paragraph 2-3 of Section 7.1, references 13 and 14 of the original
  30. # paper, and the checksums file from the tensorflow_datasets package [1] is also supported.
  31. # Some filenames in those "speech_commands_test_set_v0.0x.tar.gz" archives have the form
  32. # "xxx.wav.wav", so file extensions twice needs to be stripped twice.
  33. # [1] https://github.com/tensorflow/datasets/blob/master/tensorflow_datasets/url_checksums/speech_commands.txt
  34. speaker, _ = os.path.splitext(filename)
  35. speaker, _ = os.path.splitext(speaker)
  36. speaker_id, utterance_number = speaker.split(HASH_DIVIDER)
  37. utterance_number = int(utterance_number)
  38. # Load audio
  39. waveform, sample_rate = torchaudio.load(filepath)
  40. return waveform, sample_rate, label, speaker_id, utterance_number
  41. class SPEECHCOMMANDS(Dataset):
  42. """Create a Dataset for *Speech Commands* [:footcite:`speechcommandsv2`].
  43. Args:
  44. root (str or Path): Path to the directory where the dataset is found or downloaded.
  45. url (str, optional): The URL to download the dataset from,
  46. or the type of the dataset to dowload.
  47. Allowed type values are ``"speech_commands_v0.01"`` and ``"speech_commands_v0.02"``
  48. (default: ``"speech_commands_v0.02"``)
  49. folder_in_archive (str, optional):
  50. The top-level directory of the dataset. (default: ``"SpeechCommands"``)
  51. download (bool, optional):
  52. Whether to download the dataset if it is not found at root path. (default: ``False``).
  53. subset (str or None, optional):
  54. Select a subset of the dataset [None, "training", "validation", "testing"]. None means
  55. the whole dataset. "validation" and "testing" are defined in "validation_list.txt" and
  56. "testing_list.txt", respectively, and "training" is the rest. Details for the files
  57. "validation_list.txt" and "testing_list.txt" are explained in the README of the dataset
  58. and in the introduction of Section 7 of the original paper and its reference 12. The
  59. original paper can be found `here <https://arxiv.org/pdf/1804.03209.pdf>`_. (Default: ``None``)
  60. """
  61. def __init__(
  62. self,
  63. root: Union[str, Path],
  64. url: str = URL,
  65. folder_in_archive: str = FOLDER_IN_ARCHIVE,
  66. download: bool = False,
  67. subset: Optional[str] = None,
  68. ) -> None:
  69. assert subset is None or subset in ["training", "validation", "testing"], (
  70. "When `subset` not None, it must take a value from " + "{'training', 'validation', 'testing'}."
  71. )
  72. if url in [
  73. "speech_commands_v0.01",
  74. "speech_commands_v0.02",
  75. ]:
  76. base_url = "https://storage.googleapis.com/download.tensorflow.org/data/"
  77. ext_archive = ".tar.gz"
  78. url = os.path.join(base_url, url + ext_archive)
  79. # Get string representation of 'root' in case Path object is passed
  80. root = os.fspath(root)
  81. basename = os.path.basename(url)
  82. archive = os.path.join(root, basename)
  83. basename = basename.rsplit(".", 2)[0]
  84. folder_in_archive = os.path.join(folder_in_archive, basename)
  85. self._path = os.path.join(root, folder_in_archive)
  86. if download:
  87. if not os.path.isdir(self._path):
  88. if not os.path.isfile(archive):
  89. checksum = _CHECKSUMS.get(url, None)
  90. download_url_to_file(url, archive, hash_prefix=checksum)
  91. extract_archive(archive, self._path)
  92. else:
  93. if not os.path.exists(self._path):
  94. raise RuntimeError(
  95. f"The path {self._path} doesn't exist. "
  96. "Please check the ``root`` path or set `download=True` to download it"
  97. )
  98. if subset == "validation":
  99. self._walker = _load_list(self._path, "validation_list.txt")
  100. elif subset == "testing":
  101. self._walker = _load_list(self._path, "testing_list.txt")
  102. elif subset == "training":
  103. excludes = set(_load_list(self._path, "validation_list.txt", "testing_list.txt"))
  104. walker = sorted(str(p) for p in Path(self._path).glob("*/*.wav"))
  105. self._walker = [
  106. w
  107. for w in walker
  108. if HASH_DIVIDER in w and EXCEPT_FOLDER not in w and os.path.normpath(w) not in excludes
  109. ]
  110. else:
  111. walker = sorted(str(p) for p in Path(self._path).glob("*/*.wav"))
  112. self._walker = [w for w in walker if HASH_DIVIDER in w and EXCEPT_FOLDER not in w]
  113. def __getitem__(self, n: int) -> Tuple[Tensor, int, str, str, int]:
  114. """Load the n-th sample from the dataset.
  115. Args:
  116. n (int): The index of the sample to be loaded
  117. Returns:
  118. (Tensor, int, str, str, int):
  119. ``(waveform, sample_rate, label, speaker_id, utterance_number)``
  120. """
  121. fileid = self._walker[n]
  122. return load_speechcommands_item(fileid, self._path)
  123. def __len__(self) -> int:
  124. return len(self._walker)