1# Copyright 2023 Google LLC 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15# ----------------------------------------------------------------------------- 16# Constants 17# ----------------------------------------------------------------------------- 18# fmt: off 19 20WL = [-60, -30, 58, 172, 334, 538, 1198, 3042] 21RL42 = [0, 7, 6, 5, 4, 3, 2, 1, 7, 6, 5, 4, 3, 2, 1, 0] 22ILB = [ 23 2048, 24 2093, 25 2139, 26 2186, 27 2233, 28 2282, 29 2332, 30 2383, 31 2435, 32 2489, 33 2543, 34 2599, 35 2656, 36 2714, 37 2774, 38 2834, 39 2896, 40 2960, 41 3025, 42 3091, 43 3158, 44 3228, 45 3298, 46 3371, 47 3444, 48 3520, 49 3597, 50 3676, 51 3756, 52 3838, 53 3922, 54 4008, 55] 56WH = [0, -214, 798] 57RH2 = [2, 1, 2, 1] 58# Values in QM2/QM4/QM6 left shift three bits than original g722 specification. 59QM2 = [-7408, -1616, 7408, 1616] 60QM4 = [ 61 0, 62 -20456, 63 -12896, 64 -8968, 65 -6288, 66 -4240, 67 -2584, 68 -1200, 69 20456, 70 12896, 71 8968, 72 6288, 73 4240, 74 2584, 75 1200, 76 0, 77] 78QM6 = [ 79 -136, 80 -136, 81 -136, 82 -136, 83 -24808, 84 -21904, 85 -19008, 86 -16704, 87 -14984, 88 -13512, 89 -12280, 90 -11192, 91 -10232, 92 -9360, 93 -8576, 94 -7856, 95 -7192, 96 -6576, 97 -6000, 98 -5456, 99 -4944, 100 -4464, 101 -4008, 102 -3576, 103 -3168, 104 -2776, 105 -2400, 106 -2032, 107 -1688, 108 -1360, 109 -1040, 110 -728, 111 24808, 112 21904, 113 19008, 114 16704, 115 14984, 116 13512, 117 12280, 118 11192, 119 10232, 120 9360, 121 8576, 122 7856, 123 7192, 124 6576, 125 6000, 126 5456, 127 4944, 128 4464, 129 4008, 130 3576, 131 3168, 132 2776, 133 2400, 134 2032, 135 1688, 136 1360, 137 1040, 138 728, 139 432, 140 136, 141 -432, 142 -136, 143] 144QMF_COEFFS = [3, -11, 12, 32, -210, 951, 3876, -805, 362, -156, 53, -11] 145 146# fmt: on 147 148 149# ----------------------------------------------------------------------------- 150# Classes 151# ----------------------------------------------------------------------------- 152class G722Decoder(object): 153 """G.722 decoder with bitrate 64kbit/s. 154 155 For the Blocks in the sub-band decoders, please refer to the G.722 156 specification for the required information. G722 specification: 157 https://www.itu.int/rec/T-REC-G.722-201209-I 158 """ 159 160 def __init__(self): 161 self._x = [0] * 24 162 self._band = [Band(), Band()] 163 # The initial value in BLOCK 3L 164 self._band[0].det = 32 165 # The initial value in BLOCK 3H 166 self._band[1].det = 8 167 168 def decode_frame(self, encoded_data) -> bytearray: 169 result_array = bytearray(len(encoded_data) * 4) 170 self.g722_decode(result_array, encoded_data) 171 return result_array 172 173 def g722_decode(self, result_array, encoded_data) -> int: 174 """Decode the data frame using g722 decoder.""" 175 result_length = 0 176 177 for code in encoded_data: 178 higher_bits = (code >> 6) & 0x03 179 lower_bits = code & 0x3F 180 181 rlow = self.lower_sub_band_decoder(lower_bits) 182 rhigh = self.higher_sub_band_decoder(higher_bits) 183 184 # Apply the receive QMF 185 self._x[:22] = self._x[2:] 186 self._x[22] = rlow + rhigh 187 self._x[23] = rlow - rhigh 188 189 xout2 = sum(self._x[2 * i] * QMF_COEFFS[i] for i in range(12)) 190 xout1 = sum(self._x[2 * i + 1] * QMF_COEFFS[11 - i] for i in range(12)) 191 192 result_length = self.update_decoded_result( 193 xout1, result_length, result_array 194 ) 195 result_length = self.update_decoded_result( 196 xout2, result_length, result_array 197 ) 198 199 return result_length 200 201 def update_decoded_result(self, xout, byte_length, byte_array) -> int: 202 result = (int)(xout >> 11) 203 bytes_result = result.to_bytes(2, 'little', signed=True) 204 byte_array[byte_length] = bytes_result[0] 205 byte_array[byte_length + 1] = bytes_result[1] 206 return byte_length + 2 207 208 def lower_sub_band_decoder(self, lower_bits) -> int: 209 """Lower sub-band decoder for last six bits.""" 210 211 # Block 5L 212 # INVQBL 213 wd1 = lower_bits 214 wd2 = QM6[wd1] 215 wd1 >>= 2 216 wd2 = (self._band[0].det * wd2) >> 15 217 # RECONS 218 rlow = self._band[0].s + wd2 219 220 # Block 6L 221 # LIMIT 222 if rlow > 16383: 223 rlow = 16383 224 elif rlow < -16384: 225 rlow = -16384 226 227 # Block 2L 228 # INVQAL 229 wd2 = QM4[wd1] 230 dlowt = (self._band[0].det * wd2) >> 15 231 232 # Block 3L 233 # LOGSCL 234 wd2 = RL42[wd1] 235 wd1 = (self._band[0].nb * 127) >> 7 236 wd1 += WL[wd2] 237 238 if wd1 < 0: 239 wd1 = 0 240 elif wd1 > 18432: 241 wd1 = 18432 242 243 self._band[0].nb = wd1 244 245 # SCALEL 246 wd1 = (self._band[0].nb >> 6) & 31 247 wd2 = 8 - (self._band[0].nb >> 11) 248 249 if wd2 < 0: 250 wd3 = ILB[wd1] << -wd2 251 else: 252 wd3 = ILB[wd1] >> wd2 253 254 self._band[0].det = wd3 << 2 255 256 # Block 4L 257 self._band[0].block4(dlowt) 258 259 return rlow 260 261 def higher_sub_band_decoder(self, higher_bits) -> int: 262 """Higher sub-band decoder for first two bits.""" 263 264 # Block 2H 265 # INVQAH 266 wd2 = QM2[higher_bits] 267 dhigh = (self._band[1].det * wd2) >> 15 268 269 # Block 5H 270 # RECONS 271 rhigh = dhigh + self._band[1].s 272 273 # Block 6H 274 # LIMIT 275 if rhigh > 16383: 276 rhigh = 16383 277 elif rhigh < -16384: 278 rhigh = -16384 279 280 # Block 3H 281 # LOGSCH 282 wd2 = RH2[higher_bits] 283 wd1 = (self._band[1].nb * 127) >> 7 284 wd1 += WH[wd2] 285 286 if wd1 < 0: 287 wd1 = 0 288 elif wd1 > 22528: 289 wd1 = 22528 290 self._band[1].nb = wd1 291 292 # SCALEH 293 wd1 = (self._band[1].nb >> 6) & 31 294 wd2 = 10 - (self._band[1].nb >> 11) 295 296 if wd2 < 0: 297 wd3 = ILB[wd1] << -wd2 298 else: 299 wd3 = ILB[wd1] >> wd2 300 self._band[1].det = wd3 << 2 301 302 # Block 4H 303 self._band[1].block4(dhigh) 304 305 return rhigh 306 307 308# ----------------------------------------------------------------------------- 309class Band(object): 310 """Structure for G722 decode proccessing.""" 311 312 s: int = 0 313 nb: int = 0 314 det: int = 0 315 316 def __init__(self): 317 self._sp = 0 318 self._sz = 0 319 self._r = [0] * 3 320 self._a = [0] * 3 321 self._ap = [0] * 3 322 self._p = [0] * 3 323 self._d = [0] * 7 324 self._b = [0] * 7 325 self._bp = [0] * 7 326 self._sg = [0] * 7 327 328 def saturate(self, amp: int) -> int: 329 if amp > 32767: 330 return 32767 331 elif amp < -32768: 332 return -32768 333 else: 334 return amp 335 336 def block4(self, d: int) -> None: 337 """Block4 for both lower and higher sub-band decoder.""" 338 wd1 = 0 339 wd2 = 0 340 wd3 = 0 341 342 # RECONS 343 self._d[0] = d 344 self._r[0] = self.saturate(self.s + d) 345 346 # PARREC 347 self._p[0] = self.saturate(self._sz + d) 348 349 # UPPOL2 350 for i in range(3): 351 self._sg[i] = (self._p[i]) >> 15 352 wd1 = self.saturate((self._a[1]) << 2) 353 wd2 = -wd1 if self._sg[0] == self._sg[1] else wd1 354 355 if wd2 > 32767: 356 wd2 = 32767 357 358 wd3 = 128 if self._sg[0] == self._sg[2] else -128 359 wd3 += wd2 >> 7 360 wd3 += (self._a[2] * 32512) >> 15 361 362 if wd3 > 12288: 363 wd3 = 12288 364 elif wd3 < -12288: 365 wd3 = -12288 366 self._ap[2] = wd3 367 368 # UPPOL1 369 self._sg[0] = (self._p[0]) >> 15 370 self._sg[1] = (self._p[1]) >> 15 371 wd1 = 192 if self._sg[0] == self._sg[1] else -192 372 wd2 = (self._a[1] * 32640) >> 15 373 374 self._ap[1] = self.saturate(wd1 + wd2) 375 wd3 = self.saturate(15360 - self._ap[2]) 376 377 if self._ap[1] > wd3: 378 self._ap[1] = wd3 379 elif self._ap[1] < -wd3: 380 self._ap[1] = -wd3 381 382 # UPZERO 383 wd1 = 0 if d == 0 else 128 384 self._sg[0] = d >> 15 385 for i in range(1, 7): 386 self._sg[i] = (self._d[i]) >> 15 387 wd2 = wd1 if self._sg[i] == self._sg[0] else -wd1 388 wd3 = (self._b[i] * 32640) >> 15 389 self._bp[i] = self.saturate(wd2 + wd3) 390 391 # DELAYA 392 for i in range(6, 0, -1): 393 self._d[i] = self._d[i - 1] 394 self._b[i] = self._bp[i] 395 396 for i in range(2, 0, -1): 397 self._r[i] = self._r[i - 1] 398 self._p[i] = self._p[i - 1] 399 self._a[i] = self._ap[i] 400 401 # FILTEP 402 self._sp = 0 403 for i in range(1, 3): 404 wd1 = self.saturate(self._r[i] + self._r[i]) 405 self._sp += (self._a[i] * wd1) >> 15 406 self._sp = self.saturate(self._sp) 407 408 # FILTEZ 409 self._sz = 0 410 for i in range(6, 0, -1): 411 wd1 = self.saturate(self._d[i] + self._d[i]) 412 self._sz += (self._b[i] * wd1) >> 15 413 self._sz = self.saturate(self._sz) 414 415 # PREDIC 416 self.s = self.saturate(self._sp + self._sz) 417