Spaces:
Running
Running
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| import struct | |
| from .constants import HEADER_SIZE, MAGIC, PROTOCOL_VERSION, Codec, MessageType | |
| _HEADER = struct.Struct(">2sBBBBH I Q I") | |
| class BinaryFrameError(ValueError): | |
| pass | |
| class BinaryAudioFrame: | |
| message_type: int | |
| codec: int | |
| flags: int | |
| sample_rate: int | |
| sequence: int | |
| timestamp_us: int | |
| payload: bytes | |
| def encode(self) -> bytes: | |
| return encode_frame(self) | |
| def encode_frame(frame: BinaryAudioFrame) -> bytes: | |
| if frame.codec != int(Codec.PCM16): | |
| raise BinaryFrameError("protocol v1 supports only PCM16") | |
| if not 0 <= frame.flags <= 255: | |
| raise BinaryFrameError("flags out of range") | |
| if not 1 <= frame.sample_rate <= 192000: | |
| raise BinaryFrameError("sample_rate out of range") | |
| if not 0 <= frame.sequence <= 0xFFFFFFFF: | |
| raise BinaryFrameError("sequence out of range") | |
| if not 0 <= frame.timestamp_us <= 0xFFFFFFFFFFFFFFFF: | |
| raise BinaryFrameError("timestamp_us out of range") | |
| payload = bytes(frame.payload) | |
| header = _HEADER.pack( | |
| MAGIC, | |
| PROTOCOL_VERSION, | |
| frame.message_type, | |
| frame.codec, | |
| frame.flags, | |
| frame.sample_rate, | |
| frame.sequence, | |
| frame.timestamp_us, | |
| len(payload), | |
| ) | |
| assert len(header) == HEADER_SIZE | |
| return header + payload | |
| def decode_frame(data: bytes | bytearray | memoryview) -> BinaryAudioFrame: | |
| raw = bytes(data) | |
| if len(raw) < HEADER_SIZE: | |
| raise BinaryFrameError("frame shorter than 24-byte header") | |
| magic, version, message_type, codec, flags, sample_rate, sequence, timestamp_us, payload_len = ( | |
| _HEADER.unpack(raw[:HEADER_SIZE]) | |
| ) | |
| if magic != MAGIC: | |
| raise BinaryFrameError("bad magic") | |
| if version != PROTOCOL_VERSION: | |
| raise BinaryFrameError("unsupported protocol version") | |
| if codec != int(Codec.PCM16): | |
| raise BinaryFrameError("protocol v1 supports only PCM16") | |
| if message_type not in {int(MessageType.AUDIO_INPUT), int(MessageType.AUDIO_OUTPUT)}: | |
| raise BinaryFrameError("unknown message type") | |
| if len(raw) - HEADER_SIZE != payload_len: | |
| raise BinaryFrameError("payload length mismatch") | |
| return BinaryAudioFrame( | |
| message_type, codec, flags, sample_rate, sequence, timestamp_us, raw[HEADER_SIZE:] | |
| ) | |