| | |
|---|
| | | self.vad_opts.frame_in_ms) |
|---|
| | | self.encoder = encoder |
|---|
| | | # init variables |
|---|
| | | self.is_final = False |
|---|
| | | self.data_buf_start_frame = 0 |
|---|
| | | self.frm_cnt = 0 |
|---|
| | | self.latest_confirmed_speech_frame = 0 |
|---|
| | |
|---|
| | | self.frontend = frontend |
|---|
| | | |
|---|
| | | def AllResetDetection(self): |
|---|
| | | self.is_final = False |
|---|
| | | self.data_buf_start_frame = 0 |
|---|
| | | self.frm_cnt = 0 |
|---|
| | | self.latest_confirmed_speech_frame = 0 |
|---|
| | |
|---|
| | | 0.000001)) |
|---|
| | | |
|---|
| | | def ComputeScores(self, feats: torch.Tensor, in_cache: Dict[str, torch.Tensor]) -> None: |
|---|
| | | scores = self.encoder(feats, in_cache) # return B * T * D |
|---|
| | | scores = self.encoder(feats, in_cache).to('cpu') # return B * T * D |
|---|
| | | assert scores.shape[1] == feats.shape[1], "The shape between feats and scores does not match" |
|---|
| | | self.vad_opts.nn_eval_block_size = scores.shape[1] |
|---|
| | | self.frm_cnt += scores.shape[1] # count total frames |
|---|
| | |
|---|
| | | - 1)) / self.vad_opts.noise_frame_num_used_for_snr |
|---|
| | | |
|---|
| | | return frame_state |
|---|
| | | |
|---|
| | | |
|---|
| | | def forward(self, feats: torch.Tensor, waveform: torch.tensor, in_cache: Dict[str, torch.Tensor] = dict(), |
|---|
| | | is_final: bool = False |
|---|
| | | ) -> Tuple[List[List[List[int]]], Dict[str, torch.Tensor]]: |
|---|
| | | if not in_cache: |
|---|
| | | self.AllResetDetection() |
|---|
| | | self.waveform = waveform # compute decibel for each frame |
|---|
| | | self.ComputeDecibel() |
|---|
| | | self.ComputeScores(feats, in_cache) |
|---|
| | |
|---|
| | | return segments, in_cache |
|---|
| | | |
|---|
| | | def forward_online(self, feats: torch.Tensor, waveform: torch.tensor, in_cache: Dict[str, torch.Tensor] = dict(), |
|---|
| | | is_final: bool = False, max_end_sil: int = 800 |
|---|
| | | ) -> Tuple[List[List[List[int]]], Dict[str, torch.Tensor]]: |
|---|
| | | is_final: bool = False, max_end_sil: int = 800 |
|---|
| | | ) -> Tuple[List[List[List[int]]], Dict[str, torch.Tensor]]: |
|---|
| | | if not in_cache: |
|---|
| | | self.AllResetDetection() |
|---|
| | | self.max_end_sil_frame_cnt_thresh = max_end_sil - self.vad_opts.speech_to_sil_time_thres |
|---|
| | | self.waveform = waveform # compute decibel for each frame |
|---|
| | | |
|---|
| | | |
|---|
| | | self.ComputeScores(feats, in_cache) |
|---|
| | | self.ComputeDecibel() |
|---|
| | | if not is_final: |
|---|
