Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Removed some unneeded type hints to lower Python requirements #387

Merged
merged 20 commits into from
Feb 8, 2023
Merged
Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions Tools/Bluetooth/BLE_hci.py
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,7 @@ def send_command(self, packet, resp = True, delay = 0.01, print_cmd = True, time
else:
print(str(datetime.datetime.now()) + f" {self.id}>", packet)

print(packet)
self.port.write(bytearray.fromhex(packet))
sleep(delay)

Expand Down Expand Up @@ -646,7 +647,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
Expand Down Expand Up @@ -814,7 +815,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
Expand Down Expand Up @@ -856,6 +857,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')
Expand Down
286 changes: 107 additions & 179 deletions Tools/Bluetooth/calibration_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@
import socket
import time

from json import JSONEncoder

if socket.gethostname() == "wall-e":
rf_switch = True
else:
Expand Down Expand Up @@ -81,196 +83,122 @@ 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('serialPort',help='Serial port for slave 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('-d', '--dbb', default=5,help='Read and dump DBB registers')

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:


MXC_BASE_BTLE=0x40050000



class RegisterEncoder(JSONEncoder):
def default(self, o):
return o.__dict__
class DbbCtrlRegs:
pass
class DbbRxRegs:
pass
class DbbTxRegs:
pass
class DbbRffeRegs:
pass
class DBB:
def __init__(self,ctrlReg=None, rxReg=None, txReg=None, rffeReg=None):
self.ctrlReg = ctrlReg
self.rxReg = rxReg
self.txReg = txReg
self.rffeReg = rffeReg


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 readDbbCtrl(hciInterface):
"""
Read and return the DBB Ctrl Reg
"""

reg = MXC_BASE_BTLE + 0x1000


MXC_BASE_BTLE_DBB_CTRL = "0x%08X" % (reg)


#anything bigger than 128 bytes seems to cause a problem with the fromhex function
ctrlRegSize = "0x%02X" %(32 * 4)
regBase = reg
ctrlReg = []

#size of the ctrl reg is 304 bytes so just go through more than that
while reg < regBase + 128 * 3:

evtBytes = hciInterface.readRegFunc(Namespace(addr=MXC_BASE_BTLE_DBB_CTRL,length=ctrlRegSize))
ctrlReg.extend(evtBytes)
reg += 128

print(ctrlReg)
print('length', len(ctrlReg))




def readDbbRx(hciInterface):
MXC_BASE_BTLE_DBB_RX = MXC_BASE_BTLE + 0x3000

def readDbbTx(hciInterface):
MXC_BASE_BTLE_DBB_TX = MXC_BASE_BTLE + 0x2000
def readDbbRffe(hciInterface):
MXC_BASE_BTLE_DBB_EXT_RFFE = MXC_BASE_BTLE + 0x8000


def readDBB(hciInterface):
"""
Function to read DBB register of device
Return DBB registers as a struct
Return DBB registers as a DBB class
"""
dbb = DBB()

dbb.ctrlReg = readDbbCtrl(hciInterface)
dbb.txReg = readDbbRx(hciInterface)
dbb.rxReg = readDbbTx(hciInterface)
dbb.rffeReg = readDbbRffe(hciInterface)

return dbb


def main():
# Create the BLE_hci objects
hciInterface = BLE_hci(Namespace(serialPort=args.serialPort, monPort="", baud=115200, id=1))
hciInterface.resetFunc(None)
hciInterface.txTestFunc(Namespace(channel=0, phy=1, packetLength=0, payload=3))
dbb = readDBB(hciInterface)

hciInterface.endTestFunc(None)


sys.exit(0)

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 __name__ == "__main__":
main()