mirror of
https://github.com/espressif/openthread.git
synced 2026-08-28 21:09:54 +00:00
[thci] fix setXpanId bug and code cleanup (#4121)
This commit fixes a bug where setXpanId might raise exception when called with xPanId of length 16. This commit also renames __convertLongToString to __convertLongToHex to better reflect its functionality and optimizes its implementation.
This commit is contained in:
@@ -500,23 +500,22 @@ class OpenThread(IThci):
|
||||
strIp6Prefix = prefix[:19]
|
||||
return strIp6Prefix + '::'
|
||||
|
||||
def __convertLongToString(self, iValue):
|
||||
def __convertLongToHex(self, iValue, fillZeros=None):
|
||||
"""convert a long hex integer to string
|
||||
remove '0x' and 'L' return string
|
||||
|
||||
Args:
|
||||
iValue: long integer in hex format
|
||||
fillZeros: pad string with zeros on the left to specified width
|
||||
|
||||
Returns:
|
||||
string of this long integer without '0x' and 'L'
|
||||
"""
|
||||
string = ''
|
||||
strValue = str(hex(iValue))
|
||||
fmt = '%x'
|
||||
if fillZeros is not None:
|
||||
fmt = '%%0%dx' % fillZeros
|
||||
|
||||
string = strValue.lstrip('0x')
|
||||
string = string.rstrip('L')
|
||||
|
||||
return string
|
||||
return fmt % iValue
|
||||
|
||||
def __readCommissioningLogs(self, durationInSeconds):
|
||||
"""read logs during the commissioning process
|
||||
@@ -760,12 +759,7 @@ class OpenThread(IThci):
|
||||
address64 = self.mac
|
||||
|
||||
if not isinstance(xEUI, str):
|
||||
address64 = self.__convertLongToString(xEUI)
|
||||
|
||||
# prepend 0 at the beginning
|
||||
if len(address64) < 16:
|
||||
address64 = address64.zfill(16)
|
||||
print(address64)
|
||||
address64 = self.__convertLongToHex(xEUI, 16)
|
||||
else:
|
||||
address64 = xEUI
|
||||
|
||||
@@ -870,13 +864,7 @@ class OpenThread(IThci):
|
||||
print(key)
|
||||
try:
|
||||
if not isinstance(key, str):
|
||||
masterKey = self.__convertLongToString(key)
|
||||
|
||||
# prpend '0' at the beginning
|
||||
if len(masterKey) < 32:
|
||||
masterKey = masterKey.zfill(32)
|
||||
print(masterKey)
|
||||
|
||||
masterKey = self.__convertLongToHex(key, 32)
|
||||
cmd = 'masterkey %s' % masterKey
|
||||
datasetCmd = 'dataset masterkey %s' % masterKey
|
||||
else:
|
||||
@@ -910,7 +898,7 @@ class OpenThread(IThci):
|
||||
if isinstance(xEUI, str):
|
||||
macAddr = xEUI
|
||||
else:
|
||||
macAddr = self.__convertLongToString(xEUI)
|
||||
macAddr = self.__convertLongToHex(xEUI)
|
||||
|
||||
try:
|
||||
# if blocked device is itself
|
||||
@@ -952,7 +940,7 @@ class OpenThread(IThci):
|
||||
if isinstance(xEUI, str):
|
||||
macAddr = xEUI
|
||||
else:
|
||||
macAddr = self.__convertLongToString(xEUI)
|
||||
macAddr = self.__convertLongToHex(xEUI)
|
||||
|
||||
try:
|
||||
if self._addressfilterMode != 'whitelist':
|
||||
@@ -1871,14 +1859,9 @@ class OpenThread(IThci):
|
||||
print(xPanId)
|
||||
try:
|
||||
if not isinstance(xPanId, str):
|
||||
xpanid = self.__convertLongToString(xPanId)
|
||||
|
||||
# prepend '0' at the beginning
|
||||
if len(xpanid) < 16:
|
||||
xpanid = xpanid.zfill(16)
|
||||
print(xpanid)
|
||||
cmd = 'extpanid %s' % xpanid
|
||||
datasetCmd = 'dataset extpanid %s' % xpanid
|
||||
xpanid = self.__convertLongToHex(xPanId, 16)
|
||||
cmd = 'extpanid %s' % xpanid
|
||||
datasetCmd = 'dataset extpanid %s' % xpanid
|
||||
else:
|
||||
xpanid = xPanId
|
||||
cmd = 'extpanid %s' % xpanid
|
||||
@@ -2122,12 +2105,7 @@ class OpenThread(IThci):
|
||||
timeout = 500
|
||||
|
||||
if not isinstance(xEUI, str):
|
||||
eui64 = self.__convertLongToString(xEUI)
|
||||
|
||||
# prepend 0 at the beginning
|
||||
if len(eui64) < 16:
|
||||
eui64 = eui64.zfill(16)
|
||||
print(eui64)
|
||||
eui64 = self.__convertLongToHex(xEUI, 16)
|
||||
else:
|
||||
eui64 = xEUI
|
||||
|
||||
@@ -2317,7 +2295,7 @@ class OpenThread(IThci):
|
||||
"""
|
||||
print('%s call MGMT_ED_SCAN' % self.port)
|
||||
channelMask = ''
|
||||
channelMask = '0x' + self.__convertLongToString(
|
||||
channelMask = '0x' + self.__convertLongToHex(
|
||||
self.__convertChannelMask(listChannelMask)
|
||||
)
|
||||
try:
|
||||
@@ -2350,7 +2328,7 @@ class OpenThread(IThci):
|
||||
print('%s call MGMT_PANID_QUERY' % self.port)
|
||||
panid = ''
|
||||
channelMask = ''
|
||||
channelMask = '0x' + self.__convertLongToString(
|
||||
channelMask = '0x' + self.__convertLongToHex(
|
||||
self.__convertChannelMask(listChannelMask)
|
||||
)
|
||||
|
||||
@@ -2377,7 +2355,7 @@ class OpenThread(IThci):
|
||||
"""
|
||||
print('%s call MGMT_ANNOUNCE_BEGIN' % self.port)
|
||||
channelMask = ''
|
||||
channelMask = '0x' + self.__convertLongToString(
|
||||
channelMask = '0x' + self.__convertLongToHex(
|
||||
self.__convertChannelMask(listChannelMask)
|
||||
)
|
||||
try:
|
||||
@@ -2459,10 +2437,7 @@ class OpenThread(IThci):
|
||||
|
||||
if xExtendedPanId is not None:
|
||||
cmd += ' extpanid '
|
||||
xpanid = self.__convertLongToString(xExtendedPanId)
|
||||
|
||||
if len(xpanid) < 16:
|
||||
xpanid = xpanid.zfill(16)
|
||||
xpanid = self.__convertLongToHex(xExtendedPanId, 16)
|
||||
|
||||
cmd += xpanid
|
||||
|
||||
@@ -2480,10 +2455,7 @@ class OpenThread(IThci):
|
||||
|
||||
if xMasterKey is not None:
|
||||
cmd += ' masterkey '
|
||||
key = self.__convertLongToString(xMasterKey)
|
||||
|
||||
if len(key) < 32:
|
||||
key = key.zfill(32)
|
||||
key = self.__convertLongToHex(xMasterKey, 32)
|
||||
|
||||
cmd += key
|
||||
|
||||
@@ -2493,7 +2465,7 @@ class OpenThread(IThci):
|
||||
|
||||
if listChannelMask is not None:
|
||||
cmd += ' channelmask '
|
||||
cmd += '0x' + self.__convertLongToString(
|
||||
cmd += '0x' + self.__convertLongToHex(
|
||||
self.__convertChannelMask(listChannelMask)
|
||||
)
|
||||
|
||||
@@ -2579,7 +2551,7 @@ class OpenThread(IThci):
|
||||
cmd += locator
|
||||
|
||||
if xSteeringData is not None:
|
||||
steeringData = self.__convertLongToString(xSteeringData)
|
||||
steeringData = self.__convertLongToHex(xSteeringData)
|
||||
cmd += '08' + str(len(steeringData) / 2).zfill(2)
|
||||
cmd += steeringData
|
||||
|
||||
@@ -2670,10 +2642,7 @@ class OpenThread(IThci):
|
||||
|
||||
if xMasterKey is not None:
|
||||
cmd += ' masterkey '
|
||||
key = self.__convertLongToString(xMasterKey)
|
||||
|
||||
if len(key) < 32:
|
||||
key = key.zfill(32)
|
||||
key = self.__convertLongToHex(xMasterKey, 32)
|
||||
|
||||
cmd += key
|
||||
|
||||
|
||||
@@ -674,23 +674,22 @@ class OpenThread_WpanCtl(IThci):
|
||||
strIp6Prefix = prefix[:19]
|
||||
return strIp6Prefix + '::'
|
||||
|
||||
def __convertLongToString(self, iValue):
|
||||
def __convertLongToHex(self, iValue, fillZeros=None):
|
||||
"""convert a long hex integer to string
|
||||
remove '0x' and 'L' return string
|
||||
|
||||
Args:
|
||||
iValue: long integer in hex format
|
||||
fillZeros: pad string with zeros on the left to specified width
|
||||
|
||||
Returns:
|
||||
string of this long integer without '0x' and 'L'
|
||||
"""
|
||||
string = ''
|
||||
strValue = str(hex(iValue))
|
||||
fmt = '%x'
|
||||
if fillZeros is not None:
|
||||
fmt = '%%0%dx' % fillZeros
|
||||
|
||||
string = strValue.lstrip('0x')
|
||||
string = string.rstrip('L')
|
||||
|
||||
return string
|
||||
return fmt % iValue
|
||||
|
||||
def __convertChannelMask(self, channelsArray):
|
||||
"""convert channelsArray to bitmask format
|
||||
@@ -1019,12 +1018,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
address64 = self.mac
|
||||
|
||||
if not isinstance(xEUI, str):
|
||||
address64 = self.__convertLongToString(xEUI)
|
||||
|
||||
# prepend 0 at the beginning
|
||||
if len(address64) < 16:
|
||||
address64 = address64.zfill(16)
|
||||
print(address64)
|
||||
address64 = self.__convertLongToHex(xEUI, 16)
|
||||
else:
|
||||
address64 = xEUI
|
||||
|
||||
@@ -1180,12 +1174,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
|
||||
try:
|
||||
if not isinstance(key, str):
|
||||
masterKey = self.__convertLongToString(key)
|
||||
|
||||
# prpend '0' at the beginning
|
||||
if len(masterKey) < 32:
|
||||
masterKey = masterKey.zfill(32)
|
||||
|
||||
masterKey = self.__convertLongToHex(key, 32)
|
||||
cmd = WPANCTL_CMD + 'setprop Network:Key %s' % masterKey
|
||||
datasetCmd = (
|
||||
WPANCTL_CMD + 'setprop Dataset:MasterKey %s' % masterKey
|
||||
@@ -1223,7 +1212,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
if isinstance(xEUI, str):
|
||||
macAddr = xEUI
|
||||
else:
|
||||
macAddr = self.__convertLongToString(xEUI)
|
||||
macAddr = self.__convertLongToHex(xEUI)
|
||||
|
||||
try:
|
||||
# if blocked device is itself
|
||||
@@ -1264,7 +1253,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
if isinstance(xEUI, str):
|
||||
macAddr = xEUI
|
||||
else:
|
||||
macAddr = self.__convertLongToString(xEUI)
|
||||
macAddr = self.__convertLongToHex(xEUI)
|
||||
|
||||
try:
|
||||
if self._addressfilterMode != 'whitelist':
|
||||
@@ -2125,17 +2114,13 @@ class OpenThread_WpanCtl(IThci):
|
||||
print(xPanId)
|
||||
try:
|
||||
if not isinstance(xPanId, str):
|
||||
xpanid = self.__convertLongToString(xPanId)
|
||||
xpanid = self.__convertLongToHex(xPanId, 16)
|
||||
|
||||
# prepend '0' at the beginning
|
||||
if len(xpanid) < 16:
|
||||
xpanid = xpanid.zfill(16)
|
||||
print(xpanid)
|
||||
cmd = WPANCTL_CMD + 'setprop Network:XPANID %s' % xpanid
|
||||
datasetCmd = (
|
||||
WPANCTL_CMD
|
||||
+ 'setprop Dataset:ExtendedPanId %s' % xpanid
|
||||
)
|
||||
cmd = WPANCTL_CMD + 'setprop Network:XPANID %s' % xpanid
|
||||
datasetCmd = (
|
||||
WPANCTL_CMD
|
||||
+ 'setprop Dataset:ExtendedPanId %s' % xpanid
|
||||
)
|
||||
else:
|
||||
xpanid = xPanId
|
||||
cmd = WPANCTL_CMD + 'setprop Network:XPANID %s' % xpanid
|
||||
@@ -2354,12 +2339,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
"""
|
||||
print('%s call scanJoiner' % self.port)
|
||||
if not isinstance(xEUI, str):
|
||||
eui64 = self.__convertLongToString(xEUI)
|
||||
|
||||
# prepend 0 at the beginning
|
||||
if len(eui64) < 16:
|
||||
eui64 = eui64.zfill(16)
|
||||
print(eui64)
|
||||
eui64 = self.__convertLongToHex(xEUI, 16)
|
||||
else:
|
||||
eui64 = xEUI
|
||||
|
||||
@@ -2705,9 +2685,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
return False
|
||||
|
||||
if xExtendedPanId is not None:
|
||||
xpanid = self.__convertLongToString(xExtendedPanId)
|
||||
if len(xpanid) < 16:
|
||||
xpanid = xpanid.zfill(16)
|
||||
xpanid = self.__convertLongToHex(xExtendedPanId, 16)
|
||||
setExtendedPanIdCmd = (
|
||||
WPANCTL_CMD + 'setprop Dataset:ExtendedPanId ' + xpanid
|
||||
)
|
||||
@@ -2740,11 +2718,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
return False
|
||||
|
||||
if xMasterKey is not None:
|
||||
key = self.__convertLongToString(xMasterKey)
|
||||
|
||||
if len(key) < 32:
|
||||
key = key.zfill(32)
|
||||
|
||||
key = self.__convertLongToHex(xMasterKey, 32)
|
||||
setMasterKeyCmd = (
|
||||
WPANCTL_CMD + 'setprop Dataset:MasterKey ' + key
|
||||
)
|
||||
@@ -2763,7 +2737,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
WPANCTL_CMD
|
||||
+ 'setprop Dataset:ChannelMaskPage0 '
|
||||
+ '0x'
|
||||
+ self.__convertLongToString(
|
||||
+ self.__convertLongToHex(
|
||||
self.__convertChannelMask(listChannelMask)
|
||||
)
|
||||
)
|
||||
@@ -2852,7 +2826,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
setRawTLVCmd += locator
|
||||
|
||||
if xSteeringData is not None:
|
||||
steeringData = self.__convertLongToString(xSteeringData)
|
||||
steeringData = self.__convertLongToHex(xSteeringData)
|
||||
setRawTLVCmd += '08' + str(len(steeringData) / 2).zfill(2)
|
||||
setRawTLVCmd += steeringData
|
||||
|
||||
@@ -2993,11 +2967,7 @@ class OpenThread_WpanCtl(IThci):
|
||||
return False
|
||||
|
||||
if xMasterKey is not None:
|
||||
key = self.__convertLongToString(xMasterKey)
|
||||
|
||||
if len(key) < 32:
|
||||
key = key.zfill(32)
|
||||
|
||||
key = self.__convertLongToHex(xMasterKey, 32)
|
||||
setMasterKeyCmd = (
|
||||
WPANCTL_CMD + 'setprop Dataset:MasterKey ' + key
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user