Repository: livekit/agents
File: livekit-plugins/livekit-plugins-speechmatics/livekit/plugins/speechmatics/stt.py (Python)

Code before:
                    "sounds_like": e.sounds_like,
                }
                for e in self._stt_options.additional_vocab
            ]

        if self._stt_options.enable_diarization:
            dz_cfg: dict[str, Any] = {}
            if self._stt_options.diarization_sensitivity is not None:
                dz_cfg["speaker_sensitivity"] = self._stt_options.diarization_sensitivity
            if self._stt_options.prefer_current_speaker is not None:
                dz_cfg["prefer_current_speaker"] = self._stt_options.prefer_current_speaker
            if self._stt_options.known_speakers:
                dz_cfg["speakers"] = {
                    s.label: s.speaker_identifiers for s in self._stt_options.known_speakers
                }
            if dz_cfg:
                transcription_config.speaker_diarization_config = dz_cfg
        if (
            self._stt_options.end_of_utterance_silence_trigger
            and self._stt_options.end_of_utterance_mode == EndOfUtteranceMode.FIXED
        ):
            transcription_config.conversation_config = ConversationConfig(
                end_of_utterance_silence_trigger=self._stt_options.end_of_utterance_silence_trigger,
            )

        if self._stt_options.punctuation_overrides:
            transcription_config.punctuation_overrides = self._stt_options.punctuation_overrides

        self._transcription_config = transcription_config

    def update_speakers(
        self,
        focus_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
        ignore_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
        focus_mode: NotGivenOr[DiarizationFocusMode] = NOT_GIVEN,
    ) -> None:
        """Updates the speaker configuration.

        This can update the speakers to listen to or ignore during an in-flight
        transcription. Only available if diarization is enabled.

        Args:
            focus_speakers: List of speakers to focus on.
            ignore_speakers: List of speakers to ignore.
            focus_mode: Focus mode to use.
        """
        # Check possible
        if not self._stt_options.enable_diarization:
            raise ValueError("Diarization is not enabled")

        # Update the diarization configuration

Review comment (refactor):
we should also refactor `_process_config` to replace the dataclass value of `transcription_config` instead of assigning a dict to it.

Code after:
        transcription_config = TranscriptionConfig(
            language=self._stt_options.language,
            domain=self._stt_options.domain,
            output_locale=self._stt_options.output_locale,
            operating_point=self._stt_options.operating_point,
            diarization="speaker" if self._stt_options.enable_diarization else None,
            enable_partials=self._stt_options.enable_partials,
            max_delay=self._stt_options.max_delay,
        )

        if self._stt_options.additional_vocab:
            # API expects list of dicts, not dict format
            transcription_config.additional_vocab = [  # type: ignore
                {"content": e.content, "sounds_like": e.sounds_like}
                for e in self._stt_options.additional_vocab
            ]

        if self._stt_options.enable_diarization:
            # Use dict for speaker diarization config to support all fields including speakers
            dz_cfg: dict[str, Any] = {
                "speaker_sensitivity": self._stt_options.diarization_sensitivity,
                "prefer_current_speaker": self._stt_options.prefer_current_speaker,
            }

            # Add max_speakers if provided
            if self._stt_options.max_speakers is not None:
                dz_cfg["max_speakers"] = self._stt_options.max_speakers

            # Add speakers mapping from known speakers
            if self._stt_options.known_speakers:
                dz_cfg["speakers"] = {
                    s.label: s.speaker_identifiers for s in self._stt_options.known_speakers
                }

            transcription_config.speaker_diarization_config = dz_cfg  # type: ignore[assignment]
        if (
            self._stt_options.end_of_utterance_silence_trigger
            and self._stt_options.end_of_utterance_mode == EndOfUtteranceMode.FIXED
        ):
            transcription_config.conversation_config = ConversationConfig(
                end_of_utterance_silence_trigger=self._stt_options.end_of_utterance_silence_trigger,
            )

        if self._stt_options.punctuation_overrides:
            transcription_config.punctuation_overrides = self._stt_options.punctuation_overrides

        self._transcription_config = transcription_config

    def update_speakers(
        self,
        focus_speakers: NotGivenOr[list[str]] = NOT_GIVEN,