oliver/cls_master
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

using proto enum for types

Browse Source
This commit is contained in:
Oliver Walter
2024-02-03 03:51:57 +01:00
parent ceba7e5a35
commit 5cc20a3cb4
5 changed files with 35 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -2,4 +2,4 @@ build/
venv/
tools/*_pb2.py
tools/__pycache__
tools/*.bin
tools/**/*.bin

8
Application/Tasks/UsbDataHandler.c
View File

@@ -94,10 +94,10 @@ typedef struct {
message_handler_t message_handlers[] = {
MESSAGE_HANDLER(0xF01, FirmwareStart),
MESSAGE_HANDLER(0xF02, FirmwarePackage),
MESSAGE_HANDLER(0xF03, FirmwareDone),
MESSAGE_HANDLER(0xF04, FirmwarePackageAck),
MESSAGE_HANDLER(UsbPackageType_FIRMWARESTART, FirmwareStart),
MESSAGE_HANDLER(UsbPackageType_FIRMWAREPACKAGE, FirmwarePackage),
MESSAGE_HANDLER(UsbPackageType_FIRMWAREDONE, FirmwareDone),
MESSAGE_HANDLER(UsbPackageType_FIRMWAREPACKAGEACK, FirmwarePackageAck),
};

8
proto/firmware.proto
View File

@@ -2,6 +2,14 @@ syntax = "proto2";
import "nanopb.proto";
enum UsbPackageType {
FIRMWARESTART = 3841; // 0xF01
FIRMWAREPACKAGE = 3842; // 0xF02
FIRMWAREDONE = 3843; // 0xF03
FIRMWAREPACKAGEACK = 3844; // 0xF04
}
message FirmwareStart {
required string name = 1 [(nanopb).max_size = 32];
required uint32 size = 2;

12
tests/native/tests.c
View File

@@ -61,7 +61,7 @@ void test_UsbDataHandler(void) {
.crc_fw=START_CRC,
.device_id=START_DEVICE,
};
transmit_package(0xf01, FirmwareStart_fields, &start);
transmit_package(UsbPackageType_FIRMWARESTART, FirmwareStart_fields, &start);
// Call the runner with start message in the queue
UsbDataHandler_Runner();
@@ -69,8 +69,8 @@ void test_UsbDataHandler(void) {
//Put 2 packages in the buffer an parse them
transmit_package(0xf01, FirmwareStart_fields, &start);
transmit_package(0xf01, FirmwareStart_fields, &start);
transmit_package(UsbPackageType_FIRMWARESTART, FirmwareStart_fields, &start);
transmit_package(UsbPackageType_FIRMWARESTART, FirmwareStart_fields, &start);
UsbDataHandler_Runner();
UsbDataHandler_Runner();
TEST_ASSERT_EQUAL(StartCounter,3);
@@ -85,7 +85,7 @@ void test_UsbDataHandler(void) {
.bytes={0}
},
};
transmit_package(0xf02, FirmwarePackage_fields, &pack);
transmit_package(UsbPackageType_FIRMWAREPACKAGE, FirmwarePackage_fields, &pack);
UsbDataHandler_Runner();
TEST_ASSERT_EQUAL(PackageCounter,1);
@@ -95,7 +95,7 @@ void test_UsbDataHandler(void) {
.crc_pac = ACK_CRC,
.device_id = ACK_DEVICE,
};
transmit_package(0xf04, FirmwarePackageAck_fields, &ack);
transmit_package(UsbPackageType_FIRMWAREPACKAGEACK, FirmwarePackageAck_fields, &ack);
UsbDataHandler_Runner();
TEST_ASSERT_EQUAL(PackageCounter,1);
@@ -106,7 +106,7 @@ void test_UsbDataHandler(void) {
.size = DONE_SIZE,
};
transmit_package(0xf03, FirmwareDone_fields, &done);
transmit_package(UsbPackageType_FIRMWAREDONE, FirmwareDone_fields, &done);
UsbDataHandler_Runner();
TEST_ASSERT_EQUAL(DoneCounter,1);

30
tools/send_fw.py
View File

@@ -1,15 +1,15 @@
import serial
import struct
import time
from google.protobuf.internal import encoder
from google.protobuf.internal import decoder
import sys
from google.protobuf.message import DecodeError
from serial.tools import list_ports
from crccheck.crc import Crc32
from crccheck.crc import Crc
from firmware_pb2 import FirmwareStart, FirmwarePackage, FirmwarePackageAck, FirmwareDone
from firmware_pb2 import FirmwareStart, FirmwarePackage, FirmwarePackageAck, FirmwareDone, UsbPackageType
Crc32 = Crc(32,0x04C11DB7,0xffffffff)
def load_firmware(filename):
# Open the file in binary mode
@@ -24,15 +24,16 @@ def load_firmware(filename):
return crc, size, packages
def make_header(typeid: int, length: int) -> bytearray:
def make_header(typeid: UsbPackageType, length: int) -> bytearray:
struct_format = '<HHB' # '<' for little-endian, 'H' for uint16_t, 'B' for uint8_t
# Calculate the check byte as the sum of length and type
check = (length & 0xFF) + ((length >> 8) & 0xFF) + (typeid & 0xFF) + ((typeid >> 8) & 0xFF)
typeidint = int(typeid)
check = (length & 0xFF) + ((length >> 8) & 0xFF) + (typeidint & 0xFF) + ((typeidint >> 8) & 0xFF)
packed_data = struct.pack(struct_format, length, typeid, check)
return packed_data
def send_package(typeid :int, data: bytearray, serial: serial.Serial):
def send_package(typeid : UsbPackageType, data: bytearray, serial: serial.Serial):
head = make_header(typeid, len(data))
package = head + data
serial.write(package)
@@ -55,7 +56,7 @@ def receive_ack(serial):
print('Incomplete message')
return None
if typeid != 0xf04:
if typeid != UsbPackageType.FIRMWAREPACKAGEACK:
return None
# Parse the message
@@ -69,6 +70,8 @@ def receive_ack(serial):
return ack
FILENAME = 'firmware.bin'
if len(sys.argv) > 1:
FILENAME = sys.argv[1]
if __name__ == "__main__":
@@ -99,8 +102,8 @@ if __name__ == "__main__":
# Send the FirmwareStart message
print(start)
#send_package(0xf01, start.SerializeToString(), ser)
time.sleep(1) # wait for the device to process the message
send_package(UsbPackageType.FIRMWARESTART, start.SerializeToString(), ser)
#time.sleep(1) # wait for the device to process the message
# Send the firmware packages
for (i,pack_data) in enumerate(packages):
@@ -113,8 +116,7 @@ if __name__ == "__main__":
# Send the FirmwarePackage message
print(package)
print(hex(package.crc_pac))
exit(0)
#send_package(0xf02, package.SerializeToString(), ser)
send_package(UsbPackageType.FIRMWAREPACKAGE, package.SerializeToString(), ser)
# Wait for the FirmwarePackageAck message
@@ -122,7 +124,7 @@ if __name__ == "__main__":
print(ack)
if not ack.ack:
print(f'Package {i} not acknowledged')
break
exit(-1)
# Send the FirmwareDone message
@@ -131,4 +133,4 @@ if __name__ == "__main__":
done.crc_fw = start.crc_fw
done.device_id = start.device_id
print(done)
send_package(0xf03, done.SerializeToString(), ser)
send_package(UsbPackageType.FIRMWAREDONE, done.SerializeToString(), ser)