NODEDC_MISSION_CORE/plugins/insta360-x4/packaging/ble_options.py

"""Read-only Options codec derived from the pinned SDK's descriptors/serializer.

This establishes packet structure, not BLE support on a particular X4 firmware.
No transport, pairing, setting changes, capture, reboot or generic send API.
"""

import struct

MAX_PACKET = 4096
READ_OPTIONS = frozenset({15, 48, 95, 97})


def varint(value):
    if type(value) is not int or not 0 <= value < 2**64:
        raise ValueError("Invalid protobuf integer")
    result = bytearray()
    while value >= 128:
        result.append((value & 127) | 128)
        value >>= 7
    result.append(value)
    return bytes(result)


def read_varint(data, offset):
    value = 0
    for shift in range(0, 70, 7):
        if offset >= len(data):
            raise ValueError("Truncated varint")
        byte = data[offset]
        offset += 1
        if shift == 63 and byte > 1:
            raise ValueError("Overflowed varint")
        value |= (byte & 127) << shift
        if not byte & 128:
            return value, offset
    raise ValueError("Oversized varint")


def protobuf_fields(data):
    if len(data) > MAX_PACKET:
        raise ValueError("Oversized protobuf")
    result, offset = [], 0
    while offset < len(data):
        tag, offset = read_varint(data, offset)
        number, wire = tag >> 3, tag & 7
        if not 1 <= number <= 0x1FFFFFFF:
            raise ValueError("Invalid field number")
        if wire == 0:
            value, offset = read_varint(data, offset)
        elif wire in {1, 2, 5}:
            if wire == 2:
                size, offset = read_varint(data, offset)
            else:
                size = 8 if wire == 1 else 4
            if size > len(data) - offset:
                raise ValueError("Truncated field")
            value, offset = data[offset : offset + size], offset + size
        else:
            raise ValueError("Unsupported protobuf wire type")
        result.append((number, wire, value))
    return result


def get_options_packet(options, message_id):
    if (
        not options
        or len(options) > 4
        or len(set(options)) != len(options)
        or any(type(x) is not int or x not in READ_OPTIONS for x in options)
    ):
        raise ValueError("Only identity and USB/wakeup option reads are allowed")
    if type(message_id) is not int or not 1 <= message_id < 0x40000000:
        raise ValueError("Invalid message ID")
    payload = b"".join(b"\x08" + varint(option) for option in options)
    # CameraPacket header7 + CameraMessage header9. Command8=GET_OPTIONS.
    return (
        struct.pack("<IBHHBIH", 16 + len(payload), 4, 0, 8, 2, message_id | 0x80000000, 0) + payload
    )


class PacketReader:
    """Accept split/combined notifications; no unbounded resync or CRC guessing."""

    def __init__(self):
        self.buffer = bytearray()

    def feed(self, chunk):
        if not chunk or len(self.buffer) + len(chunk) > MAX_PACKET * 2:
            raise ValueError("Invalid notification size")
        self.buffer.extend(chunk)
        result = []
        while len(self.buffer) >= 4:
            size = struct.unpack_from("<I", self.buffer)[0]
            if not 16 <= size <= MAX_PACKET:
                raise ValueError("Unknown packet framing")
            if len(self.buffer) < size:
                break
            raw = bytes(self.buffer[:size])
            del self.buffer[:size]
            _, kind, reserved, code, content, ident, tail = struct.unpack("<IBHHBIH", raw[:16])
            if kind != 4 or reserved or tail or content != 2 or ident & 0xC0000000 != 0xC0000000:
                raise ValueError("Unsupported message header/direction/fragmentation")
            result.append(
                {"code": code, "message_id": ident & 0x3FFFFFFF, "payload": raw[16:], "raw": raw}
            )
        return result


def decode_options_response(packet, message_id, requested):
    if packet["message_id"] != message_id:
        raise ValueError("Response belongs to another request")
    # HTTP-style response codes in the common protocol. Hardware must confirm.
    if packet["code"] != 200:
        raise ValueError("Camera did not return a successful response")
    listed, values = [], None
    for number, wire, value in protobuf_fields(packet["payload"]):
        if number == 1:
            if wire == 0:
                listed.append(value)
            elif wire == 2:
                offset = 0
                while offset < len(value):
                    item, offset = read_varint(value, offset)
                    listed.append(item)
            else:
                raise ValueError("Invalid option list")
        elif number == 2:
            if wire != 2 or values is not None:
                raise ValueError("Invalid or duplicate Options value")
            values = value
    if values is None or len(listed) != len(set(listed)) or not set(listed) <= set(requested):
        raise ValueError("Missing/ambiguous Options response")
    decoded = {}
    for number, wire, value in protobuf_fields(values):
        if number not in requested:
            continue
        if number in decoded or number not in listed:
            raise ValueError("Duplicate or unacknowledged option")
        if number in {15, 48}:
            if wire != 2 or not 1 <= len(value) <= 256:
                raise ValueError("Invalid camera identity")
            decoded[number] = value.decode("utf-8", errors="strict")
        else:
            if wire != 0 or value not in {0, 1}:
                raise ValueError("Unsupported option enum")
            decoded[number] = value
    # Missing scalar is unknown, NEVER default PC=0 or wakeup=0.
    return {
        number: {
            "acknowledged": number in listed,
            "present": number in decoded,
            "value": decoded.get(number),
        }
        for number in requested
    }