Decoding a LIFX packet

import binascii
import struct


class IncompleteHeader(Exception):
    def __init__(self):
        super().__init__("Insufficient data to unpack a Header")


class Header:
    @classmethod
    def read(kls, data):
        if len(data) < 36:
            raise IncompleteHeader()
        return Header(data[:36])

    def __init__(self, bts):
        self._bts = bts

    def __getitem__(self, rng):
        return self._bts[rng]

    @property
    def size(self):
        """returns the size of the total message."""
        return struct.unpack("<H", self[0:2])[0]

    @property
    def protocol(self):
        """returns the protocol version of the header."""
        v = struct.unpack("<H", self[2:4])[0]
        return v & 0b111111111111

    @property
    def addressable(self):
        """returns whether the addressable bit is set."""
        v = self[3]
        v = v >> 4
        return (v & 0b1) != 0

    @property
    def tagged(self):
        """returns whether the tagged bit is set."""
        v = self[3]
        v = v >> 5
        return (v & 0b1) != 0

    @property
    def source(self):
        """returns then number used by clients to differentiate themselves from other clients"""
        return struct.unpack("<I", self[4:8])[0]

    @property
    def target(self):
        """returns the target Serial from the header."""
        return binascii.hexlify(self[8:16][:6]).decode()

    @property
    def response_required(self):
        """returns whether the response required bit is set in the header."""
        v = self[22]
        return (v & 0b1) != 0

    @property
    def ack_required(self):
        """returns whether the ack required bit is set in the header."""
        v = self[22]
        v = v >> 1
        return (v & 0b1) != 0

    @property
    def sequence(self):
        """returns the sequence ID from the header."""
        return self[23]

    @property
    def pkt_type(self):
        """returns the Payload ID for the accompanying payload in the message."""
        return struct.unpack("<H", self[32:34])[0]


if __name__ == "__main__":
    bts = binascii.unhexlify(
        "2400001400034746d073d500133700000000000000000701000000000000000014000000"
    )
    header = Header.read(bts)

    print("size:", header.size)
    print("protocol:", header.protocol)
    print("addressable:", header.addressable)
    print("tagged:", header.tagged)
    print("source:", header.source)
    print("target:", header.target)
    print("res_required:", header.response_required)
    print("ack_required:", header.ack_required)
    print("sequence:", header.sequence)
    print("pkt_type:", header.pkt_type)

    # size: 36
    # protocol: 1024
    # addressable: True
    # tagged: False
    # source: 1179058944
    # target: d073d5001337
    # res_required: True
    # ack_required: True
    # sequence: 1
    # pkt_type: 20

package main

import (
	"encoding/binary"
	"encoding/hex"
	"fmt"
	"log"
)

const headerSize = 36

type headerError string

func (e headerError) Error() string {
	return string(e)
}

// Errors associated with unpacking headers.
const (
	ErrHeaderIncomplete headerError = "Insufficient data to unpack a Header"
)

// Header is a LIFX protocol message header containing
// the bytes in the serialized message header, and methods to
// extract meaningful data from it.
type Header [36]byte

// ReadHeader reads just the header from the data.
func ReadHeader(data []byte) (*Header, error) {
	if len(data) < 36 {
		return nil, ErrHeaderIncomplete
	}

	var h Header
	copy(h[:], data[:36])

	return &h, nil
}

// Size returns the size of the total message.
func (h *Header) Size() uint16 {
	return binary.LittleEndian.Uint16(h[0:2])
}

// Protocol returns the protocol version of the header.
func (h *Header) Protocol() uint16 {
	v := binary.LittleEndian.Uint16(h[2:4])
	v &= 0b111111111111
	return v
}

// Addressable returns whether the addressable bit is set.
func (h *Header) Addressable() bool {
	v := h[3]
	v >>= 4
	v &= 0b1
	return v != 0
}

// Tagged returns whether the tagged bit is set.
func (h *Header) Tagged() bool {
	v := h[3]
	v >>= 5
	v &= 0b1
	return v != 0
}

// Source returns then number used by clients to differentiate themselves from other clients
func (h *Header) Source() uint32 {
	return binary.LittleEndian.Uint32(h[4:8])
}

// Target returns the target Serial from the header.
func (h *Header) Target() string {
	t := make([]byte, 8)
	copy(t, h[8:16])
	return hex.EncodeToString(t[:6])
}

// ResponseRequired returns whether the response required bit is set in the header.
func (h *Header) ResponseRequired() bool {
	v := h[22]
	v &= 0b1
	return v != 0
}

// AckRequired returns whether the ack required bit is set in the header.
func (h *Header) AckRequired() bool {
	v := h[22]
	v >>= 1
	v &= 0b1
	return v != 0
}

// Sequence returns the sequence ID from the header.
func (h *Header) Sequence() uint8 {
	return h[23]
}

// Type returns the Payload ID for the accompanying payload in the message.
func (h *Header) Type() uint16 {
	return binary.LittleEndian.Uint16(h[32:34])
}

// Example of using the Header class
func main() {
	bts, err := hex.DecodeString("2400001400034746d073d500133700000000000000000701000000000000000014000000")
	if err != nil {
		log.Fatal(fmt.Errorf("Failed to decode bytes, %w", err))
	}
	header, err := ReadHeader(bts)
	if err != nil {
		log.Fatal(fmt.Errorf("Failed to read header, %w", err))
	}

	fmt.Printf("size: %d\n", header.Size())
	fmt.Printf("protocol: %d\n", header.Protocol())
	fmt.Printf("addressable: %t\n", header.Addressable())
	fmt.Printf("tagged: %t\n", header.Tagged())
	fmt.Printf("source: %d\n", header.Source())
	fmt.Printf("target: %s\n", header.Target())
	fmt.Printf("res_required: %t\n", header.ResponseRequired())
	fmt.Printf("ack_required: %t\n", header.AckRequired())
	fmt.Printf("sequence: %d\n", header.Sequence())
	fmt.Printf("pkt_type: %d\n", header.Type())

	// size: 36
	// protocol: 1024
	// addressable: true
	// tagged: false
	// source: 1179058944
	// target: d073d5001337
	// res_required: true
	// ack_required: true
	// sequence: 1
	// pkt_type: 20
}

#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#pragma pack(push, 1)
typedef struct lx_frame_t {
  uint16_t size;
  uint16_t protocol : 12;
  uint8_t addressable : 1;
  uint8_t tagged : 1;
  uint8_t reserved_1 : 2;
  uint32_t source;

  uint64_t target;
  uint8_t reserved_2[6];
  uint8_t res_required : 1;
  uint8_t ack_required : 1;
  uint8_t reserved_3 : 6;
  uint8_t sequence;

  uint64_t reserved_4;
  uint16_t pkt_type;
  uint16_t reserved_5;
} lx_frame_t;
#pragma pack(pop)

void extract_frame_serial(lx_frame_t *frame, char *serial) {
  char ss[16];
  sprintf(ss, "%#06llx", frame->target);
  for (int i = 12; i > 0; i = i - 2) {
    serial[12 - i] = ss[i];
    serial[13 - i] = ss[i + 1];
  }
}

lx_frame_t *read_frame(char *hex) {
  int str_len = strlen(hex);

  if (str_len < 72) {
    return NULL;
  }

  uint8_t buffer[36] = {};

  for (int i = 0; i < (str_len / 2) && i < 72; i++) {
    unsigned int nxt;
    sscanf(hex + 2 * i, "%02x", &nxt);
    buffer[i] = (uint8_t)nxt;
  }

  struct lx_frame_t *f;
  f = (lx_frame_t *)malloc(sizeof(struct lx_frame_t));
  memcpy(f, buffer, 36);

  return f;
}

int main() {
  char *hex = "2400001400034746d073d5001337000000000000000007010000000"
              "00000000014000000";

  lx_frame_t *frame = read_frame(hex);

  if (frame == NULL) {
    fprintf(stdout, "Failed to convert hex into a header");
    return 1;
  }

  char serial[12] = {};
  extract_frame_serial(frame, serial);

  char *t = "true";
  char *f = "false";
  char *bools[2] = {f, t};

  fprintf(stdout, "size: %d\n", frame->size);
  fprintf(stdout, "protocol: %hu\n", frame->protocol);
  fprintf(stdout, "addressable: %s\n", bools[frame->addressable]);
  fprintf(stdout, "lagged: %s\n", bools[frame->tagged]);
  fprintf(stdout, "source: %d\n", frame->source);
  fprintf(stdout, "target: %s\n", serial);
  fprintf(stdout, "res_required: %s\n", bools[frame->res_required]);
  fprintf(stdout, "ack_required: %s\n", bools[frame->ack_required]);
  fprintf(stdout, "sequence: %d\n", frame->sequence);
  fprintf(stdout, "pkt_type: %d\n", frame->pkt_type);

  /*
  # size: 36
  # protocol: 1024
  # addressable: True
  # tagged: False
  # source: 1179058944
  # target: d073d5001337
  # res_required: True
  # ack_required: True
  # sequence: 1
  # pkt_type: 20
  */

  free(frame);
}

function read_header(s) {
  if (s.length < 72) {
    throw new Error("Need atleast 72 characters in hex for frame header");
  }

  if (!/^[a-fA-F0-9]*$/.test(s)) {
    throw new Error(`Invalid HEX input: "${s}"`);
  }

  const buffer = new Uint8Array(36).buffer;
  const data = new DataView(buffer, 0, 36);
  const bytes = new Uint8Array(data.buffer);

  bytes.forEach((_, i, b) => (b[i] = parseInt(s.substr(2 * i, 2), 16)));

  return data;
}

class Header {
  constructor(s) {
    this.data = read_header(s);
  }

  get size() {
    let v = this.data.getUint16(0, true);

    return v;
  }

  get protocol() {
    let v = this.data.getUint16(2, true);
    v &= 0b111111111111;
    return v;
  }

  get addressable() {
    let v = this.data.getUint8(3);
    v >>= 4;
    v &= 0b1;
    return !!v;
  }

  get tagged() {
    let v = this.data.getUint8(3);
    v >>= 5;
    v &= 0b1;
    return !!v;
  }

  get source() {
    return this.data.getUint32(4, true);
  }

  get target() {
    const bytes = new Uint8Array(this.data.buffer, 8, 6);
    const serial = [];
    bytes.map(b => serial.push(b.toString(16).padStart(2, "0")));
    return serial.join("");
  }

  get resRequired() {
    let v = this.data.getUint8(22);
    v &= 0b1;
    return !!v;
  }

  get ackRequired() {
    let v = this.data.getUint8(22);
    v >>= 1;
    v &= 0b1;
    return !!v;
  }

  get sequence() {
    let v = this.data.getUint8(23);

    return v;
  }

  get type() {
    let v = this.data.getUint16(32, true);

    return v;
  }
}

header = new Header(
  "2400001400034746d073d500133700000000000000000701000000000000000014000000"
);

console.log("size:", header.size);
console.log("protocol:", header.protocol);
console.log("addressable:", header.addressable);
console.log("tagged:", header.tagged);
console.log("source:", header.source);
console.log("target:", header.target);
console.log("res_required:", header.resRequired);
console.log("ack_required:", header.ackRequired);
console.log("sequence:", header.sequence);
console.log("pkt_type:", header.type);

// size: 36
// protocol: 1024
// addressable: true
// tagged: false
// source: 1179058944
// target: d073d5001337
// res_required: true
// ack_required: true
// sequence: 1
// pkt_type: 20