(
self, uid: str, data: Dict[str, Union[str, np.ndarray]]
)
| 1863 | |
| 1864 | @typechecked |
| 1865 | def _speech_process( |
| 1866 | self, uid: str, data: Dict[str, Union[str, np.ndarray]] |
| 1867 | ) -> Dict[str, Union[str, np.ndarray]]: |
| 1868 | |
| 1869 | ref_names = [k for k in data.keys() if re.match(r"speech_ref\d+", k)] |
| 1870 | num_spk = len(ref_names) |
| 1871 | |
| 1872 | aux_names = [k for k in data.keys() if re.match(r"enroll_ref\d+", k)] |
| 1873 | if self.flexible_numspk: |
| 1874 | # The number of speaker varies in each sample. |
| 1875 | # Different speaker signals are stacked in the first dimension. |
| 1876 | enroll_name = "enroll_ref1" |
| 1877 | for name in aux_names: |
| 1878 | if name != enroll_name: |
| 1879 | data.pop(name) |
| 1880 | aux_names = [f"enroll_ref{i + 1}" for i in range(num_spk)] |
| 1881 | # Divide the concatenated enrollments into single speaker enrollments |
| 1882 | # NOTE(wangyou): whitespace is not allowed inside each path |
| 1883 | tup = data[enroll_name].split() |
| 1884 | if len(tup) == num_spk: |
| 1885 | # normal format in `enroll_spk1.scp`: |
| 1886 | # MIXTURE_UID /path/to/enrollment_or_embedding |
| 1887 | for i in range(num_spk - 1, -1, -1): |
| 1888 | data[f"enroll_ref{i + 1}"] = tup[i] |
| 1889 | elif len(tup) == num_spk * 2: |
| 1890 | # a special format in `enroll_spk1.scp`: |
| 1891 | # MIXTURE_UID *UID SPEAKER_ID |
| 1892 | for i in range(num_spk - 1, -1, -1): |
| 1893 | data[f"enroll_ref{i + 1}"] = " ".join(tup[i * 2 : i * 2 + 2]) |
| 1894 | else: |
| 1895 | raise ValueError( |
| 1896 | f"Invalid format with in enroll_spk1.scp. Expected {num_spk} or " |
| 1897 | f"{num_spk * 2} columns, got {len(tup)} columns:\n{tup}" |
| 1898 | ) |
| 1899 | |
| 1900 | if self.train: |
| 1901 | assert len(ref_names) == len(aux_names), (len(ref_names), len(aux_names)) |
| 1902 | if not self.load_all_speakers: |
| 1903 | # only load one target-speaker data |
| 1904 | spk = np.random.randint(0, num_spk) |
| 1905 | for i, name in enumerate(ref_names): |
| 1906 | if i == 0: |
| 1907 | data[name] = data[ref_names[spk]] |
| 1908 | else: |
| 1909 | data.pop(name) |
| 1910 | continue |
| 1911 | |
| 1912 | for i, name in enumerate(aux_names): |
| 1913 | if not self.load_all_speakers: |
| 1914 | if i == 0: |
| 1915 | data[name] = data[aux_names[spk]] |
| 1916 | else: |
| 1917 | data.pop(name) |
| 1918 | continue |
| 1919 | if self.train_spk2enroll is None: |
| 1920 | # normal format in `enroll_spk?.scp`: |
| 1921 | # MIXTURE_UID /path/to/enrollment_or_embedding |
| 1922 | assert not data[name].startswith("*"), data[name] |
no test coverage detected