sources/asr-reviewed/src/utils/wire_packet.py

Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл

   1 from __future__ import annotations
   2 
   3 import json
   4 from typing import Any
   5 
   6 import numpy as np
   7 
   8 
   9 def parse_wire_packet(payload: bytes) -> tuple[dict[str, Any], np.ndarray]:
  10     """
  11     Parse `[meta_len:u32 BE][meta_json UTF-8][pcm int16 LE bytes]`.
  12 
  13     Returns `(meta, audio_float32)` where audio is normalized to [-1, 1].
  14     PCM may be empty (e.g. for control packets) and will be returned as an empty array.
  15     """
  16     if len(payload) < 4:
  17         raise ValueError("Payload is too short to contain metadata")
  18 
  19     meta_len = int.from_bytes(payload[:4], "big")
  20     meta_bytes = payload[4 : 4 + meta_len]
  21     if len(meta_bytes) != meta_len:
  22         raise ValueError("Invalid metadata length in packet")
  23 
  24     try:
  25         meta = json.loads(meta_bytes.decode("utf-8"))
  26     except json.JSONDecodeError as exc:
  27         raise ValueError("Invalid metadata JSON") from exc
  28 
  29     pcm_bytes = payload[4 + meta_len :]
  30     if not pcm_bytes:
  31         return meta, np.zeros((0,), dtype=np.float32)
  32 
  33     audio = np.frombuffer(pcm_bytes, dtype="<i2").astype(np.float32) / 32768.0
  34     return meta, audio
  35