Python client for GEX
No puede seleccionar más de 25 temas Los temas deben comenzar con una letra o número, pueden incluir guiones ('-') y pueden tener hasta 35 caracteres de largo.
gex-client-py/PayloadParser.py

123 líneas
3.5 KiB

import struct
from gex.TinyFrame import TF_Msg
class PayloadParser:
"""
Utility for parsing a binary payload
"""
def __init__(self, buf, endian:str='little'):
"""
buf - buffer to parse (bytearray or binary string)
"""
if type(buf) == TF_Msg:
buf = buf.data
self.buf = buf
self.ptr = 0
self.endian = endian
def _slice(self, n:int) -> bytearray:
""" Extract a slice and advance the read pointer for the next slice """
if self.ptr + n > len(self.buf):
raise Exception("Payload parser underrun - frame: %s" % str(self.buf))
slice = self.buf[self.ptr:self.ptr + n]
self.ptr += n
return slice
def length(self) -> int:
""" Measure the tail """
return len(self.buf) - self.ptr
def rewind(self):
""" Reset the slice pointer to the beginning """
self.ptr = 0
def tail(self) -> bytearray:
""" Get all remaining bytes """
return self._slice(len(self.buf) - self.ptr)
def u8(self) -> int:
""" Read a uint8_t """
slice = self._slice(1)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u16(self) -> int:
""" Read a uint16_t """
slice = self._slice(2)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u24(self) -> int:
""" Read a uint24_t """
slice = self._slice(3)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u32(self) -> int:
""" Read a uint32_t """
slice = self._slice(4)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u64(self) -> int:
""" Read a uint64_t """
slice = self._slice(8)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def i8(self) -> int:
""" Read a int8_t """
slice = self._slice(1)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def i16(self) -> int:
""" Read a int16_t """
slice = self._slice(2)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def i32(self) -> int:
""" Read a int32_t """
slice = self._slice(4)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def i64(self) -> int:
""" Read a int64_t """
slice = self._slice(8)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def float(self) -> float:
""" Read a float (4 bytes) """
slice = self._slice(4)
fmt = '<f' if self.endian == 'little' else '>f'
return struct.unpack(fmt, slice)[0]
def double(self) -> float:
""" Read a double (8 bytes) """
slice = self._slice(8)
fmt = '<d' if self.endian == 'little' else '>d'
return struct.unpack(fmt, slice)[0]
def bool(self) -> bool:
""" Read a bool (1 byte, True if != 0) """
return 0 != self._slice(1)[0]
def str(self) -> str:
""" Read a zero-terminated string """
p = self.ptr
while p < len(self.buf) and self.buf[p] != 0:
p += 1
bs = self._slice(p - self.ptr)
self.ptr += 1
return bs.decode('utf-8')
def blob(self, length) -> bytearray:
""" Read a blob of given length """
return self._slice(length)
def skip(self, nbytes:int):
""" Skip some bytes. returns self for chaining. """
self.blob(nbytes)
return self