"""Frame builders for the two byte transports. * **BLE** — ``communication_center/bluetooth/bluetooth_frame.dart`` (see ``docs/RE_DART_FINDINGS.md`` §9). Frame layout and magics are confirmed; the *exact* on-air order of the magic relative to the frame and the MTU continuation scheme still need a live capture. * **Local TCP socket** — ``communication_center/socket`` uses ``int16ToBytes`` / ``bytesToInt16`` (little-endian) framing (see §8). A length-prefixed frame is the working candidate; adjust once a capture is available. """ from __future__ import annotations import json from typing import List, Optional, Tuple from .crc import crc16_hex_string, crc16_modbus, int16_to_le_bytes, le_bytes_to_int16 #: First two bytes of a received **ACK** frame. MAGIC_ACK = b"\x4a\xa4" #: First two bytes of a **data** frame (receiving one triggers the app's ack). MAGIC_DATA = b"\x5a\xa5" MAGIC_NAMES = { MAGIC_ACK[0:2]: "ack", MAGIC_DATA[0:2]: "data", } # In createBluetoothFrame the CRC placeholder sits at byte offsets 2..5 of the # frame body (content ++ [00 00 00 00] ++ int16LE(totalLen) ++ int16LE(value) ++ sn). _CRC_SLOTS = slice(2, 4) _FRAME_TAIL_SIZE = 4 + 2 + 2 # crc(4) + totalLen(2) + value(2) def create_ble_frame(content: bytes, value: int, sn: str) -> bytes: """``createBluetoothFrame(content, ?, value)`` — BLE frame *body* (no magic). Layout:: content ++ [crcHi crcLo 00 00] ++ int16LE(totalLen) ++ int16LE(value) ++ sn * ``totalLen = len(content) + 6 + len(sn)`` (Dart source). * The CRC is computed over the buffer with a ``00 00 00 00`` placeholder; ``Crc16::calculate`` yields a 4-hex-digit (hi/lo-swapped) string whose two bytes are written into placeholder indices 2..3. """ sn_bytes = sn.encode("utf-8") total_len = len(content) + 6 + len(sn_bytes) buf = bytearray( content + b"\x00\x00\x00\x00" + int16_to_le_bytes(total_len) + int16_to_le_bytes(value) + sn_bytes ) crc_hex = crc16_hex_string(bytes(buf)) buf[_CRC_SLOTS] = bytes.fromhex(crc_hex) return bytes(buf) def build_ble_data_frame(content: bytes, value: int, sn: str) -> bytes: """Full on-air data frame: ``0x5A 0xA5`` magic + frame body.""" return MAGIC_DATA + create_ble_frame(content, value, sn) def build_ble_ack_frame(content: bytes, value: int, sn: str) -> bytes: """Full on-air ACK frame: ``0x4A 0xA4`` magic + frame body. The exact ACK ``content``/``value`` the app emits are not confirmed — pass whatever the capture shows (typically the SN or an empty body). """ return MAGIC_ACK + create_ble_frame(content, value, sn) def magic_in_data(data: bytes) -> Optional[str]: """Return ``"ack"`` / ``"data"`` if *data* starts with a known magic.""" return MAGIC_NAMES.get(bytes(data[:2])) def sn_in_data(data: bytes, sn: str) -> bool: """True if the frame (after any magic) ends with the device SN.""" sn_bytes = sn.encode("utf-8") return data.endswith(sn_bytes) def content_in_data(data: bytes, sn: str) -> Optional[bytes]: """Extract the ``content`` bytes of a data/ack frame. Layout after the 2-byte magic is fixed-size except for ``content``, so:: content = frame[2 : len(frame) - len(sn) - FRAME_TAIL_SIZE] """ body = data[2:] if len(data) >= 2 else data sn_bytes = sn.encode("utf-8") if not body.endswith(sn_bytes): return None end = len(body) - len(sn_bytes) - _FRAME_TAIL_SIZE if end < 2: # need at least the 4-byte crc slots return None return body[2:end] def verify_frame_crc(data: bytes) -> bool: """Recompute the CRC-16 over the frame body and compare slots 2..3.""" body = data[2:] if len(data) >= 2 else data if len(body) < 4: return False candidate = body[:] candidate[2:4] = b"\x00\x00\x00\x00" return bytes.fromhex(crc16_hex_string(bytes(candidate))) == body[2:4] def chunk_for_mtu(frame: bytes, mtu: int = 20) -> List[bytes]: """Split a frame for GATT writes. First chunk carries the 2-byte magic; the rest are raw continuation chunks. The app's ``mtuToData`` does something similar — the exact continuation framing (length headers? re-magic?) needs confirmation. """ if mtu <= 2: raise ValueError("mtu must exceed the 2-byte magic") if len(frame) <= mtu: return [frame] return [frame[i:i + mtu] for i in range(0, len(frame), mtu)] def magicEquals(a: bytes, b: bytes) -> bool: """``magicEquals`` — first two bytes equal.""" return bytes(a[:2]) == bytes(b[:2]) # --------------------------------------------------------------------- socket def frame_socket_payload(payload: bytes) -> bytes: """Local-socket frame: the app sends the encrypted envelope **raw**. ``ClientSocket.send`` forwards its argument straight to ``dart:io _Socket.add`` with no length prefix, markers, or handshake (see §8); ``SocketManager._sendData`` only clears the receive buffer afterwards. This helper is therefore the identity (kept so call sites read like the app). """ return bytes(payload) class SocketFrameDecoder: """Accumulating socket receiver (``SocketManager._onSocketReceiveData``). Each incoming chunk is appended to a growable buffer and ``decrypt_data`` is attempted over the whole buffer (inside try/catch in the app); a non-empty decoded string that parses as JSON is one complete reply, at which point the buffer is cleared. Use :meth:`feed` for raw chunks and :meth:`feed_json` when you want parsed envelopes back. """ def __init__(self) -> None: self._buf = bytearray() def feed(self, data: bytes) -> List[str]: """Append *data*; return a list of complete decoded JSON strings.""" from .crypto import decrypt_data # local import: cheap, no cycles self._buf.extend(data) decoded = decrypt_data(bytes(self._buf)) if not decoded: return [] try: json.loads(decoded) except ValueError: return [] del self._buf[:] return [decoded] # Backwards-compatible alias for the old length-prefixed probing API. def feed_frames(self, data: bytes) -> List[bytes]: """Append *data*; return complete **raw wire** buffers (legacy).""" self._buf.extend(data) if not self._buf: return [] out = [bytes(self._buf)] del self._buf[:] return out