Module livekit.plugins.krisp_internal.plugin

Classes

class KrispVivaFilterFrameProcessor (*,
mode: VivaMode | None = VivaMode.VOICE_ISOLATION,
noise_suppression_level: float = 90.0,
frame_duration_ms: int | None = 10,
sample_rate: int | None = 16000)

Expand source code

class KrispVivaFilterFrameProcessor(rtc.FrameProcessor[rtc.AudioFrame]):
    """Krisp Viva noise filter as a LiveKit FrameProcessor.

    The native filter is created on the first frame after credentials arrive.
    Creation does not block the event loop: the SDK init and model load run on
    a native background thread, and frames pass through unfiltered until it
    is ready.
    """

    def __init__(
        self,
        *,
        mode: VivaMode | None = VivaMode.VOICE_ISOLATION,
        noise_suppression_level: float = 90.0,
        frame_duration_ms: int | None = 10,
        sample_rate: int | None = 16_000,
    ) -> None:
        self._mode = mode
        self._level = float(noise_suppression_level)
        self._frame_duration_ms = frame_duration_ms
        self._enabled = True

        self._filter: KrispVivaFilter | None = None
        self._info: StreamInfo | None = None
        self._credentials: Credentials | None = None
        self._effective_rate: int | None = sample_rate
        self._warned_channels = False

    @property
    def enabled(self) -> bool:
        return self._enabled

    @enabled.setter
    def enabled(self, value: bool) -> None:
        self._enabled = value

    @property
    def noise_suppression_level(self) -> float:
        return self._level

    @noise_suppression_level.setter
    def noise_suppression_level(self, value: float) -> None:
        clamped = max(0.0, min(100.0, float(value)))
        self._level = clamped
        if self._filter is not None:
            self._filter.set_noise_suppression_level(clamped)

    def _on_stream_info_updated(
        self,
        *,
        room_name: str,
        participant_identity: str,
        publication_sid: str,
    ) -> None:
        self._info = StreamInfo(
            room_id="",
            room_name=room_name,
            participant_identity=participant_identity,
            participant_id="",
            track_id=publication_sid,
        )
        if self._filter is not None:
            self._filter.update_stream_info(self._info)

    def _on_credentials_updated(self, *, token: str, url: str) -> None:
        self._credentials = Credentials(token=token, url=url)
        if self._filter is not None:
            try:
                self._filter.update_credentials(self._credentials)
            except FilterError as e:
                logger.error("Krisp credential rotation failed: %s", e)

    def _process(self, frame: rtc.AudioFrame) -> rtc.AudioFrame:
        if not self._enabled:
            return frame
        if self._credentials is None:
            return frame
        if frame.num_channels != 1:
            # Krisp NC expects mono. Multi-channel support would require
            # processing each channel independently. Pass through untouched.
            if not self._warned_channels:
                logger.warning(
                    "Krisp filter not applied: expected mono audio but got %d "
                    "channels; frames are passed through unprocessed.",
                    frame.num_channels,
                )
                self._warned_channels = True
            return frame

        if self._filter is None or self._effective_rate != frame.sample_rate:
            try:
                self._filter = KrispVivaFilter(
                    KrispVivaFilterSettings(
                        mode=self._mode,
                        sample_rate=frame.sample_rate,
                        frame_duration_ms=self._frame_duration_ms,
                        noise_suppression_level=self._level,
                        credentials=self._credentials,
                    )
                )
                self._effective_rate = frame.sample_rate
                if self._info is not None and self._filter is not None:
                    self._filter.update_stream_info(self._info)
            except FilterError as e:
                logger.error("Krisp filter init failed: %s", e)
                self._filter = None
                return frame

        in_arr = np.frombuffer(frame.data, dtype=np.int16)
        out_arr = np.empty_like(in_arr)
        in_buf = NativeInt16BufferMut(ptr=in_arr.ctypes.data, len=len(in_arr))
        out_buf = NativeInt16BufferMut(ptr=out_arr.ctypes.data, len=len(out_arr))
        try:
            self._filter.process(in_buf, out_buf)
        except FilterError as e:
            logger.error("Krisp processing failed: %s", e)
            return frame

        return rtc.AudioFrame(
            data=out_arr.tobytes(),
            sample_rate=frame.sample_rate,
            num_channels=frame.num_channels,
            samples_per_channel=frame.samples_per_channel,
            userdata=frame.userdata,
        )

    def _close(self) -> None:
        if self._filter is not None:
            self._filter.close_session()
        self._effective_rate = None
        self._info = None

Krisp Viva noise filter as a LiveKit FrameProcessor.

The native filter is created on the first frame after credentials arrive. Creation does not block the event loop: the SDK init and model load run on a native background thread, and frames pass through unfiltered until it is ready.

Ancestors

• FrameProcessor
• typing.Generic
• abc.ABC

Instance variables

prop enabled : bool

Expand source code

@property
def enabled(self) -> bool:
    return self._enabled

prop noise_suppression_level : float

Expand source code

@property
def noise_suppression_level(self) -> float:
    return self._level

class VivaMode (*args, **kwds)

Expand source code

class VivaMode(enum.Enum):
    
    VOICE_ISOLATION = 0
    
    VOICE_ISOLATION_TELEPHONY = 1

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var VOICE_ISOLATION

var VOICE_ISOLATION_TELEPHONY