Removed some unneeded type hints to lower Python requirements

Tools/Bluetooth/BLE_hci.py

@@ -295,17 +295,24 @@ def wait_events(self, seconds = 2, print_evt = True):
      # Send a HCI command to the serial port. Will add a small delay and wait for
      # and print an HCI event by default.
     ################################################################################
-    def send_command(self, packet, resp = True, delay = 0.01, print_cmd = True, timeout=3):
+    def send_command(self, packet, resp = True, delay = 0.01, print_cmd = True, timeout=16):
         # Send the command and data
         if(print_cmd):
             if self.id == "-":
                 print(str(datetime.datetime.now()) + "  >", packet)
             else:
                 print(str(datetime.datetime.now()) + f" {self.id}>", packet)
 
-        self.port.write(bytearray.fromhex(packet))
+        print('Packet', packet)
+
+        arr = bytearray.fromhex(packet)
+
+        print(arr)
+
+        self.port.write(arr)
+
         sleep(delay)
-
+            
         if(resp):
             return self.wait_event(timeout=timeout)
 
@@ -646,7 +653,7 @@ def initFunc(self, args):
 
             return per
 
-        # Listen for events for a few seconds
+        # Listen for events for a few seconds``
         if(args.listen != "True"):
             self.wait_events(int(args.listen))
             return
@@ -814,7 +821,7 @@ def rxTestFunc(self, args):
         self.send_command("01332003"+channel+phy+modulationIndex)
 
 
-    def endTestVSFunc(self, args) -> dict | None:
+    def endTestVSFunc(self, args):
         """
         Vendor specific command to end test\n
         Returns a dictionary of entire test report\n
@@ -856,6 +863,9 @@ def endTestFunc(self, args):
 
         # Parse the event and print the number of received packets
         try:
+
+            print("evtString", evtString)
+
             evtData = int(evtString, 16)
         except ValueError:
             print('Value Error Has occured. Response most likely empty')
@@ -942,6 +952,41 @@ def cmdFunc(self,args, timeout=None):
         else:
             self.send_command(args.cmd, timeout=timeout)
 
+    def readReg(self, addr, length):
+
+        # Reverse the bytes to LSB first
+        addrBytes = parseAddr(addr)
+
+        # Get the read length
+        readLen = length
+        if(readLen[:2] != "0x"):
+            print("Length must be a hex number starting with 0x")
+            return
+        readLen = readLen[2:]
+
+        # assert(readLen < 256)
+
+        readLenString = "%0.2X"%int(readLen, 16)
+        print(readLenString)
+        # Calculate the total length, 1 for the read len, 4 for the address length
+        totalLen = "%0.2X"%(1+4)
+
+        # Send the command and save the event
+        evtString = self.send_command("0101FF"+totalLen+readLenString+addrBytes)
+
+        print('Return String',evtString)
+        # print('EvtString Length', len(evtString))
+
+        # Get the data
+        evtString = evtString[14:]
+        # print('EvtString Length', len(evtString))
+        print(evtString)
+        # Split the data into bytes
+        chunks, chunk_size = len(evtString), 2
+        evtBytes = [ evtString[i:i+chunk_size] for i in range(0, chunks, chunk_size) ]
+
+        return evtBytes
+
     ## Read register function.
      #
      # Sends HCI command to read a register.
@@ -960,8 +1005,11 @@ def readRegFunc(self, args):
             print("Length must be a hex number starting with 0x")
             return
         readLen = readLen[2:]
-        readLenString = "%0.2X"%int(readLen, 16)
 
+        # assert(readLen < 256)
+
+        readLenString = "%0.2X"%int(readLen, 16)
+        print(readLenString)
         # Calculate the total length, 1 for the read len, 4 for the address length
         totalLen = "%0.2X"%(1+4)
 
@@ -975,6 +1023,7 @@ def readRegFunc(self, args):
         chunks, chunk_size = len(evtString), 2
         evtBytes = [ evtString[i:i+chunk_size] for i in range(0, chunks, chunk_size) ]
 
+
         # Print the data
         startingAddr = int(args.addr, 16)
 
@@ -999,7 +1048,7 @@ def readRegFunc(self, args):
                 print("__", end="")
             else:
                 print(evtBytes[lineAddr], end="")
-
+                
             # Print a new line at the end of the 32 bit value
             if(i%4 == 3):
                 print()

Tools/Bluetooth/calibration_init.py

@@ -50,6 +50,9 @@
 from BLE_hci import Namespace
 import socket
 import time
+import mxc_radio
+
+from json import JSONEncoder
 
 if socket.gethostname() == "wall-e":
     rf_switch = True
@@ -81,196 +84,75 @@ def printError(msg):
 
 # Parse the command line arguments
 parser = argparse.ArgumentParser(description=descText, formatter_class=RawTextHelpFormatter)
-parser.add_argument('slaveSerial',help='Serial port for slave device')
-parser.add_argument('masterSerial',help='Serial port for master device')
-parser.add_argument('results',help='CSV files to store the results')
-parser.add_argument('-d', '--delay', default=5,help='Number of seconds to wait before ending the test')
-parser.add_argument('-l', '--limit', default=0,help='PER limit for return value')
-parser.add_argument('-p', '--phys', default="1",help='PHYs to test with, comma separated list with 1-4.')
-parser.add_argument('-c', '--channel', default="0", help="Test channel, 0-39")
-parser.add_argument('-t', '--txpows', default="0",help='TX powers to test with, comma separated list.')
-parser.add_argument('-a', '--attens', help='Attenuation settings to use, comma separated list.')
-parser.add_argument('-s', '--step', default=10, help='Attenuation sweep step size in dBm.')
-parser.add_argument('-e', '--pktlen', default="250", help="packet length, comma separated list.")
-parser.add_argument('-n', '--numpkt', default=5,help='Number of packets in test.')
-parser.add_argument('--mtp', default="", help="master TRACE serial port")
-parser.add_argument('--stp', default="", help="slave TRACE serial port")
-
+parser.add_argument('serialPort',help='Serial port for slave device')
+parser.add_argument('-d', '--dbb', default=5,help='Read and dump DBB registers')
+parser.add_argument('-r', '--results', default='',help='File to store results')
+
 args = parser.parse_args()
 print(args)
 
 print("--------------------------------------------------------------------------------------------")
-packetLengths    = args.pktlen.strip().split(",")
-numPackets       = args.numpkt.strip().split(",")
-phys             = args.phys.strip().split(",")
-txPowers         = args.txpows.strip().split(",")
-chan             = args.channel.strip().split(",")
-
-if args.attens is None:
-    if int(args.step) == 0:
-        attens = [20, 70]
-    else:
-        attens = list(range(20, 90, int(args.step)))
-
-    # Add the max attenuation
-    attens.append(90)
-else:
-    attens = args.attens.strip().split(",")
 
-print("slaveSerial   :", args.slaveSerial)
-print("masterSerial  :", args.masterSerial)
+print("Serial Port   :", args.serialPort)
 print("results       :", args.results)
-print("delay         :", args.delay)
-print("packetLengths :", packetLengths)
-print("numPackets    :", numPackets)
-print("phys          :", phys)
-print("attens        :", attens)
-print("txPowers      :", txPowers)
-print("Channel       :", chan)
-print("PER limit     :", args.limit)
+
 
 # Open the results file, write the parameters
-results = open(args.results, "a")
-if 0:
-    results.write("# slaveSerial   : "+str(args.slaveSerial)+"\n")
-    results.write("# masterSerial  : "+str(args.masterSerial)+"\n")
-    results.write("# results       : "+str(args.results)+"\n")
-    results.write("# delay         : "+str(args.delay)+"\n")
-    results.write("# packetLengths : "+str(packetLengths)+"\n")
-    results.write("# numPackets    : "+str(numPackets)+"\n")
-    results.write("# phys          : "+str(phys)+"\n")
-    results.write("# attens        : "+str(attens)+"\n")
-    results.write("# txPower       : "+str(txPower)+"\n")
-    results.write("# Channel       : "+str(chan)+"\n")
-    results.write("# PER limit     : "+str(args.limit)+"\n")
-
-    # Write the header line
-    results.write("packetLen,numPkt,phy,atten,txPower,channel,perMaster,perSlave\n")
-
-# Create the BLE_hci objects
-hciSlave  = BLE_hci(Namespace(serialPort=args.slaveSerial,  monPort=args.stp, baud=115200, id=2))
-hciMaster = BLE_hci(Namespace(serialPort=args.masterSerial, monPort=args.mtp, baud=115200, id=1))
-
-perMax = 0
-
-def readDBB(hciInterface: BLE_hci) -> dict | None:
-    """
-    Function to read DBB register of device 
-    Return DBB registers as a struct
-    """
-
-    dbbReg = {}
-    DBB_START_REG= 0x00000000
-    DBB_SIZE = 0x4
-    evtBytes = hciInterface.readRegFunc(Namespace(addr=DBB_START_REG,length=DBB_SIZE))
-
-
-    if evtBytes is None:
-        printError('Failed to read DBB register. Returning None')
-        return None
-
-
-for packetLen, numPkt, phy, txPower, dtmCh in itertools.product(packetLengths, numPackets, phys, txPowers, chan):
-    per_100 = 0
-    for atten in attens:
-        RETRY = 2
-        while per_100 < RETRY:
-            start_secs = time.time()
-            print(f'\n---------------------------------------------------------------------------------------')
-            print(f'packetLen: {packetLen}, numPackets: {numPkt}, phy: {phy}, atten: {atten}, txPower: {txPower}, Channel: {dtmCh}\n')
-
-            print("Set the requested attenuation.")
-            if rf_switch:
-                mini_RCDAT = mini_RCDAT_USB(Namespace(atten=atten))
-            sleep(0.1)
-
-            print("\nReset the devices.")
-            hciSlave.resetFunc(None)
-            hciMaster.resetFunc(None)
-            sleep(0.1)
-
-            print("\nSet the PHY.")
-            hciMaster.phyFunc(Namespace(phy=str(phy)), timeout=1)
-            #hciMaster.listenFunc(Namespace(time=2, stats="False"))
-
-            print("\nSet the txPower.")
-            hciSlave.txPowerFunc(Namespace(power=txPower, handle="0")) 
-            hciMaster.txPowerFunc(Namespace(power=txPower, handle="0"))
-
-            print('--------------')
-            print("\nSet slave to RX.")
-            hciSlave.rxTestFunc(Namespace(channel=str(chan), phy=str(phy)))
-            print("\nSet master to TX, start test.")
-            hciMaster.txTestVSFunc(Namespace(channel=str(chan), phy=str(phy), packetLength=str(packetLen), numPackets=str(numPkt)))            
-
-            print(f"\nWait {args.delay} secs for the DTM Test to complete.")
-            sleep(int(args.delay))
-
-            print("\nEnd test.")
-            hciMaster.endTestFunc(None)
-            perSlave = hciSlave.endTestFunc(None) / int(numPkt) * 100
-
-            print('--------------')
-            print("\nReset the devices.")
-            hciSlave.resetFunc(None)
-            hciMaster.resetFunc(None)
-            sleep(0.1)
-
-            print("\nSet master to RX.")
-            hciMaster.rxTestFunc(Namespace(channel=str(chan), phy=str(phy)))
-            print("\nSet slave to TX, start test.")
-            hciSlave.txTestVSFunc(Namespace(channel=str(chan), phy=str(phy), packetLength=str(packetLen), numPackets=str(numPkt)))            
-
-            print(f"\nWait {args.delay} secs for the DTM Test to complete.")
-            sleep(int(args.delay))
-
-            print("\nEnd test.")
-            hciSlave.endTestFunc(None)
-            perMaster = hciMaster.endTestFunc(None) / int(numPkt) * 100
-
-            print("\nCollect results.")
-            print("perMaster  : ", perMaster)
-            print("perSlave   : ", perSlave)
-
-            if perMaster is None or perSlave is None:
-                per_100 += 1
-                print(f'Retry: {per_100}')
-                continue
-
-            # Record max per
-            if perMaster > perMax:
-                perMax = perMaster
-            if perSlave > perMax:
-                perMax = perSlave
-            print("perMax     : ", perMax)
-
-            break
-
-        if per_100 >= RETRY:
-            print(f'Tried {per_100} times, give up.')
-            perMaster = 100
-            perSlave = 100
-            perMax = 100
-
-        # Save the results to file
-        results.write(str(packetLen)+","+str(numPkt)+","+str(phy)+",-"+str(atten)+","+str(txPower)+","+str(dtmCh)+","+str(perMaster)+","+str(perSlave)+"\n")
-        end_secs = time.time()
-        print(f'\nUsed {(end_secs - start_secs):.0f} seconds.')
-
-print('--------------------------------------------------------------------------------------------')
-print("Reset the devices.")
-hciSlave.resetFunc(None)
-hciMaster.resetFunc(None)
-sleep(0.1)
-
-results.write("\n")
-results.close()
-
-print("perMax: ", perMax)
-
-if float(args.limit) != 0.0:
-    if perMax > float(args.limit):
-        print("PER too high!")
-        sys.exit(1)
-
-sys.exit(0)
+
+if args.results != '':
+    results = open(args.results, "a")
+
+
+
+
+
+class RegisterEncoder(JSONEncoder):
+    def default(self, o):
+            return o.__dict__
+
+
+
+def listEq(list1,list2):
+
+    if len(list1) != len(list2):
+        return False
+    length = len(list1)
+
+    for i in range(length):
+        if list1[i] != list2[i]:
+            return False
+
+    return True
+
+
+
+def verifyDBB(dbbReference):
+
+    #get expected dbb values
+
+
+    #make sure the all values match
+    dbbMatches = True
+
+
+    return dbbMatches
+
+
+def main():
+    # Create the BLE_hci objects
+    hciInterface  = BLE_hci(Namespace(serialPort=args.serialPort,  monPort="", baud=115200, id=1))
+
+    dbb = mxc_radio.DBB(hciInterface=hciInterface)
+    # ctrlReg = dbb.readCtrlReg()
+    rxReg = dbb.readRxReg()
+
+
+    # print(ctrlReg)
+    # print('Rx', rxReg)
+
+
+
+    sys.exit(0)
+
+if __name__ == "__main__":
+    main()