# This file was autogenerated by some hot garbage in the `uniffi` crate. # Trust me, you don't want to mess with it! # Common helper code. # # Ideally this would live in a separate .py file where it can be unittested etc # in isolation, and perhaps even published as a re-useable package. # # However, it's important that the details of how this helper code works (e.g. the # way that different builtin types are passed across the FFI) exactly match what's # expected by the rust code on the other side of the interface. In practice right # now that means coming from the exact some version of `uniffi` that was used to # compile the rust component. The easiest way to ensure this is to bundle the Python # helpers directly inline like we're doing here. from __future__ import annotations import os import sys import ctypes from dataclasses import dataclass import enum import struct import contextlib import datetime import threading import itertools import traceback import typing import platform # Used for default argument values _DEFAULT = object() # type: typing.Any class _UniffiRustBuffer(ctypes.Structure): _fields_ = [ ("capacity", ctypes.c_uint64), ("len", ctypes.c_uint64), ("data", ctypes.POINTER(ctypes.c_char)), ] @staticmethod def default(): return _UniffiRustBuffer(0, 0, None) @staticmethod def alloc(size): return _uniffi_rust_call(_UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_alloc, size) @staticmethod def reserve(rbuf, additional): return _uniffi_rust_call(_UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_reserve, rbuf, additional) def free(self): return _uniffi_rust_call(_UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_free, self) def __str__(self): return "_UniffiRustBuffer(capacity={}, len={}, data={})".format( self.capacity, self.len, self.data[0:self.len] ) @contextlib.contextmanager def alloc_with_builder(*args): """Context-manger to allocate a buffer using a _UniffiRustBufferBuilder. The allocated buffer will be automatically freed if an error occurs, ensuring that we don't accidentally leak it. """ builder = _UniffiRustBufferBuilder() try: yield builder except: builder.discard() raise @contextlib.contextmanager def consume_with_stream(self): """Context-manager to consume a buffer using a _UniffiRustBufferStream. The _UniffiRustBuffer will be freed once the context-manager exits, ensuring that we don't leak it even if an error occurs. """ try: s = _UniffiRustBufferStream.from_rust_buffer(self) yield s if s.remaining() != 0: raise RuntimeError(f"junk data left in buffer at end of consume_with_stream {s.remaining()}") finally: self.free() @contextlib.contextmanager def read_with_stream(self): """Context-manager to read a buffer using a _UniffiRustBufferStream. This is like consume_with_stream, but doesn't free the buffer afterwards. It should only be used with borrowed `_UniffiRustBuffer` data. """ s = _UniffiRustBufferStream.from_rust_buffer(self) yield s if s.remaining() != 0: raise RuntimeError(f"junk data left in buffer at end of read_with_stream {s.remaining()}") class _UniffiForeignBytes(ctypes.Structure): _fields_ = [ ("len", ctypes.c_int32), ("data", ctypes.POINTER(ctypes.c_char)), ] def __str__(self): return "_UniffiForeignBytes(len={}, data={})".format(self.len, self.data[0:self.len]) class _UniffiRustBufferStream: """ Helper for structured reading of bytes from a _UniffiRustBuffer """ def __init__(self, data, len): self.data = data self.len = len self.offset = 0 @classmethod def from_rust_buffer(cls, buf): return cls(buf.data, buf.len) def remaining(self): return self.len - self.offset def _unpack_from(self, size, format): if self.offset + size > self.len: raise InternalError("read past end of rust buffer") value = struct.unpack(format, self.data[self.offset:self.offset+size])[0] self.offset += size return value def read(self, size): if self.offset + size > self.len: raise InternalError("read past end of rust buffer") data = self.data[self.offset:self.offset+size] self.offset += size return data def read_i8(self): return self._unpack_from(1, ">b") def read_u8(self): return self._unpack_from(1, ">B") def read_i16(self): return self._unpack_from(2, ">h") def read_u16(self): return self._unpack_from(2, ">H") def read_i32(self): return self._unpack_from(4, ">i") def read_u32(self): return self._unpack_from(4, ">I") def read_i64(self): return self._unpack_from(8, ">q") def read_u64(self): return self._unpack_from(8, ">Q") def read_float(self): v = self._unpack_from(4, ">f") return v def read_double(self): return self._unpack_from(8, ">d") class _UniffiRustBufferBuilder: """ Helper for structured writing of bytes into a _UniffiRustBuffer. """ def __init__(self): self.rbuf = _UniffiRustBuffer.alloc(16) self.rbuf.len = 0 def finalize(self): rbuf = self.rbuf self.rbuf = None return rbuf def discard(self): if self.rbuf is not None: rbuf = self.finalize() rbuf.free() @contextlib.contextmanager def _reserve(self, num_bytes): if self.rbuf.len + num_bytes > self.rbuf.capacity: self.rbuf = _UniffiRustBuffer.reserve(self.rbuf, num_bytes) yield None self.rbuf.len += num_bytes def _pack_into(self, size, format, value): with self._reserve(size): # XXX TODO: I feel like I should be able to use `struct.pack_into` here but can't figure it out. for i, byte in enumerate(struct.pack(format, value)): self.rbuf.data[self.rbuf.len + i] = byte def write(self, value): with self._reserve(len(value)): for i, byte in enumerate(value): self.rbuf.data[self.rbuf.len + i] = byte def write_i8(self, v): self._pack_into(1, ">b", v) def write_u8(self, v): self._pack_into(1, ">B", v) def write_i16(self, v): self._pack_into(2, ">h", v) def write_u16(self, v): self._pack_into(2, ">H", v) def write_i32(self, v): self._pack_into(4, ">i", v) def write_u32(self, v): self._pack_into(4, ">I", v) def write_i64(self, v): self._pack_into(8, ">q", v) def write_u64(self, v): self._pack_into(8, ">Q", v) def write_float(self, v): self._pack_into(4, ">f", v) def write_double(self, v): self._pack_into(8, ">d", v) def write_c_size_t(self, v): self._pack_into(ctypes.sizeof(ctypes.c_size_t) , "@N", v) # A handful of classes and functions to support the generated data structures. # This would be a good candidate for isolating in its own ffi-support lib. class InternalError(Exception): pass class _UniffiRustCallStatus(ctypes.Structure): """ Error runtime. """ _fields_ = [ ("code", ctypes.c_int8), ("error_buf", _UniffiRustBuffer), ] # These match the values from the uniffi::rustcalls module CALL_SUCCESS = 0 CALL_ERROR = 1 CALL_UNEXPECTED_ERROR = 2 @staticmethod def default(): return _UniffiRustCallStatus(code=_UniffiRustCallStatus.CALL_SUCCESS, error_buf=_UniffiRustBuffer.default()) def __str__(self): if self.code == _UniffiRustCallStatus.CALL_SUCCESS: return "_UniffiRustCallStatus(CALL_SUCCESS)" elif self.code == _UniffiRustCallStatus.CALL_ERROR: return "_UniffiRustCallStatus(CALL_ERROR)" elif self.code == _UniffiRustCallStatus.CALL_UNEXPECTED_ERROR: return "_UniffiRustCallStatus(CALL_UNEXPECTED_ERROR)" else: return "_UniffiRustCallStatus()" def _uniffi_rust_call(fn, *args): # Call a rust function return _uniffi_rust_call_with_error(None, fn, *args) def _uniffi_rust_call_with_error(error_ffi_converter, fn, *args): # Call a rust function and handle any errors # # This function is used for rust calls that return Result<> and therefore can set the CALL_ERROR status code. # error_ffi_converter must be set to the _UniffiConverter for the error class that corresponds to the result. call_status = _UniffiRustCallStatus.default() args_with_error = args + (ctypes.byref(call_status),) result = fn(*args_with_error) _uniffi_check_call_status(error_ffi_converter, call_status) return result def _uniffi_check_call_status(error_ffi_converter, call_status): if call_status.code == _UniffiRustCallStatus.CALL_SUCCESS: pass elif call_status.code == _UniffiRustCallStatus.CALL_ERROR: if error_ffi_converter is None: call_status.error_buf.free() raise InternalError("_uniffi_rust_call_with_error: CALL_ERROR, but error_ffi_converter is None") else: raise error_ffi_converter.lift(call_status.error_buf) elif call_status.code == _UniffiRustCallStatus.CALL_UNEXPECTED_ERROR: # When the rust code sees a panic, it tries to construct a _UniffiRustBuffer # with the message. But if that code panics, then it just sends back # an empty buffer. if call_status.error_buf.len > 0: msg = _UniffiFfiConverterString.lift(call_status.error_buf) else: msg = "Unknown rust panic" raise InternalError(msg) else: raise InternalError("Invalid _UniffiRustCallStatus code: {}".format( call_status.code)) def _uniffi_trait_interface_call(call_status, make_call, write_return_value): try: return write_return_value(make_call()) except Exception as e: call_status.code = _UniffiRustCallStatus.CALL_UNEXPECTED_ERROR call_status.error_buf = _UniffiFfiConverterString.lower(repr(e)) def _uniffi_trait_interface_call_with_error(call_status, make_call, write_return_value, error_type, lower_error): try: try: return write_return_value(make_call()) except error_type as e: call_status.code = _UniffiRustCallStatus.CALL_ERROR call_status.error_buf = lower_error(e) except Exception as e: call_status.code = _UniffiRustCallStatus.CALL_UNEXPECTED_ERROR call_status.error_buf = _UniffiFfiConverterString.lower(repr(e)) # Initial value and increment amount for handles. # These ensure that Python-generated handles always have the lowest bit set _UNIFFI_HANDLEMAP_INITIAL = 1 _UNIFFI_HANDLEMAP_DELTA = 2 class _UniffiHandleMap: """ A map where inserting, getting and removing data is synchronized with a lock. """ def __init__(self): # type Handle = int self._map = {} # type: Dict[Handle, Any] self._lock = threading.Lock() self._counter = _UNIFFI_HANDLEMAP_INITIAL def insert(self, obj): with self._lock: return self._insert(obj) """Low-level insert, this assumes `self._lock` is held.""" def _insert(self, obj): handle = self._counter self._counter += _UNIFFI_HANDLEMAP_DELTA self._map[handle] = obj return handle def get(self, handle): try: with self._lock: return self._map[handle] except KeyError: raise InternalError(f"_UniffiHandleMap.get: Invalid handle {handle}") def clone(self, handle): try: with self._lock: obj = self._map[handle] return self._insert(obj) except KeyError: raise InternalError(f"_UniffiHandleMap.clone: Invalid handle {handle}") def remove(self, handle): try: with self._lock: return self._map.pop(handle) except KeyError: raise InternalError(f"_UniffiHandleMap.remove: Invalid handle: {handle}") def __len__(self): return len(self._map) # Types conforming to `_UniffiConverterPrimitive` pass themselves directly over the FFI. class _UniffiConverterPrimitive: @classmethod def lift(cls, value): return value @classmethod def lower(cls, value): return value class _UniffiConverterPrimitiveInt(_UniffiConverterPrimitive): @classmethod def check_lower(cls, value): try: value = value.__index__() except Exception: raise TypeError("'{}' object cannot be interpreted as an integer".format(type(value).__name__)) if not isinstance(value, int): raise TypeError("__index__ returned non-int (type {})".format(type(value).__name__)) if not cls.VALUE_MIN <= value < cls.VALUE_MAX: raise ValueError("{} requires {} <= value < {}".format(cls.CLASS_NAME, cls.VALUE_MIN, cls.VALUE_MAX)) class _UniffiConverterPrimitiveFloat(_UniffiConverterPrimitive): @classmethod def check_lower(cls, value): try: value = value.__float__() except Exception: raise TypeError("must be real number, not {}".format(type(value).__name__)) if not isinstance(value, float): raise TypeError("__float__ returned non-float (type {})".format(type(value).__name__)) # Helper class for wrapper types that will always go through a _UniffiRustBuffer. # Classes should inherit from this and implement the `read` and `write` static methods. class _UniffiConverterRustBuffer: @classmethod def lift(cls, rbuf): with rbuf.consume_with_stream() as stream: return cls.read(stream) @classmethod def lower(cls, value): with _UniffiRustBuffer.alloc_with_builder() as builder: cls.write(value, builder) return builder.finalize() # Contains loading, initialization code, and the FFI Function declarations. # Define some ctypes FFI types that we use in the library """ Function pointer for a Rust task, which a callback function that takes a opaque pointer """ _UNIFFI_RUST_TASK = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int8) def _uniffi_future_callback_t(return_type): """ Factory function to create callback function types for async functions """ return ctypes.CFUNCTYPE(None, ctypes.c_uint64, return_type, _UniffiRustCallStatus) def _uniffi_load_indirect(): """ This is how we find and load the dynamic library provided by the component. For now we just look it up by name. """ if sys.platform == "darwin": libname = "lib{}.dylib" elif sys.platform.startswith("win"): # As of python3.8, ctypes does not seem to search $PATH when loading DLLs. # We could use `os.add_dll_directory` to configure the search path, but # it doesn't feel right to mess with application-wide settings. Let's # assume that the `.dll` is next to the `.py` file and load by full path. libname = os.path.join( os.path.dirname(__file__), "{}.dll", ) else: # Anything else must be an ELF platform - Linux, *BSD, Solaris/illumos libname = "lib{}.so" libname = libname.format("plugins_ai_coustics_uniffi") path = os.path.join(os.path.dirname(__file__), libname) lib = ctypes.cdll.LoadLibrary(path) return lib def _uniffi_check_contract_api_version(lib): # Get the bindings contract version from our ComponentInterface bindings_contract_version = 30 # Get the scaffolding contract version by calling the into the dylib scaffolding_contract_version = lib.ffi_plugins_ai_coustics_uniffi_uniffi_contract_version() if bindings_contract_version != scaffolding_contract_version: raise InternalError("UniFFI contract version mismatch: try cleaning and rebuilding your project") def _uniffi_check_api_checksums(lib): if lib.uniffi_plugins_ai_coustics_uniffi_checksum_func_model_parameters_equal() != 8734: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_constructor_enhancer_new() != 64497: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process() != 9700: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_planar() != 34627: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_with_vad() != 36609: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_credentials() != 20156: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_model_parameters() != 12347: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") if lib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_stream_info() != 15469: raise InternalError("UniFFI API checksum mismatch: try cleaning and rebuilding your project") # A ctypes library to expose the extern-C FFI definitions. # This is an implementation detail which will be called internally by the public API. _UniffiLib = _uniffi_load_indirect() _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_alloc.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_alloc.restype = _UniffiRustBuffer _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_from_bytes.argtypes = ( _UniffiForeignBytes, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_from_bytes.restype = _UniffiRustBuffer _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_free.argtypes = ( _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_free.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_reserve.argtypes = ( _UniffiRustBuffer, ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rustbuffer_reserve.restype = _UniffiRustBuffer _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK = ctypes.CFUNCTYPE(None,ctypes.c_uint64,ctypes.c_int8, ) _UNIFFI_FOREIGN_FUTURE_DROPPED_CALLBACK = ctypes.CFUNCTYPE(None,ctypes.c_uint64, ) class _UniffiForeignFutureDroppedCallbackStruct(ctypes.Structure): _fields_ = [ ("handle", ctypes.c_uint64), ("free", _UNIFFI_FOREIGN_FUTURE_DROPPED_CALLBACK), ] _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u8.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u8.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u8.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u8.restype = ctypes.c_uint8 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u8.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i8.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i8.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i8.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i8.restype = ctypes.c_int8 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i8.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i8.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u16.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u16.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u16.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u16.restype = ctypes.c_uint16 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u16.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i16.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i16.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i16.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i16.restype = ctypes.c_int16 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i16.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i16.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u32.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u32.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u32.restype = ctypes.c_uint32 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i32.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i32.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i32.restype = ctypes.c_int32 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u64.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_u64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_u64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u64.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_u64.restype = ctypes.c_uint64 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_u64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i64.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_i64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_i64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i64.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_i64.restype = ctypes.c_int64 _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_i64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_f32.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_f32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_f32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_f32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_f32.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_f32.restype = ctypes.c_float _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_f32.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_f32.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_f64.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_f64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_f64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_f64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_f64.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_f64.restype = ctypes.c_double _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_f64.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_f64.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_rust_buffer.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_rust_buffer.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_rust_buffer.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_rust_buffer.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_rust_buffer.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_rust_buffer.restype = _UniffiRustBuffer _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_rust_buffer.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_rust_buffer.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_void.argtypes = ( ctypes.c_uint64, _UNIFFI_RUST_FUTURE_CONTINUATION_CALLBACK, ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_poll_void.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_void.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_cancel_void.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_void.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_complete_void.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_void.argtypes = ( ctypes.c_uint64, ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_rust_future_free_void.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_clone_enhancer.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_clone_enhancer.restype = ctypes.c_uint64 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_free_enhancer.argtypes = ( ctypes.c_uint64, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_free_enhancer.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_func_model_parameters_equal.argtypes = ( _UniffiRustBuffer, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_func_model_parameters_equal.restype = ctypes.c_int8 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_constructor_enhancer_new.argtypes = ( _UniffiRustBuffer, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_constructor_enhancer_new.restype = ctypes.c_uint64 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_planar.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_planar.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_with_vad.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_with_vad.restype = ctypes.c_int8 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_credentials.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_credentials.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_model_parameters.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_model_parameters.restype = None _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_stream_info.argtypes = ( ctypes.c_uint64, _UniffiRustBuffer, ctypes.POINTER(_UniffiRustCallStatus), ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_stream_info.restype = None _UniffiLib.ffi_plugins_ai_coustics_uniffi_uniffi_contract_version.argtypes = ( ) _UniffiLib.ffi_plugins_ai_coustics_uniffi_uniffi_contract_version.restype = ctypes.c_uint32 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_func_model_parameters_equal.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_func_model_parameters_equal.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_constructor_enhancer_new.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_constructor_enhancer_new.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_planar.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_planar.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_with_vad.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_process_with_vad.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_credentials.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_credentials.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_model_parameters.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_model_parameters.restype = ctypes.c_uint16 _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_stream_info.argtypes = ( ) _UniffiLib.uniffi_plugins_ai_coustics_uniffi_checksum_method_enhancer_update_stream_info.restype = ctypes.c_uint16 _uniffi_check_contract_api_version(_UniffiLib) # _uniffi_check_api_checksums(_UniffiLib) # Public interface members begin here. class _UniffiFfiConverterString: @staticmethod def check_lower(value): if not isinstance(value, str): raise TypeError("argument must be str, not {}".format(type(value).__name__)) return value @staticmethod def read(buf): size = buf.read_i32() if size < 0: raise InternalError("Unexpected negative string length") utf8_bytes = buf.read(size) return utf8_bytes.decode("utf-8") @staticmethod def write(value, buf): utf8_bytes = value.encode("utf-8") buf.write_i32(len(utf8_bytes)) buf.write(utf8_bytes) @staticmethod def lift(buf): with buf.consume_with_stream() as stream: return stream.read(stream.remaining()).decode("utf-8") @staticmethod def lower(value): with _UniffiRustBuffer.alloc_with_builder() as builder: builder.write(value.encode("utf-8")) return builder.finalize() @dataclass class Credentials: def __init__(self, *, url:str, token:str): self.url = url self.token = token def __str__(self): return "Credentials(url={}, token={})".format(self.url, self.token) def __eq__(self, other): if self.url != other.url: return False if self.token != other.token: return False return True class _UniffiFfiConverterTypeCredentials(_UniffiConverterRustBuffer): @staticmethod def read(buf): return Credentials( url=_UniffiFfiConverterString.read(buf), token=_UniffiFfiConverterString.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterString.check_lower(value.url) _UniffiFfiConverterString.check_lower(value.token) @staticmethod def write(value, buf): _UniffiFfiConverterString.write(value.url, buf) _UniffiFfiConverterString.write(value.token, buf) class _UniffiFfiConverterUInt32(_UniffiConverterPrimitiveInt): CLASS_NAME = "u32" VALUE_MIN = 0 VALUE_MAX = 2**32 @staticmethod def read(buf): return buf.read_u32() @staticmethod def write(value, buf): buf.write_u32(value) class _UniffiFfiConverterUInt16(_UniffiConverterPrimitiveInt): CLASS_NAME = "u16" VALUE_MIN = 0 VALUE_MAX = 2**16 @staticmethod def read(buf): return buf.read_u16() @staticmethod def write(value, buf): buf.write_u16(value) class EnhancerModel(enum.Enum): QUAIL_L = 0 QUAIL_VF_L = 1 QUAIL_VF_S = 2 SPARROW_S = 3 ROOK_S = 4 class _UniffiFfiConverterTypeEnhancerModel(_UniffiConverterRustBuffer): @staticmethod def read(buf): variant = buf.read_i32() if variant == 1: return EnhancerModel.QUAIL_L if variant == 2: return EnhancerModel.QUAIL_VF_L if variant == 3: return EnhancerModel.QUAIL_VF_S if variant == 4: return EnhancerModel.SPARROW_S if variant == 5: return EnhancerModel.ROOK_S raise InternalError("Raw enum value doesn't match any cases") @staticmethod def check_lower(value): if value == EnhancerModel.QUAIL_L: return if value == EnhancerModel.QUAIL_VF_L: return if value == EnhancerModel.QUAIL_VF_S: return if value == EnhancerModel.SPARROW_S: return if value == EnhancerModel.ROOK_S: return raise ValueError(value) @staticmethod def write(value, buf): if value == EnhancerModel.QUAIL_L: buf.write_i32(1) if value == EnhancerModel.QUAIL_VF_L: buf.write_i32(2) if value == EnhancerModel.QUAIL_VF_S: buf.write_i32(3) if value == EnhancerModel.SPARROW_S: buf.write_i32(4) if value == EnhancerModel.ROOK_S: buf.write_i32(5) class _UniffiFfiConverterFloat32(_UniffiConverterPrimitiveFloat): @staticmethod def read(buf): return buf.read_float() @staticmethod def write(value, buf): buf.write_float(value) class _UniffiFfiConverterOptionalFloat32(_UniffiConverterRustBuffer): @classmethod def check_lower(cls, value): if value is not None: _UniffiFfiConverterFloat32.check_lower(value) @classmethod def write(cls, value, buf): if value is None: buf.write_u8(0) return buf.write_u8(1) _UniffiFfiConverterFloat32.write(value, buf) @classmethod def read(cls, buf): flag = buf.read_u8() if flag == 0: return None elif flag == 1: return _UniffiFfiConverterFloat32.read(buf) else: raise InternalError("Unexpected flag byte for optional type") @dataclass class ModelParameters: def __init__(self, *, bypass:typing.Optional[float], enhancement_level:typing.Optional[float]): self.bypass = bypass self.enhancement_level = enhancement_level def __str__(self): return "ModelParameters(bypass={}, enhancement_level={})".format(self.bypass, self.enhancement_level) def __eq__(self, other): if self.bypass != other.bypass: return False if self.enhancement_level != other.enhancement_level: return False return True class _UniffiFfiConverterTypeModelParameters(_UniffiConverterRustBuffer): @staticmethod def read(buf): return ModelParameters( bypass=_UniffiFfiConverterOptionalFloat32.read(buf), enhancement_level=_UniffiFfiConverterOptionalFloat32.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterOptionalFloat32.check_lower(value.bypass) _UniffiFfiConverterOptionalFloat32.check_lower(value.enhancement_level) @staticmethod def write(value, buf): _UniffiFfiConverterOptionalFloat32.write(value.bypass, buf) _UniffiFfiConverterOptionalFloat32.write(value.enhancement_level, buf) @dataclass class VadSettings: def __init__(self, *, speech_hold_duration:typing.Optional[float], sensitivity:typing.Optional[float], minimum_speech_duration:typing.Optional[float]): self.speech_hold_duration = speech_hold_duration self.sensitivity = sensitivity self.minimum_speech_duration = minimum_speech_duration def __str__(self): return "VadSettings(speech_hold_duration={}, sensitivity={}, minimum_speech_duration={})".format(self.speech_hold_duration, self.sensitivity, self.minimum_speech_duration) def __eq__(self, other): if self.speech_hold_duration != other.speech_hold_duration: return False if self.sensitivity != other.sensitivity: return False if self.minimum_speech_duration != other.minimum_speech_duration: return False return True class _UniffiFfiConverterTypeVadSettings(_UniffiConverterRustBuffer): @staticmethod def read(buf): return VadSettings( speech_hold_duration=_UniffiFfiConverterOptionalFloat32.read(buf), sensitivity=_UniffiFfiConverterOptionalFloat32.read(buf), minimum_speech_duration=_UniffiFfiConverterOptionalFloat32.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterOptionalFloat32.check_lower(value.speech_hold_duration) _UniffiFfiConverterOptionalFloat32.check_lower(value.sensitivity) _UniffiFfiConverterOptionalFloat32.check_lower(value.minimum_speech_duration) @staticmethod def write(value, buf): _UniffiFfiConverterOptionalFloat32.write(value.speech_hold_duration, buf) _UniffiFfiConverterOptionalFloat32.write(value.sensitivity, buf) _UniffiFfiConverterOptionalFloat32.write(value.minimum_speech_duration, buf) @dataclass class EnhancerSettings: def __init__(self, *, sample_rate:int, num_channels:int, samples_per_channel:int, model:EnhancerModel, model_parameters:ModelParameters, vad:VadSettings): self.sample_rate = sample_rate self.num_channels = num_channels self.samples_per_channel = samples_per_channel self.model = model self.model_parameters = model_parameters self.vad = vad def __str__(self): return "EnhancerSettings(sample_rate={}, num_channels={}, samples_per_channel={}, model={}, model_parameters={}, vad={})".format(self.sample_rate, self.num_channels, self.samples_per_channel, self.model, self.model_parameters, self.vad) def __eq__(self, other): if self.sample_rate != other.sample_rate: return False if self.num_channels != other.num_channels: return False if self.samples_per_channel != other.samples_per_channel: return False if self.model != other.model: return False if self.model_parameters != other.model_parameters: return False if self.vad != other.vad: return False return True class _UniffiFfiConverterTypeEnhancerSettings(_UniffiConverterRustBuffer): @staticmethod def read(buf): return EnhancerSettings( sample_rate=_UniffiFfiConverterUInt32.read(buf), num_channels=_UniffiFfiConverterUInt16.read(buf), samples_per_channel=_UniffiFfiConverterUInt32.read(buf), model=_UniffiFfiConverterTypeEnhancerModel.read(buf), model_parameters=_UniffiFfiConverterTypeModelParameters.read(buf), vad=_UniffiFfiConverterTypeVadSettings.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterUInt32.check_lower(value.sample_rate) _UniffiFfiConverterUInt16.check_lower(value.num_channels) _UniffiFfiConverterUInt32.check_lower(value.samples_per_channel) _UniffiFfiConverterTypeEnhancerModel.check_lower(value.model) _UniffiFfiConverterTypeModelParameters.check_lower(value.model_parameters) _UniffiFfiConverterTypeVadSettings.check_lower(value.vad) @staticmethod def write(value, buf): _UniffiFfiConverterUInt32.write(value.sample_rate, buf) _UniffiFfiConverterUInt16.write(value.num_channels, buf) _UniffiFfiConverterUInt32.write(value.samples_per_channel, buf) _UniffiFfiConverterTypeEnhancerModel.write(value.model, buf) _UniffiFfiConverterTypeModelParameters.write(value.model_parameters, buf) _UniffiFfiConverterTypeVadSettings.write(value.vad, buf) class _UniffiFfiConverterUInt64(_UniffiConverterPrimitiveInt): CLASS_NAME = "u64" VALUE_MIN = 0 VALUE_MAX = 2**64 @staticmethod def read(buf): return buf.read_u64() @staticmethod def write(value, buf): buf.write_u64(value) @dataclass class NativeAudioBufferMut: """ A buffer owned by and whose lifetime is managed by the foreign language. """ def __init__(self, *, ptr:int, len:int): self.ptr = ptr self.len = len def __str__(self): return "NativeAudioBufferMut(ptr={}, len={})".format(self.ptr, self.len) def __eq__(self, other): if self.ptr != other.ptr: return False if self.len != other.len: return False return True class _UniffiFfiConverterTypeNativeAudioBufferMut(_UniffiConverterRustBuffer): @staticmethod def read(buf): return NativeAudioBufferMut( ptr=_UniffiFfiConverterUInt64.read(buf), len=_UniffiFfiConverterUInt64.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterUInt64.check_lower(value.ptr) _UniffiFfiConverterUInt64.check_lower(value.len) @staticmethod def write(value, buf): _UniffiFfiConverterUInt64.write(value.ptr, buf) _UniffiFfiConverterUInt64.write(value.len, buf) @dataclass class StreamInfo: def __init__(self, *, room_id:str, room_name:str, participant_identity:str, participant_id:str, track_id:str): self.room_id = room_id self.room_name = room_name self.participant_identity = participant_identity self.participant_id = participant_id self.track_id = track_id def __str__(self): return "StreamInfo(room_id={}, room_name={}, participant_identity={}, participant_id={}, track_id={})".format(self.room_id, self.room_name, self.participant_identity, self.participant_id, self.track_id) def __eq__(self, other): if self.room_id != other.room_id: return False if self.room_name != other.room_name: return False if self.participant_identity != other.participant_identity: return False if self.participant_id != other.participant_id: return False if self.track_id != other.track_id: return False return True class _UniffiFfiConverterTypeStreamInfo(_UniffiConverterRustBuffer): @staticmethod def read(buf): return StreamInfo( room_id=_UniffiFfiConverterString.read(buf), room_name=_UniffiFfiConverterString.read(buf), participant_identity=_UniffiFfiConverterString.read(buf), participant_id=_UniffiFfiConverterString.read(buf), track_id=_UniffiFfiConverterString.read(buf), ) @staticmethod def check_lower(value): _UniffiFfiConverterString.check_lower(value.room_id) _UniffiFfiConverterString.check_lower(value.room_name) _UniffiFfiConverterString.check_lower(value.participant_identity) _UniffiFfiConverterString.check_lower(value.participant_id) _UniffiFfiConverterString.check_lower(value.track_id) @staticmethod def write(value, buf): _UniffiFfiConverterString.write(value.room_id, buf) _UniffiFfiConverterString.write(value.room_name, buf) _UniffiFfiConverterString.write(value.participant_identity, buf) _UniffiFfiConverterString.write(value.participant_id, buf) _UniffiFfiConverterString.write(value.track_id, buf) class AuthMode: """ Selects the authentication mode for the audio enhancer. """ def __init__(self): raise RuntimeError("AuthMode cannot be instantiated directly") # Each enum variant is a nested class of the enum itself. @dataclass class LIVE_KIT_CLOUD: """ Use LiveKit Cloud for authorization and usage reporting (default). """ def __init__(self, url:str, token:str): self.url = url self.token = token pass def __str__(self): return "AuthMode.LIVE_KIT_CLOUD(url={}, token={})".format(self.url, self.token) def __eq__(self, other): if not other.is_LIVE_KIT_CLOUD(): return False if self.url != other.url: return False if self.token != other.token: return False return True @dataclass class AI_COUSTICS_API: """ Use your own ai-coustics API credentials directly, bypassing LiveKit Cloud. """ def __init__(self, license_key:str): self.license_key = license_key pass def __str__(self): return "AuthMode.AI_COUSTICS_API(license_key={})".format(self.license_key) def __eq__(self, other): if not other.is_AI_COUSTICS_API(): return False if self.license_key != other.license_key: return False return True # For each variant, we have `is_NAME` and `is_name` methods for easily checking # whether an instance is that variant. def is_LIVE_KIT_CLOUD(self) -> bool: return isinstance(self, AuthMode.LIVE_KIT_CLOUD) def is_live_kit_cloud(self) -> bool: return isinstance(self, AuthMode.LIVE_KIT_CLOUD) def is_AI_COUSTICS_API(self) -> bool: return isinstance(self, AuthMode.AI_COUSTICS_API) def is_ai_coustics_api(self) -> bool: return isinstance(self, AuthMode.AI_COUSTICS_API) # Now, a little trick - we make each nested variant class be a subclass of the main # enum class, so that method calls and instance checks etc will work intuitively. # We might be able to do this a little more neatly with a metaclass, but this'll do. AuthMode.LIVE_KIT_CLOUD = type("AuthMode.LIVE_KIT_CLOUD", (AuthMode.LIVE_KIT_CLOUD, AuthMode,), {}) # type: ignore AuthMode.AI_COUSTICS_API = type("AuthMode.AI_COUSTICS_API", (AuthMode.AI_COUSTICS_API, AuthMode,), {}) # type: ignore class _UniffiFfiConverterTypeAuthMode(_UniffiConverterRustBuffer): @staticmethod def read(buf): variant = buf.read_i32() if variant == 1: return AuthMode.LIVE_KIT_CLOUD( _UniffiFfiConverterString.read(buf), _UniffiFfiConverterString.read(buf), ) if variant == 2: return AuthMode.AI_COUSTICS_API( _UniffiFfiConverterString.read(buf), ) raise InternalError("Raw enum value doesn't match any cases") @staticmethod def check_lower(value): if value.is_LIVE_KIT_CLOUD(): _UniffiFfiConverterString.check_lower(value.url) _UniffiFfiConverterString.check_lower(value.token) return if value.is_AI_COUSTICS_API(): _UniffiFfiConverterString.check_lower(value.license_key) return raise ValueError(value) @staticmethod def write(value, buf): if value.is_LIVE_KIT_CLOUD(): buf.write_i32(1) _UniffiFfiConverterString.write(value.url, buf) _UniffiFfiConverterString.write(value.token, buf) if value.is_AI_COUSTICS_API(): buf.write_i32(2) _UniffiFfiConverterString.write(value.license_key, buf) # EnhancerError # We want to define each variant as a nested class that's also a subclass, # which is tricky in Python. To accomplish this we're going to create each # class separately, then manually add the child classes to the base class's # __dict__. All of this happens in dummy class to avoid polluting the module # namespace. class EnhancerError(Exception): pass _UniffiTempEnhancerError = EnhancerError class EnhancerError: # type: ignore class Model(_UniffiTempEnhancerError): def __repr__(self): return "EnhancerError.Model({})".format(repr(str(self))) _UniffiTempEnhancerError.Model = Model # type: ignore class Authorization(_UniffiTempEnhancerError): def __repr__(self): return "EnhancerError.Authorization({})".format(repr(str(self))) _UniffiTempEnhancerError.Authorization = Authorization # type: ignore EnhancerError = _UniffiTempEnhancerError # type: ignore del _UniffiTempEnhancerError class _UniffiFfiConverterTypeEnhancerError(_UniffiConverterRustBuffer): @staticmethod def read(buf): variant = buf.read_i32() if variant == 1: return EnhancerError.Model( _UniffiFfiConverterString.read(buf), ) if variant == 2: return EnhancerError.Authorization( _UniffiFfiConverterString.read(buf), ) raise InternalError("Raw enum value doesn't match any cases") @staticmethod def check_lower(value): if isinstance(value, EnhancerError.Model): return if isinstance(value, EnhancerError.Authorization): return @staticmethod def write(value, buf): if isinstance(value, EnhancerError.Model): buf.write_i32(1) if isinstance(value, EnhancerError.Authorization): buf.write_i32(2) class _UniffiFfiConverterSequenceTypeNativeAudioBufferMut(_UniffiConverterRustBuffer): @classmethod def check_lower(cls, value): for item in value: _UniffiFfiConverterTypeNativeAudioBufferMut.check_lower(item) @classmethod def write(cls, value, buf): items = len(value) buf.write_i32(items) for item in value: _UniffiFfiConverterTypeNativeAudioBufferMut.write(item, buf) @classmethod def read(cls, buf): count = buf.read_i32() if count < 0: raise InternalError("Unexpected negative sequence length") return [ _UniffiFfiConverterTypeNativeAudioBufferMut.read(buf) for i in range(count) ] class _UniffiFfiConverterBoolean: @classmethod def check_lower(cls, value): return not not value @classmethod def lower(cls, value): return 1 if value else 0 @staticmethod def lift(value): return value != 0 @classmethod def read(cls, buf): return cls.lift(buf.read_u8()) @classmethod def write(cls, value, buf): buf.write_u8(value) class EnhancerProtocol(typing.Protocol): """ Ai-coustics audio enhancer. """ def process(self, frame: NativeAudioBufferMut) -> None: """ Process an interleaved, 10ms frame. """ raise NotImplementedError def process_planar(self, channels: typing.List[NativeAudioBufferMut]) -> None: """ Process a planar, 10ms frame. """ raise NotImplementedError def process_with_vad(self, frame: NativeAudioBufferMut) -> bool: """ Process an interleaved, 10ms frame. Returns vad information alongside the mutated frame. """ raise NotImplementedError def update_credentials(self, credentials: Credentials) -> None: """ Update credentials for model authorization. The model must always hold a set of valid credentials for continued operation. """ raise NotImplementedError def update_model_parameters(self, model_parameters: ModelParameters) -> None: raise NotImplementedError def update_stream_info(self, info: StreamInfo) -> None: """ Report information about the current audio stream being processed. """ raise NotImplementedError class Enhancer(EnhancerProtocol): """ Ai-coustics audio enhancer. """ _handle: ctypes.c_uint64 def __init__(self, auth: AuthMode,settings: EnhancerSettings): """ Creates a new audio filter with the provided settings. If provided settings are invalid or model use cannot be authorized, the result is an error. """ _UniffiFfiConverterTypeAuthMode.check_lower(auth) _UniffiFfiConverterTypeEnhancerSettings.check_lower(settings) _uniffi_lowered_args = ( _UniffiFfiConverterTypeAuthMode.lower(auth), _UniffiFfiConverterTypeEnhancerSettings.lower(settings), ) _uniffi_lift_return = _UniffiFfiConverterTypeEnhancer.lift _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_constructor_enhancer_new, *_uniffi_lowered_args, ) self._handle = _uniffi_ffi_result def __del__(self): # In case of partial initialization of instances. handle = getattr(self, "_handle", None) if handle is not None: _uniffi_rust_call(_UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_free_enhancer, handle) def _uniffi_clone_handle(self): return _uniffi_rust_call(_UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_clone_enhancer, self._handle) # Used by alternative constructors or any methods which return this type. @classmethod def _uniffi_make_instance(cls, handle): # Lightly yucky way to bypass the usual __init__ logic # and just create a new instance with the required handle. inst = cls.__new__(cls) inst._handle = handle return inst def process(self, frame: NativeAudioBufferMut) -> None: """ Process an interleaved, 10ms frame. """ _UniffiFfiConverterTypeNativeAudioBufferMut.check_lower(frame) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterTypeNativeAudioBufferMut.lower(frame), ) _uniffi_lift_return = lambda val: None _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) def process_planar(self, channels: typing.List[NativeAudioBufferMut]) -> None: """ Process a planar, 10ms frame. """ _UniffiFfiConverterSequenceTypeNativeAudioBufferMut.check_lower(channels) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterSequenceTypeNativeAudioBufferMut.lower(channels), ) _uniffi_lift_return = lambda val: None _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_planar, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) def process_with_vad(self, frame: NativeAudioBufferMut) -> bool: """ Process an interleaved, 10ms frame. Returns vad information alongside the mutated frame. """ _UniffiFfiConverterTypeNativeAudioBufferMut.check_lower(frame) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterTypeNativeAudioBufferMut.lower(frame), ) _uniffi_lift_return = _UniffiFfiConverterBoolean.lift _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_process_with_vad, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) def update_credentials(self, credentials: Credentials) -> None: """ Update credentials for model authorization. The model must always hold a set of valid credentials for continued operation. """ _UniffiFfiConverterTypeCredentials.check_lower(credentials) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterTypeCredentials.lower(credentials), ) _uniffi_lift_return = lambda val: None _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_credentials, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) def update_model_parameters(self, model_parameters: ModelParameters) -> None: _UniffiFfiConverterTypeModelParameters.check_lower(model_parameters) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterTypeModelParameters.lower(model_parameters), ) _uniffi_lift_return = lambda val: None _uniffi_error_converter = _UniffiFfiConverterTypeEnhancerError _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_model_parameters, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) def update_stream_info(self, info: StreamInfo) -> None: """ Report information about the current audio stream being processed. """ _UniffiFfiConverterTypeStreamInfo.check_lower(info) _uniffi_lowered_args = ( self._uniffi_clone_handle(), _UniffiFfiConverterTypeStreamInfo.lower(info), ) _uniffi_lift_return = lambda val: None _uniffi_error_converter = None _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_method_enhancer_update_stream_info, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) class _UniffiFfiConverterTypeEnhancer: @staticmethod def lift(value: int) -> Enhancer: return Enhancer._uniffi_make_instance(value) @staticmethod def check_lower(value: Enhancer): if not isinstance(value, Enhancer): raise TypeError("Expected Enhancer instance, {} found".format(type(value).__name__)) @staticmethod def lower(value: Enhancer) -> ctypes.c_uint64: return value._uniffi_clone_handle() @classmethod def read(cls, buf: _UniffiRustBuffer) -> Enhancer: ptr = buf.read_u64() if ptr == 0: raise InternalError("Raw handle value was null") return cls.lift(ptr) @classmethod def write(cls, value: Enhancer, buf: _UniffiRustBuffer): buf.write_u64(cls.lower(value)) class _UniffiFfiConverterUInt8(_UniffiConverterPrimitiveInt): CLASS_NAME = "u8" VALUE_MIN = 0 VALUE_MAX = 2**8 @staticmethod def read(buf): return buf.read_u8() @staticmethod def write(value, buf): buf.write_u8(value) def model_parameters_equal(a: ModelParameters,b: ModelParameters) -> bool: _UniffiFfiConverterTypeModelParameters.check_lower(a) _UniffiFfiConverterTypeModelParameters.check_lower(b) _uniffi_lowered_args = ( _UniffiFfiConverterTypeModelParameters.lower(a), _UniffiFfiConverterTypeModelParameters.lower(b), ) _uniffi_lift_return = _UniffiFfiConverterBoolean.lift _uniffi_error_converter = None _uniffi_ffi_result = _uniffi_rust_call_with_error( _uniffi_error_converter, _UniffiLib.uniffi_plugins_ai_coustics_uniffi_fn_func_model_parameters_equal, *_uniffi_lowered_args, ) return _uniffi_lift_return(_uniffi_ffi_result) __all__ = [ "InternalError", "EnhancerModel", "AuthMode", "EnhancerError", "Credentials", "ModelParameters", "VadSettings", "EnhancerSettings", "NativeAudioBufferMut", "StreamInfo", "model_parameters_equal", "Enhancer", "EnhancerProtocol", ]