o Si2 @sdZddlZddlZddlZddlZddlZddlmZm Z   ddej de d e d e d e d ej f d dZGdddZGdddZddZedkrOedSdS)zArithmetic coder.N) BitPacker BitUnpacker:0yE>Tpdftotal_range_bitsroundoff min_rangecheckreturnc Cs|}|r|||}d|}t|}|||}|dks$Jdd|||}||7}tj|dd} |dkrCtd|rm| dd|ksSJ| d| dd| dd|ksi| d|krmtd | S) a}Turn the given PDF into a quantized CDF that splits [0, 2 ** self.total_range_bits - 1] into chunks of size roughly proportional to the PDF. Args: pdf (torch.Tensor): probability distribution, shape should be `[N]`. total_range_bits (int): see `ArithmeticCoder`, the typical range we expect during the coding process is `[0, 2 ** total_range_bits - 1]`. roundoff (float): will round the pdf up to that level to remove difference coming from e.g. evaluating the Language Model on different architectures. min_range (int): minimum range width. Should always be at least 2 for numerical stability. Use this to avoid pathological behavior is a value that is expected to be rare actually happens in real life. check (bool): if True, checks that nothing bad happened, can be deactivated for speed. rzyou must reduce min_rangedimzmin_range must be at least 2.Nrz(You must increase your total_range_bits.)detachfloorlenlongtorchcumsum ValueErrorany) rrrr r total_range cardinalityalpharanges quantized_cdfrK/home/ubuntu/.local/lib/python3.10/site-packages/encodec/quantization/ac.pybuild_stable_quantized_cdfs" ,rc@s`eZdZdZddejedefddZe defdd Z d d Z d ed e j fddZddZdS)ArithmeticCodera ArithmeticCoder, Let us take a distribution `p` over `N` symbols, and assume we have a stream of random variables `s_t` sampled from `p`. Let us assume that we have a budget of `B` bits that we can afford to write on device. There are `2**B` possible numbers, corresponding to the range `[0, 2 ** B - 1]`. We can map each of those number to a single sequence `(s_t)` by doing the following: 1) Initialize the current range to` [0 ** 2 B - 1]`. 2) For each time step t, split the current range into contiguous chunks, one for each possible outcome, with size roughly proportional to `p`. For instance, if `p = [0.75, 0.25]`, and the range is `[0, 3]`, the chunks would be `{[0, 2], [3, 3]}`. 3) Select the chunk corresponding to `s_t`, and replace the current range with this. 4) When done encoding all the values, just select any value remaining in the range. You will notice that this procedure can fail: for instance if at any point in time the range is smaller than `N`, then we can no longer assign a non-empty chunk to each possible outcome. Intuitively, the more likely a value is, the less the range width will reduce, and the longer we can go on encoding values. This makes sense: for any efficient coding scheme, likely outcomes would take less bits, and more of them can be coded with a fixed budget. In practice, we do not know `B` ahead of time, but we have a way to inject new bits when the current range decreases below a given limit (given by `total_range_bits`), without having to redo all the computations. If we encode mostly likely values, we will seldom need to inject new bits, but a single rare value can deplete our stock of entropy! In this explanation, we assumed that the distribution `p` was constant. In fact, the present code works for any sequence `(p_t)` possibly different for each timestep. We also assume that `s_t ~ p_t`, but that doesn't need to be true, although the smaller the KL between the true distribution and `p_t`, the most efficient the coding will be. Args: fo (IO[bytes]): file-like object to which the bytes will be written to. total_range_bits (int): the range `M` described above is `2 ** total_range_bits. Any time the current range width fall under this limit, new bits will be injected to rescale the initial range. forcCsB|dksJ||_td|d|_d|_d|_d|_g|_g|_dS)Nr bitsr"rr )rrpackerlowhighmax_bit_dbg_dbg2selfr"rrrr__init__`s  zArithmeticCoder.__init__r cC|j|jdS)zReturn the current range width.r r(r'r-rrrdeltajszArithmeticCoder.deltacCs|j|jksJ|j|jf|jd|jdksJ|jdkrq|j|j?}|j|j?}||krh|j||j>8_|j||j>8_|j|jksSJ|j|j|jf|jdksZJ|jd8_|j|ndS|jdksdSdS)Nrr r)r(r'r)r&pushr-b1b2rrr_flush_common_prefixos    z$ArithmeticCoder._flush_common_prefixsymbolrcCs^|jd|jkr&|jd9_|jdd|_|jd7_|jd|jks|dkr,dn||d}||d}tt||jd|j}tt ||jd|j}|j|jksbJ|j||_|j||_|j|jks|J||||f|j |j|jf|j |j|jf| }|j|jksJ|jdksJ|jdksJ|j|S)aPush the given symbol on the stream, flushing out bits if possible. Args: symbol (int): symbol to encode with the AC. quantized_cdf (torch.Tensor): use `build_stable_quantized_cdf` to build this from your pdf estimate. rr rr =)r2rr'r(r)itemintmathceilrr*appendr+r7)r-r8r range_low range_high effective_loweffective_highoutsrrrr3s(   zArithmeticCoder.pushcCsL|jdkr|j|j?d@}|j||jd8_|jdks|jdS)z7Flush the remaining information to the stream. rr N)r)r'r&r3flush)r-r5rrrrDs   zArithmeticCoder.flushNr!)__name__ __module__ __qualname____doc__tpIObytesr;r.propertyr2r7rTensorr3rDrrrrr 8s'  r c@s^eZdZdZddejedefddZe defdd Z d d Z d e j dejefd dZdS)ArithmeticDecoderanArithmeticDecoder, see `ArithmeticCoder` for a detailed explanation. Note that this must be called with **exactly** the same parameters and sequence of quantized cdf as the arithmetic encoder or the wrong values will be decoded. If the AC encoder current range is [L, H], with `L` and `H` having the some common prefix (i.e. the same most significant bits), then this prefix will be flushed to the stream. For instances, having read 3 bits `b1 b2 b3`, we know that `[L, H]` is contained inside `[b1 b2 b3 0 ... 0 b1 b3 b3 1 ... 1]`. Now this specific sub-range can only be obtained for a specific sequence of symbols and a binary-search allows us to decode those symbols. At some point, the prefix `b1 b2 b3` will no longer be sufficient to decode new symbols, and we will need to read new bits from the stream and repeat the process. r!r"rcCsB||_d|_d|_d|_d|_td|d|_g|_g|_d|_ dS)Nrr r r$) rr'r(currentr)runpackerr*r+_lastr,rrrr.s zArithmeticDecoder.__init__r cCr/)Nr r0r1rrrr2szArithmeticDecoder.deltacCs|jdkrS|j|j?}|j|j?}||krJ|j||j>8_|j||j>8_|j||j>8_|j|jks;J|jdksBJ|jd8_ndS|jdksdSdS)Nrr )r)r'r(rPr4rrrr7s   z&ArithmeticDecoder._flush_common_prefixrcsjdjkr9j}|durdSjd9_jdd_jd|_jd7_jdjksdtdtffdd jjjjf_ dt d\}___j jjjf j jjjf|S) apPull a symbol, reading as many bits from the stream as required. This returns `None` when the stream has been exhausted. Args: quantized_cdf (torch.Tensor): use `build_stable_quantized_cdf` to build this from your pdf estimate. This must be **exatly** the same cdf as the one used at encoding time. rNr low_idxhigh_idxc s||krtd||d}|dkr|dnd}|d}tt|jdj}tt|jdj}|j}|j}j |krdj |kr]|||j fS|d|S||dS)NzBinary search failedrrr ) RuntimeErrorr:r;r<r=r2rrr'rP) rSrTmidr?r@rArBr'r( bin_searchrr-rrrXs     z*ArithmeticDecoder.pull..bin_searchr)r2rrQpullr'r(rPr)r;rRrr*r>r7r+)r-rbitsymrrWrrYs    zArithmeticDecoder.pullNrE)rFrGrHrIrJrKrLr;r.rMr2r7rrNOptionalrYrrrrrOs rOcCs&tdtdtdD]}g}td}tdd}t}t|}g}t|D]*}tj t |dd}| |t ||j } t|d} | | || | q+||dt|} tt||D]\} \}} t ||j } | | } | | ksJ| qj| tddusJqdS) Niidirrr )r manual_seedrandomseedrange randrangeioBytesIOr softmaxrandnr>rr multinomialr:r3rDseekrO enumerateziprYzeros)_pdfsrstepsr"encodersymbolssteprq_cdfr8decoderidxdecoded_symbolrrrtests2            rw__main__)rrT)rIrdr<r`typingrJrbinaryrrrNr;floatboolrr rOrwrFrrrrs0 &r]