gex/gex-client-py
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

implemented payload parser and builder

Browse Source
This commit is contained in:
MightyPork
2017-12-27 11:37:55 +01:00
parent 105d1c1a3c
commit 2642e9184f
5 changed files with 181 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gex/PayloadBuilder.py
+38
View File
@@ -0,0 +1,38 @@
import struct
class PayloadBuilder:
def __init__(self, endian='little'):
self.buf = bytearray()
self.endian = endian
def close(self):
return self.buf
def u8(self, num):
self.buf.extend(num.to_bytes(length=1, byteorder=self.endian, signed=False))
def u16(self, num):
self.buf.extend(num.to_bytes(length=2, byteorder=self.endian, signed=False))
def u32(self, num):
self.buf.extend(num.to_bytes(length=4, byteorder=self.endian, signed=False))
def i8(self, num):
self.buf.extend(num.to_bytes(length=1, byteorder=self.endian, signed=True))
def i16(self, num):
self.buf.extend(num.to_bytes(length=2, byteorder=self.endian, signed=True))
def i32(self, num):
self.buf.extend(num.to_bytes(length=4, byteorder=self.endian, signed=True))
def float(self, num):
fmt = '<f' if self.endian == 'little' else '>f'
self.buf.extend(struct.pack(fmt, num))
def bool(self, num):
self.buf.append(1 if num != False else 0)
def str(self, string):
self.buf.extend(string.encode('utf-8'))
self.buf.append(0)
gex/PayloadParser.py
+62
View File
@@ -0,0 +1,62 @@
import struct
class PayloadParser:
def __init__(self, buf, endian='little'):
self.buf = buf
self.ptr = 0
self.endian = endian
def slice(self, n):
if self.ptr + n > len(self.buf):
raise Exception("Out of bounds")
slice = self.buf[self.ptr:self.ptr + n]
self.ptr += n
return slice
def u8(self):
slice = self.slice(1)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u16(self):
slice = self.slice(2)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def u32(self):
slice = self.slice(4)
return int.from_bytes(slice, byteorder=self.endian, signed=False)
def i8(self):
slice = self.slice(1)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def i16(self):
slice = self.slice(2)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def i32(self):
slice = self.slice(4)
return int.from_bytes(slice, byteorder=self.endian, signed=True)
def float(self):
slice = self.slice(4)
fmt = '<f' if self.endian == 'little' else '>f'
return struct.unpack(fmt, slice)[0]
def bool(self):
return 0 != self.slice(1)[0]
def str(self):
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 rewind(self):
self.ptr = 0
def tail(self):
return self.slice(len(self.buf) - self.ptr)
gex/TinyFrame.py
+5 -6
View File
@@ -1,12 +1,11 @@
class TF_Msg:
def __init__(self):
self.data = bytearray()
self.len = 0
self.type = 0
self.id = 0
def __init__(self, id=0, type=0, data=None):
self.data = data if data is not None else bytearray()
self.type = type
self.id = id
def __str__(self):
return f"ID {self.id:X}h, type {self.type:X}h, len {self.len:d}, body: {self.data}"
return f"ID {self.id:X}h, type {self.type:X}h, body: {self.data}"
class TF:
STAY = 'STAY'
gex/__init__.py
+39 -26
View File
@@ -2,38 +2,45 @@
import serial
from gex.TinyFrame import TinyFrame
from gex.TinyFrame import TinyFrame, TF_Msg
# General, low level
MSG_SUCCESS = 0x00 # Generic success response; used by default in all responses; payload is transaction-specific
MSG_PING = 0x01 # Ping request (or response), used to test connection
MSG_ERROR = 0x02 # Generic failure response (when a request fails to execute)
MSG_BULK_READ_OFFER = 0x03 # Offer of data to read. Payload: u32 total len
MSG_BULK_READ_POLL = 0x04 # Request to read a previously announced chunk. Payload: u32 max chunk
MSG_BULK_WRITE_OFFER = 0x05 # Offer to receive data in a write transaction. Payload: u32 max size, u32 max chunk
MSG_BULK_DATA = 0x06 # Writing a chunk, or sending a chunk to master.
MSG_BULK_END = 0x07 # Bulk transfer is done, no more data to read or write.
# Recipient shall check total len and discard it on mismatch. There could be a checksum ...
MSG_BULK_ABORT = 0x08 # Discard the ongoing transfer
# Unit messages
MSG_UNIT_REQUEST = 0x10 # Command addressed to a particular unit
MSG_UNIT_REPORT = 0x11 # Spontaneous report from a unit
# System messages
MSG_LIST_UNITS = 0x20 # Get all unit call-signs and names
MSG_INI_READ = 0x21 # Read the ini file via bulk
MSG_INI_WRITE = 0x22 # Write the ini file via bulk
MSG_PERSIST_SETTINGS = 0x23 # Write current settings to Flash
class Gex:
# General, low level
MSG_SUCCESS = 0x00 # Generic success response; used by default in all responses; payload is transaction-specific
MSG_PING = 0x01 # Ping request (or response), used to test connection
MSG_ERROR = 0x02 # Generic failure response (when a request fails to execute)
MSG_BULK_READ_OFFER = 0x03 # Offer of data to read. Payload: u32 total len
MSG_BULK_READ_POLL = 0x04 # Request to read a previously announced chunk. Payload: u32 max chunk
MSG_BULK_WRITE_OFFER = 0x05 # Offer to receive data in a write transaction. Payload: u32 max size, u32 max chunk
MSG_BULK_DATA = 0x06 # Writing a chunk, or sending a chunk to master.
MSG_BULK_END = 0x07 # Bulk transfer is done, no more data to read or write.
# Recipient shall check total len and discard it on mismatch. There could be a checksum ...
MSG_BULK_ABORT = 0x08 # Discard the ongoing transfer
# Unit messages
MSG_UNIT_REQUEST = 0x10 # Command addressed to a particular unit
MSG_UNIT_REPORT = 0x11 # Spontaneous report from a unit
# System messages
MSG_LIST_UNITS = 0x20 # Get all unit call-signs and names
MSG_INI_READ = 0x21 # Read the ini file via bulk
MSG_INI_WRITE = 0x22 # Write the ini file via bulk
MSG_PERSIST_SETTINGS = 0x23 # Write current settings to Flash
def __init__(self, port='/dev/ttyACM0', timeout=0.2):
self.port = port
self.serial = serial.Serial(port=port, timeout=timeout)
self.tf = TinyFrame()
self.tf.write = self._write
# test connection
self.query_raw(type=MSG_PING)
self.load_units()
def _write(self, data):
self.serial.write(data)
pass
@@ -77,11 +84,13 @@ class Gex:
buf = bytearray([cs, cmd])
buf.extend(pld)
self._send(type=self.MSG_UNIT_REQUEST, id=id, pld=buf, listener=listener)
self._send(type=MSG_UNIT_REQUEST, id=id, pld=buf, listener=listener)
def query(self, cs, cmd, id=None, pld=None):
""" Query a unit """
self._theframe = None
def lst(tf, frame):
self._theframe = frame
@@ -99,4 +108,8 @@ class Gex:
def send_raw(self, type, id=None, pld=None):
""" Send without addressing a unit """
return self.send(cs=None, cmd=type, id=id, pld=pld)
return self.send(cs=None, cmd=type, id=id, pld=pld)
def load_units(self):
resp = self.query_raw(type=MSG_LIST_UNITS)
main.py
+37 -9
View File
@@ -1,18 +1,46 @@
#!/bin/env python3
import time
from gex import Gex
import gex
from gex.PayloadParser import PayloadParser
from gex.PayloadBuilder import PayloadBuilder
gex = Gex()
if False:
pb = PayloadBuilder()
pb.u8(128)
pb.i8(-1)
pb.u16(1)
pb.u32(123456789)
pb.float(3.1415)
pb.bool(True)
pb.bool(False)
pb.str("FUCKLE")
# Check connection
resp = gex.query_raw(type=Gex.MSG_PING)
print("Ping resp = ", resp.data.decode("ascii"))
buf = pb.close()
print(buf)
# Blink a LED at call-sign 1, command 0x02 = toggle
for i in range(0,10):
gex.send(cs=1, cmd=0x02)
time.sleep(.2)
pp = PayloadParser(buf)
print('>',pp.u8())
print('>',pp.i8())
print('>',pp.u16())
print('>',pp.u32())
print('>',pp.float())
print('>',pp.bool())
print('>',pp.bool())
print('>',pp.str())
if False:
client = gex.Gex()
# Check connection
resp = client.query_raw(type=gex.MSG_PING)
print("Ping resp = ", resp.data.decode("ascii"))
# Blink a LED at call-sign 1, command 0x02 = toggle
for i in range(0,10):
client.send(cs=1, cmd=0x02)
time.sleep(.1)
#