Method update_segments

whisper_live/backend/base.py:375–475

Processes the segments from Whisper and updates the transcript. Uses helper methods to account for differences between backends. Args: segments (list): List of segments returned by the transcriber. duration (float): Duration of the current au

(self, segments, duration)

        ]

    def update_segments(self, segments, duration):
        """
        Processes the segments from Whisper and updates the transcript.
        Uses helper methods to account for differences between backends.
        
        Args:
            segments (list): List of segments returned by the transcriber.
            duration (float): Duration of the current audio chunk.
        
        Returns:
            dict or None: The last processed segment (if any).
        """
        offset = None
        self.current_out = ''
        last_segment = None

        # Process complete segments only if there are more than one
        # and if the last segment's no_speech_prob is below the threshold.
        if len(segments) > 1 and self.get_segment_no_speech_prob(segments[-1]) <= self.no_speech_thresh:
            for s in segments[:-1]:
                text_ = s.text
                self.text.append(text_)
                with self.lock:
                    start = self.timestamp_offset + self.get_segment_start(s)
                    end = self.timestamp_offset + min(duration, self.get_segment_end(s))
                if start >= end:
                    continue
                if self.get_segment_no_speech_prob(s) > self.no_speech_thresh:
                    continue
                speaker = self._identify_speaker(s)
                words = self._extract_words(s, self.timestamp_offset)
                completed_segment = self.format_segment(start, end, text_, completed=True, speaker=speaker, words=words)
                self.transcript.append(completed_segment)

                if self.translation_queue:
                    try:
                        self.translation_queue.put(completed_segment.copy(), timeout=0.1)
                    except queue.Full:
                        logging.warning("Translation queue is full, skipping segment")
                offset = min(duration, self.get_segment_end(s))

        # Process the last segment if its no_speech_prob is acceptable.
        if self.get_segment_no_speech_prob(segments[-1]) <= self.no_speech_thresh:
            self.current_out += segments[-1].text
            words = self._extract_words(segments[-1], self.timestamp_offset)
            with self.lock:
                last_segment = self.format_segment(
                    self.timestamp_offset + self.get_segment_start(segments[-1]),
                    self.timestamp_offset + min(duration, self.get_segment_end(segments[-1])),
                    self.current_out,
                    completed=False,
                    words=words
                )

        # Handle repeated output logic.
        if self.current_out.strip() == self.prev_out.strip() and self.current_out != '':
            self.same_output_count += 1