Mock: update client

BLE_GattServer_FOTAService/scripts/packets.txt

@@ -0,0 +1,8 @@
+58cd7164304c4aa8b32b51856d369f
+796ab013bf500bc3a6ba3a050ba73d
+f097d22619c8c41602cf24d9e031a4
+9debe25f7a865ace0d56aeea8316de
+e3df3ada187a389178ecdaf96caa09
+035044aa213836e20acea27ee47f8f
+584b4ac00b995c819fb3ccbe1f25fe
+ec273816af362eab7ae4a145e052b9

BLE_GattServer_FOTAService/scripts/test_fota.py

@@ -16,8 +16,8 @@
 
 import logging
 import asyncio
+from logging import Logger
 
-from os             import urandom
 from bleak          import BleakScanner, BleakClient, BleakError
 from bleak.uuids    import uuid16_dict
 from typing         import Optional
@@ -27,7 +27,9 @@
 
 fileConfig('logging.conf')
 
-logger = logging.getLogger('FOTA')
+logger: Logger = logging.getLogger('FOTA')
+
+packets = open('packets.txt', 'r')
 
 # FOTA UUIDS
 UUID_FOTA_SERVICE                  = "53880000-65fd-4651-ba8e-91527f06c887"
@@ -62,30 +64,20 @@
 FOTA_OP_CODE_SET_FRAGMENT_ID       = bytearray(b'\x43')
 
 # Number of bytes in a single BSC packet, including fragment ID
-FRAGMENT_SIZE   = 8
+FRAGMENT_SIZE   = 16
 
-MAXIMUM_RETRIES = 100
+MAXIMUM_RETRIES = 8
 
+PACKET_COUNT    = 8
 
-class DataGenerator:
 
-    def __init__(self, chunk_size):
-        self.chunk_size = chunk_size
-        self.chunks = []
+class DataReader:
 
-    def get_chunk_size(self):
-        return self.chunk_size
+    def __init__(self):
+        self.chunks = packets.readlines()
 
     def get_chunk_n(self, n: int) -> bytearray:
-        # See if we've already generated the requested chunk number
-        if n < len(self.chunks):
-            return self.chunks[n]
-        else:
-            # Generate up to the requested chunk number
-            while len(self.chunks) < n+1:
-                self.chunks.append([])
-                self.chunks[len(self.chunks)-1] = bytearray(urandom(self.chunk_size))
-            return self.chunks[n]
+        return bytearray(self.chunks[n].rstrip().encode())
 
 
 class StatusNotificationHandler:
@@ -108,10 +100,9 @@ class FOTASession:
 
     def __init__(self, client: BleakClient):
         self.client = client
-        self.data_generator = DataGenerator(FRAGMENT_SIZE)
+        self.data_reader = DataReader()
         self.handler = StatusNotificationHandler()
         self.fragment_id = 0
-        self.rollover_counter = 0
 
     async def start(self):
         # Subscribe to notifications from the status characteristic
@@ -134,40 +125,33 @@ async def start(self):
                 else:
                     logger.warning(f'Received unknown type of status notification ({status_code[0]})')
             except asyncio.TimeoutError:
-                logger.info(f'Waiting for status notification: ({retries }/{MAXIMUM_RETRIES})')
+                logger.info(f'Waiting for status notification: ({retries}/{MAXIMUM_RETRIES})')
                 retries += 1
                 if retries >= MAXIMUM_RETRIES:
-                    # We timed out, raise an exception
                     raise asyncio.TimeoutError
 
     async def write_n_packets(self, n: int) -> Optional[bytearray]:
-        i = (self.rollover_counter * 0xFF) + self.fragment_id
+        count = self.fragment_id
         limit = self.fragment_id + n
-        while i < limit:
+        while count < limit:
             # Check for status notification
             if self.handler.status_event.is_set():
-                self.handler.status_event.clear()
                 if self.handler.status_value[0] == FOTA_STATUS_SYNC_LOST[0]:
-                    logger.info(f'Retransmission requested from fragment ID: {self.handler.status_value[1]}')
+                    logger.info(f'Retransmission requested from fragment ID {self.handler.status_value[1]}')
+                self.handler.status_event.clear()
                 return self.handler.status_value
 
             # Send the next packet
             packet = bytearray([self.fragment_id])
-            data_chunk = self.data_generator.get_chunk_n(i)
+            data_chunk = self.data_reader.get_chunk_n(count)
             packet += data_chunk
-            logger.info(f'Writing packet {i}: data={bytes(packet)}')
+            logger.info(f'Writing packet {count}: data={bytes(packet)}')
             await asyncio.wait_for(self.client.write_gatt_char(UUID_BINARY_STREAM_CHAR, packet), timeout=0.2)
 
-            # Increment fragment ID
+            # Increment loop counter and fragment ID
+            count += 1
             self.fragment_id += 1
 
-            # Roll over fragment ID
-            if self.fragment_id >= 256:
-                self.fragment_id = 0
-                self.rollover_counter += 1
-
-            i = (self.rollover_counter * 0xFF) + self.fragment_id
-
         return None
 
 
@@ -196,9 +180,9 @@ async def find(name: str) -> Optional[BleakClient]:
 async def main():
     client = await find('FOTA')
 
-    if not client:
-        logger.error('Could not connect to FOTA target')
-        quit()
+    if client is None:
+        return
+
     session = FOTASession(client)
 
     try:
@@ -208,9 +192,9 @@ async def main():
         await client.disconnect()
         return
 
-    logger.info("FOTA session started")
+    logger.info('FOTA session started.')
 
-    await session.write_n_packets(10)
+    await session.write_n_packets(PACKET_COUNT)
 
     await client.disconnect()