qiXddlZddlmZddlmZddlmZmZmZm Z m Z ddl m Z ejZdZdZd ezZd Zd Ze e d e d e e e de de eededeeededeeededeiZdZdZGddZd dedededej4fdZd dej4dededej4fdZy)!N) ArrayLike)soxr_ext)QQLQMQHQVHQ)versionizJInput should be a `np.ndarray` with matching dtype for ResampleStream({}).z2Channel num({}) out of limit. Should be in [1, %d]zAData type must be one of [float32, float64, int16, int32], not {}z,Quality must be one of [QQ, LQ, MQ, HQ, VHQ]vhqsoxr_vhqhqsoxr_hqmqsoxr_mqlqsoxr_lqqqsoxr_qqct|tr|j} t|S#tt f$rt twxYwN) isinstancestrlower_QUALITY_ENUM_DICTKeyError TypeError ValueError_QUALITY_ERR_STR)qs ?/opt/pipecat/venv/lib/python3.12/site-packages/soxr/__init__.py_quality_to_enumr"&sG!S GGI+!!$$ i +)**+s +A cV|tjk(rtjS|tjk(rtj S|tj k(rtjS|tjk(rtjSttj|r) npfloat32rSOXR_FLOAT32_Ifloat64SOXR_FLOAT64_Iint32 SOXR_INT32_Iint16 SOXR_INT16_Ir_DTYPE_ERR_STRformat)ntypes r!_to_soxr_datatyper00sx  &&& "** &&& "(( $$$ "(( $$$--e455ceZdZdZ d dededefdZddejdejfdZ defd Z defd Z dd Z y )ResampleStreama? Streaming resampler Use `ResampleStream` for real-time processing or very long signal. Parameters ---------- in_rate : float Input sample-rate. out_rate : float Output sample-rate. num_channels : int Number of channels. dtype : type or str, optional Internal data type processed with. Should be one of float32, float64, int16, int32. quality : int or str, optional Quality setting. One of `QQ`, `LQ`, `MQ`, `HQ`, `VHQ`. in_rateout_rate num_channelsc|dks|dkr td|dks t|krttj|t j ||_t|j }t|}tj||||||_ t|jd|j |_ y)NrSample rate should be over 0rprocess_)r _CH_LIMIT_CH_EXEED_ERR_STRr.r$dtype_typer0r"rCSoxr_csoxrgetattr_process)selfr4r5r6r<qualitystyper s r!__init__zResampleStream.__init__Rs a<8q=;< < ! y<7.55lCD DXXe_ !$**- W %nnWh eQO  x |-DE r1xreturnct|tjk7s|j|jk7r(t t j|j|jdk(r>|j|ddtjf|}tj|dS|jdk(r|j||Std)a/ Resample chunk with streaming resampler Parameters ---------- x : np.ndarray Input array. Input can be mono(1D) or multi-channel(2D of [frame, channel]). dtype should match with constructor. last : bool, optional Set True at final chunk to flush last outputs. It should be `True` only once at the end of a continuous sequence. Returns ------- np.ndarray Resampled data. Output is np.ndarray with same ndim with input. rNaxisInput must be 1-D or 2-D array) typer$ndarrayr<r=r_DTYPE_UNMATCH_ERR_STRr.ndimrAnewaxissqueezer)rBrFlastys r!resample_chunkzResampleStream.resample_chunkcs( 7bjj AGGtzz$9299$**EF F 66Q; a2:: .5A::aa( ( VVq[==D) )=> >r1c6|jjS)zx Clip counter. (for int I/O) Returns ------- int Count of clipped samples. )r? num_clipsrBs r!rWzResampleStream.num_clipss{{$$&&r1c6|jjS)z Get current delay. SoXR output has an algorithmic delay. This function returns the length of current pending output. Returns ------- float Current delay in output samples. )r?delayrXs r!rZzResampleStream.delays{{  ""r1Nc8|jjy)zv Reset resampler. Ready for fresh signal, same config. This can be used to save initialization time. N)r?clearrXs r!r\zResampleStream.clears r1)r%r )F)rGN) __name__ __module__ __qualname____doc__floatintrEr$rNrUrWrZr\r1r!r3r3=se,+/FF+0F@CF"? ?2::?>'3' #u #r1r3rFr4r5rGc*|dks|dkr tdt|tjk7r%tj|tj } |j d|jk(rttd|j}nttd|j}t|}|j dk(r7||||ddtj"f|}tj$|dS|j d k(rF|j&d}|dks t(|krtt*j||||||Std #t$r)ttj|jwxYw) a Resample signal Parameters ---------- x : array_like Input array. Input can be mono(1D) or multi-channel(2D of [frame, channel]). If input is not `np.ndarray`, it will be converted to `np.ndarray(dtype='float32')`. Its dtype should be one of float32, float64, int16, int32. in_rate : float Input sample-rate. out_rate : float Output sample-rate. quality : int or str, optional Quality setting. One of `QQ`, `LQ`, `MQ`, `HQ`, `VHQ`. Returns ------- np.ndarray Resampled data. Output is `np.ndarray` with same ndim and dtype with input. rr8)r<csoxr_split_ch_csoxr_divide_proc_rNrIrKrL)rrMr$rNasarrayr%stridesitemsizer@rr<AttributeErrorrr-r.r"rPrQrRshaper:r;)rFr4r5rC divide_procr rTr6s r!resamplermsT.!|x1}788 Aw"** JJq +8 99Q<1:: %!(oaggY,GHK!(.@ ,JKK !Avv{ 1Q ]+;Q ?zz!!$$ 1wwqz ! y<7.55lCD D7Ha339:: 8--agg6778s AE 2Fcn ttd|j}|jdk(r@||||ddtjft|}tj|dS||||t|S#t$r)t t j |jwxYw)z Resample using libsoxr's `soxr_oneshot()`. Use `resample()` for general use. `soxr_oneshot()` becomes slow with long input. This function exists for test purpose. csoxr_oneshot_rNrI) r@rr<rjrr-r.rPr$rQr"rR)rFr4r5rConeshotrTs r!_resample_oneshotrqs 8(nQWWI$>? vv{ GXqBJJ'79I'9R Szz!!$$ 7Ha)9')B CC 8--agg6778s B2B4)r )numpyr$ numpy.typingrrrrrr r _versionr __version__libsoxr_version__libsoxr_version__r:rOr;r-rrr"r0r3rarNrmrqrcr1r!rys")),/h..0  eH9TTAeS*cD"iD"iD"iD"i + 6``F1; 1;E1;U1;RZZ1;hDDeDuDWYWaWaDr1