#!/usr/bin/env python # # Copyright (c) 2016, The OpenThread Authors. # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # 1. Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # 2. Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # 3. Neither the name of the copyright holder nor the # names of its contributors may be used to endorse or promote products # derived from this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. """ >> Thread Host Controller Interface >> Device : OpenThread THCI >> Class : OpenThread """ import re import time import socket import logging from Queue import Queue import serial from IThci import IThci from GRLLibs.UtilityModules.Test import Thread_Device_Role, Device_Data_Requirement, MacType from GRLLibs.UtilityModules.enums import PlatformDiagnosticPacket_Direction, PlatformDiagnosticPacket_Type from GRLLibs.UtilityModules.ModuleHelper import ModuleHelper, ThreadRunner from GRLLibs.ThreadPacket.PlatformPackets import PlatformDiagnosticPacket, PlatformPackets from GRLLibs.UtilityModules.Plugins.AES_CMAC import Thread_PBKDF2 LINESEPX = re.compile(r'\r\n|\n') """regex: used to split lines""" class OpenThread(IThci): LOWEST_POSSIBLE_PARTATION_ID = 0x1 LINK_QUALITY_CHANGE_TIME = 100 # Used for reference firmware version control for Test Harness. # This variable will be updated to match the OpenThread reference firmware officially released. firmwarePrefix = "OPENTHREAD/201806" #def __init__(self, SerialPort=COMPortName, EUI=MAC_Address): def __init__(self, **kwargs): """initialize the serial port and default network parameters Args: **kwargs: Arbitrary keyword arguments Includes 'EUI' and 'SerialPort' """ try: self.UIStatusMsg = '' self.mac = kwargs.get('EUI') self.port = kwargs.get('SerialPort') self.handle = None self.AutoDUTEnable = False self._is_net = False # whether device is through ser2net self.logStatus = {'stop':'stop', 'running':'running', "pauseReq":'pauseReq', 'paused':'paused'} self.joinStatus = {'notstart':'notstart', 'ongoing':'ongoing', 'succeed':'succeed', "failed":'failed'} self.logThreadStatus = self.logStatus['stop'] self.intialize() except Exception, e: ModuleHelper.WriteIntoDebugLogger("initialize() Error: " + str(e)) def __del__(self): """close the serial port connection""" try: self.closeConnection() self.deviceConnected = False except Exception, e: ModuleHelper.WriteIntoDebugLogger("delete() Error: " + str(e)) def _expect(self, expected, times=50): """Find the `expected` line within `times` trials. Args: expected str: the expected string times int: number of trials """ print '[%s] Expecting [%s]' % (self.port, expected) retry_times = 10 while times > 0 and retry_times > 0: line = self._readline() print '[%s] Got line [%s]' % (self.port, line) if line == expected: print '[%s] Expected [%s]' % (self.port, expected) return if not line: retry_times -= 1 time.sleep(0.1) times -= 1 raise Exception('failed to find expected string[%s]' % expected) def _read(self, size=512): logging.info('%s: reading', self.port) if self._is_net: return self.handle.recv(size) else: return self.handle.read(size) def _write(self, data): logging.info('%s: writing', self.port) if self._is_net: self.handle.sendall(data) else: self.handle.write(data) def _readline(self): """Read exactly one line from the device Returns: None on no data """ logging.info('%s: reading line', self.port) if len(self._lines) > 1: return self._lines.pop(0) tail = '' if len(self._lines): tail = self._lines.pop() try: tail += self._read() except socket.error: logging.exception('%s: No new data', self.port) time.sleep(0.1) self._lines += LINESEPX.split(tail) if len(self._lines) > 1: return self._lines.pop(0) def _sendline(self, line): """Send exactly one line to the device Args: line str: data send to device """ logging.info('%s: sending line', self.port) # clear buffer self._lines = [] try: self._read() except socket.error: logging.debug('%s: Nothing cleared', self.port) print 'sending [%s]' % line self._write(line + '\r\n') # wait for write to complete time.sleep(0.1) def __sendCommand(self, cmd): """send specific command to reference unit over serial port Args: cmd: OpenThread CLI string Returns: Done: successfully send the command to reference unit and parse it Value: successfully retrieve the desired value from reference unit Error: some errors occur, indicates by the followed specific error number """ logging.info('%s: sendCommand[%s]', self.port, cmd) if self.logThreadStatus == self.logStatus['running']: self.logThreadStatus = self.logStatus['pauseReq'] while self.logThreadStatus != self.logStatus['paused'] and self.logThreadStatus != self.logStatus['stop']: pass try: # command retransmit times retry_times = 3 while retry_times > 0: retry_times -= 1 try: self._sendline(cmd) self._expect(cmd) except Exception as e: logging.exception('%s: failed to send command[%s]: %s', self.port, cmd, str(e)) if retry_times == 0: raise else: break line = None response = [] retry_times = 10 while retry_times > 0: line = self._readline() logging.info('%s: the read line is[%s]', self.port, line) if line: response.append(line) if line == 'Done': break else: retry_times -= 1 time.sleep(0.2) if line != 'Done': raise Exception('%s: failed to find end of response' % self.port) logging.info('%s: send command[%s] done!', self.port, cmd) return response except Exception, e: ModuleHelper.WriteIntoDebugLogger("sendCommand() Error: " + str(e)) raise def __getIp6Address(self, addressType): """get specific type of IPv6 address configured on thread device Args: addressType: the specific type of IPv6 address link local: link local unicast IPv6 address that's within one-hop scope global: global unicast IPv6 address rloc: mesh local unicast IPv6 address for routing in thread network mesh EID: mesh Endpoint Identifier Returns: IPv6 address string """ addrType = ['link local', 'global', 'rloc', 'mesh EID'] addrs = [] globalAddr = [] linkLocal64Addr = '' rlocAddr = '' meshEIDAddr = '' addrs = self.__sendCommand('ipaddr') for ip6Addr in addrs: if ip6Addr == 'Done': break ip6AddrPrefix = ip6Addr.split(':')[0] if ip6AddrPrefix == 'fe80': # link local address if ip6Addr.split(':')[4] != '0': linkLocal64Addr = ip6Addr elif ip6AddrPrefix == 'fd00': # mesh local address if ip6Addr.split(':')[4] == '0': # rloc rlocAddr = ip6Addr else: # mesh EID meshEIDAddr = ip6Addr else: # global ipv6 address if ip6Addr != None: globalAddr.append(ip6Addr) else: pass if addressType == addrType[0]: return linkLocal64Addr elif addressType == addrType[1]: return globalAddr elif addressType == addrType[2]: return rlocAddr elif addressType == addrType[3]: return meshEIDAddr else: pass def __setDeviceMode(self, mode): """set thread device mode: Args: mode: thread device mode r: rx-on-when-idle s: secure IEEE 802.15.4 data request d: full function device n: full network data Returns: True: successful to set the device mode False: fail to set the device mode """ print 'call __setDeviceMode' print mode try: cmd = 'mode %s' % mode return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setDeviceMode() Error: " + str(e)) def __setRouterUpgradeThreshold(self, iThreshold): """set router upgrade threshold Args: iThreshold: the number of active routers on the Thread network partition below which a REED may decide to become a Router. Returns: True: successful to set the ROUTER_UPGRADE_THRESHOLD False: fail to set ROUTER_UPGRADE_THRESHOLD """ print 'call __setRouterUpgradeThreshold' try: cmd = 'routerupgradethreshold %s' % str(iThreshold) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setRouterUpgradeThreshold() Error: " + str(e)) def __setRouterDowngradeThreshold(self, iThreshold): """set router downgrade threshold Args: iThreshold: the number of active routers on the Thread network partition above which an active router may decide to become a child. Returns: True: successful to set the ROUTER_DOWNGRADE_THRESHOLD False: fail to set ROUTER_DOWNGRADE_THRESHOLD """ print 'call __setRouterDowngradeThreshold' try: cmd = 'routerdowngradethreshold %s' % str(iThreshold) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setRouterDowngradeThreshold() Error: " + str(e)) def __setRouterSelectionJitter(self, iRouterJitter): """set ROUTER_SELECTION_JITTER parameter for REED to upgrade to Router Args: iRouterJitter: a random period prior to request Router ID for REED Returns: True: successful to set the ROUTER_SELECTION_JITTER False: fail to set ROUTER_SELECTION_JITTER """ print 'call _setRouterSelectionJitter' try: cmd = 'routerselectionjitter %s' % str(iRouterJitter) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setRouterSelectionJitter() Error: " + str(e)) def __setAddressfilterMode(self, mode): """set address filter mode Returns: True: successful to set address filter mode. False: fail to set address filter mode. """ print 'call setAddressFilterMode() ' + mode try: cmd = 'macfilter addr ' + mode if self.__sendCommand(cmd)[0] == 'Done': return True return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("__setAddressFilterMode() Error: " + str(e)) def __startOpenThread(self): """start OpenThread stack Returns: True: successful to start OpenThread stack and thread interface up False: fail to start OpenThread stack """ print 'call startOpenThread' try: if self.hasActiveDatasetToCommit: if self.__sendCommand('dataset commit active')[0] != 'Done': raise Exception('failed to commit active dataset') else: self.hasActiveDatasetToCommit = False # restore whitelist/blacklist address filter mode if rejoin after reset if self.isPowerDown: if self._addressfilterMode == 'whitelist': if self.__setAddressfilterMode('whitelist'): for addr in self._addressfilterSet: self.addAllowMAC(addr) elif self._addressfilterMode == 'blacklist': if self.__setAddressfilterMode('blacklist'): for addr in self._addressfilterSet: self.addBlockedMAC(addr) if self.__sendCommand('ifconfig up')[0] == 'Done': self.__setRouterSelectionJitter(1) if self.__sendCommand('thread start')[0] == 'Done': self.isPowerDown = False return True else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("startOpenThread() Error: " + str(e)) def __stopOpenThread(self): """stop OpenThread stack Returns: True: successful to stop OpenThread stack and thread interface down False: fail to stop OpenThread stack """ print 'call stopOpenThread' try: if self.__sendCommand('thread stop')[0] == 'Done': return self.__sendCommand('ifconfig down')[0] == 'Done' else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("stopOpenThread() Error: " + str(e)) def __isOpenThreadRunning(self): """check whether or not OpenThread is running Returns: True: OpenThread is running False: OpenThread is not running """ print 'call isOpenThreadRunning' return self.__sendCommand('state')[0] != 'disabled' # rloc16 might be hex string or integer, need to return actual allocated router id def __convertRlocToRouterId(self, xRloc16): """mapping Rloc16 to router id Args: xRloc16: hex rloc16 short address Returns: actual router id allocated by leader """ routerList = [] routerList = self.__sendCommand('router list')[0].split() print routerList print xRloc16 for index in routerList: router = [] cmd = 'router %s' % index router = self.__sendCommand(cmd) for line in router: if 'Done' in line: break elif 'Router ID' in line: routerid = line.split()[2] elif 'Rloc' in line: rloc16 = line.split()[1] else: pass # process input rloc16 if isinstance(xRloc16, str): rloc16 = '0x' + rloc16 if rloc16 == xRloc16: return routerid elif isinstance(xRloc16, int): if int(rloc16, 16) == xRloc16: return routerid else: pass return None def __convertIp6PrefixStringToIp6Address(self, strIp6Prefix): """convert IPv6 prefix string to IPv6 dotted-quad format for example: 2001000000000000 -> 2001:: Args: strIp6Prefix: IPv6 address string Returns: IPv6 address dotted-quad format """ prefix1 = strIp6Prefix.rstrip('L') prefix2 = prefix1.lstrip("0x") hexPrefix = str(prefix2).ljust(16,'0') hexIter = iter(hexPrefix) finalMac = ':'.join(a + b + c + d for a,b,c,d in zip(hexIter, hexIter,hexIter,hexIter)) prefix = str(finalMac) strIp6Prefix = prefix[:20] return strIp6Prefix +':' def __convertLongToString(self, iValue): """convert a long hex integer to string remove '0x' and 'L' return string Args: iValue: long integer in hex format Returns: string of this long integer without "0x" and "L" """ string = '' strValue = str(hex(iValue)) string = strValue.lstrip('0x') string = string.rstrip('L') return string def __readCommissioningLogs(self, durationInSeconds): """read logs during the commissioning process Args: durationInSeconds: time duration for reading commissioning logs Returns: Commissioning logs """ self.logThreadStatus = self.logStatus['running'] logs = Queue() t_end = time.time() + durationInSeconds while time.time() < t_end: time.sleep(0.3) if self.logThreadStatus == self.logStatus['pauseReq']: self.logThreadStatus = self.logStatus['paused'] if self.logThreadStatus != self.logStatus['running']: continue try: line = self._readline() if line: print line logs.put(line) if "Join success" in line: self.joinCommissionedStatus = self.joinStatus['succeed'] break elif "Join failed" in line: self.joinCommissionedStatus = self.joinStatus['failed'] break except Exception: pass self.logThreadStatus = self.logStatus['stop'] return logs def __convertChannelMask(self, channelsArray): """convert channelsArray to bitmask format Args: channelsArray: channel array (i.e. [21, 22]) Returns: bitmask format corresponding to a given channel array """ maskSet = 0 for eachChannel in channelsArray: mask = 1 << eachChannel maskSet = (maskSet | mask) return maskSet def __setChannelMask(self, channelMask): print 'call _setChannelMask' try: cmd = 'dataset channelmask %s' % channelMask self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setChannelMask() Error: " + str(e)) def __setSecurityPolicy(self, securityPolicySecs, securityPolicyFlags): print 'call _setSecurityPolicy' try: cmd = 'dataset securitypolicy %s %s' % (str(securityPolicySecs), securityPolicyFlags) self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setSecurityPolicy() Error: " + str(e)) def __setKeySwitchGuardTime(self, iKeySwitchGuardTime): """ set the Key switch guard time Args: iKeySwitchGuardTime: key switch guard time Returns: True: successful to set key switch guard time False: fail to set key switch guard time """ print '%s call setKeySwitchGuardTime' % self.port print iKeySwitchGuardTime try: cmd = 'keysequence guardtime %s' % str(iKeySwitchGuardTime) if self.__sendCommand(cmd)[0] == 'Done': time.sleep(1) return True else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("setKeySwitchGuardTime() Error; " + str(e)) def __getCommissionerSessionId(self): """ get the commissioner session id allocated from Leader """ print '%s call getCommissionerSessionId' % self.port return self.__sendCommand('commissioner sessionid')[0] def _connect(self): print 'My port is %s' % self.port if self.port.startswith('COM'): self.handle = serial.Serial(self.port, 115200, timeout=0) time.sleep(1) self.handle.write('\r\n') time.sleep(0.1) self._is_net = False elif ':' in self.port: host, port = self.port.split(':') self.handle = socket.create_connection((host, port)) self.handle.setblocking(0) self._is_net = True else: raise Exception('Unknown port schema') self.UIStatusMsg = self.getVersionNumber() def closeConnection(self): """close current serial port connection""" print '%s call closeConnection' % self.port try: if self.handle: self.handle.close() self.handle = None except Exception, e: ModuleHelper.WriteIntoDebugLogger("closeConnection() Error: " + str(e)) def intialize(self): """initialize the serial port with baudrate, timeout parameters""" print '%s call intialize' % self.port try: self.deviceConnected = False # init serial port self._connect() if self.firmwarePrefix in self.UIStatusMsg: self.deviceConnected = True else: self.UIStatusMsg = "Firmware Not Matching Expecting " + self.firmwarePrefix + " Now is " + self.UIStatusMsg ModuleHelper.WriteIntoDebugLogger("Err: OpenThread device Firmware not matching..") except Exception, e: ModuleHelper.WriteIntoDebugLogger("intialize() Error: " + str(e)) self.deviceConnected = False def setNetworkName(self, networkName='GRL'): """set Thread Network name Args: networkName: the networkname string to be set Returns: True: successful to set the Thread Networkname False: fail to set the Thread Networkname """ print '%s call setNetworkName' % self.port print networkName try: cmd = 'networkname %s' % networkName datasetCmd = 'dataset networkname %s' % networkName self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' and self.__sendCommand(datasetCmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setNetworkName() Error: " + str(e)) def getNetworkName(self): """get Thread Network name""" print '%s call getNetworkname' % self.port return self.__sendCommand('networkname')[0] def setChannel(self, channel=11): """set channel of Thread device operates on. Args: channel: (0 - 10: Reserved) (11 - 26: 2.4GHz channels) (27 - 65535: Reserved) Returns: True: successful to set the channel False: fail to set the channel """ print '%s call setChannel' % self.port print channel try: cmd = 'channel %s' % channel datasetCmd = 'dataset channel %s' % channel self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' and self.__sendCommand(datasetCmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setChannel() Error: " + str(e)) def getChannel(self): """get current channel""" print '%s call getChannel' % self.port return self.__sendCommand('channel')[0] def setMAC(self, xEUI): """set the extended addresss of Thread device Args: xEUI: extended address in hex format Returns: True: successful to set the extended address False: fail to set the extended address """ print '%s call setMAC' % self.port print xEUI address64 = '' try: if not xEUI: 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 else: address64 = xEUI cmd = 'extaddr %s' % address64 print cmd if self.__sendCommand(cmd)[0] == 'Done': self.mac = address64 return True else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("setMAC() Error: " + str(e)) def getMAC(self, bType=MacType.RandomMac): """get one specific type of MAC address currently OpenThread only supports Random MAC address Args: bType: indicate which kind of MAC address is required Returns: specific type of MAC address """ print '%s call getMAC' % self.port print bType # if power down happens, return extended address assigned previously if self.isPowerDown: macAddr64 = self.mac else: if bType == MacType.FactoryMac: macAddr64 = self.__sendCommand('eui64')[0] elif bType == MacType.HashMac: macAddr64 = self.__sendCommand('joinerid')[0] else: macAddr64 = self.__sendCommand('extaddr')[0] print macAddr64 return int(macAddr64, 16) def getLL64(self): """get link local unicast IPv6 address""" print '%s call getLL64' % self.port return self.__getIp6Address('link local') def getMLEID(self): """get mesh local endpoint identifier address""" print '%s call getMLEID' % self.port return self.__getIp6Address('mesh EID') def getRloc16(self): """get rloc16 short address""" print '%s call getRloc16' % self.port rloc16 = self.__sendCommand('rloc16')[0] return int(rloc16, 16) def getRloc(self): """get router locator unicast IPv6 address""" print '%s call getRloc' % self.port return self.__getIp6Address('rloc') def getGlobal(self): """get global unicast IPv6 address set if configuring multiple entries """ print '%s call getGlobal' % self.port return self.__getIp6Address('global') def setNetworkKey(self, key): """set Thread Network master key Args: key: Thread Network master key used in secure the MLE/802.15.4 packet Returns: True: successful to set the Thread Network master key False: fail to set the Thread Network master key """ masterKey = '' print '%s call setNetworkKey' % self.port 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 cmd = 'masterkey %s' %masterKey datasetCmd = 'dataset masterkey %s' % masterKey else: masterKey = key cmd = 'masterkey %s' % masterKey datasetCmd = 'dataset masterkey %s' % masterKey self.networkKey = masterKey self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' and self.__sendCommand(datasetCmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setNetworkkey() Error: " + str(e)) def getNetworkKey(self): """get the current Thread Network master key""" print '%s call getNetwokKey' % self.port return self.networkKey def addBlockedMAC(self, xEUI): """add a given extended address to the blacklist entry Args: xEUI: extended address in hex format Returns: True: successful to add a given extended address to the blacklist entry False: fail to add a given extended address to the blacklist entry """ print '%s call addBlockedMAC' % self.port print xEUI if isinstance(xEUI, str): macAddr = xEUI else: macAddr = self.__convertLongToString(xEUI) try: # if blocked device is itself if macAddr == self.mac: print 'block device itself' return True if self._addressfilterMode != 'blacklist': if self.__setAddressfilterMode('blacklist'): self._addressfilterMode = 'blacklist' cmd = 'macfilter addr add %s' % macAddr print cmd ret = self.__sendCommand(cmd)[0] == 'Done' self._addressfilterSet.add(macAddr) print 'current blacklist entries:' for addr in self._addressfilterSet: print addr return ret except Exception, e: ModuleHelper.WriteIntoDebugLogger("addBlockedMAC() Error: " + str(e)) def addAllowMAC(self, xEUI): """add a given extended address to the whitelist addressfilter Args: xEUI: a given extended address in hex format Returns: True: successful to add a given extended address to the whitelist entry False: fail to add a given extended address to the whitelist entry """ print '%s call addAllowMAC' % self.port print xEUI if isinstance(xEUI, str): macAddr = xEUI else: macAddr = self.__convertLongToString(xEUI) try: if self._addressfilterMode != 'whitelist': if self.__setAddressfilterMode('whitelist'): self._addressfilterMode = 'whitelist' cmd = 'macfilter addr add %s' % macAddr print cmd ret = self.__sendCommand(cmd)[0] == 'Done' self._addressfilterSet.add(macAddr) print 'current whitelist entries:' for addr in self._addressfilterSet: print addr return ret except Exception, e: ModuleHelper.WriteIntoDebugLogger("addAllowMAC() Error: " + str(e)) def clearBlockList(self): """clear all entries in blacklist table Returns: True: successful to clear the blacklist False: fail to clear the blacklist """ print '%s call clearBlockList' % self.port # remove all entries in blacklist try: print 'clearing blacklist entries:' for addr in self._addressfilterSet: print addr # disable blacklist if self.__setAddressfilterMode('disable'): self._addressfilterMode = 'disable' # clear ops cmd = 'macfilter addr clear' if self.__sendCommand(cmd)[0] == 'Done': self._addressfilterSet.clear() return True return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("clearBlockList() Error: " + str(e)) def clearAllowList(self): """clear all entries in whitelist table Returns: True: successful to clear the whitelist False: fail to clear the whitelist """ print '%s call clearAllowList' % self.port # remove all entries in whitelist try: print 'clearing whitelist entries:' for addr in self._addressfilterSet: print addr # disable whitelist if self.__setAddressfilterMode('disable'): self._addressfilterMode = 'disable' # clear ops cmd = 'macfilter addr clear' if self.__sendCommand(cmd)[0] == 'Done': self._addressfilterSet.clear() return True return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("clearAllowList() Error: " + str(e)) def getDeviceRole(self): """get current device role in Thread Network""" print '%s call getDeviceRole' % self.port return self.__sendCommand('state')[0] def joinNetwork(self, eRoleId): """make device ready to join the Thread Network with a given role Args: eRoleId: a given device role id Returns: True: ready to set Thread Network parameter for joining desired Network """ print '%s call joinNetwork' % self.port print eRoleId self.deviceRole = eRoleId mode = '' try: if ModuleHelper.LeaderDutChannelFound: self.channel = ModuleHelper.Default_Channel # FIXME: when Harness call setNetworkDataRequirement()? # only sleep end device requires stable networkdata now if eRoleId == Thread_Device_Role.Leader: print 'join as leader' mode = 'rsdn' if self.AutoDUTEnable is False: # set ROUTER_DOWNGRADE_THRESHOLD self.__setRouterDowngradeThreshold(33) elif eRoleId == Thread_Device_Role.Router: print 'join as router' mode = 'rsdn' if self.AutoDUTEnable is False: # set ROUTER_DOWNGRADE_THRESHOLD self.__setRouterDowngradeThreshold(33) elif eRoleId == Thread_Device_Role.SED: print 'join as sleepy end device' mode = 's' self.setPollingRate(self.sedPollingRate) elif eRoleId == Thread_Device_Role.EndDevice: print 'join as end device' mode = 'rsn' elif eRoleId == Thread_Device_Role.REED: print 'join as REED' mode = 'rsdn' # set ROUTER_UPGRADE_THRESHOLD self.__setRouterUpgradeThreshold(0) elif eRoleId == Thread_Device_Role.EndDevice_FED: # always remain an ED, never request to be a router print 'join as FED' mode = 'rsdn' # set ROUTER_UPGRADE_THRESHOLD self.__setRouterUpgradeThreshold(0) elif eRoleId == Thread_Device_Role.EndDevice_MED: print 'join as MED' mode = 'rsn' else: pass # set Thread device mode with a given role self.__setDeviceMode(mode) self.__setKeySwitchGuardTime(0) # start OpenThread self.__startOpenThread() time.sleep(3) return True except Exception, e: ModuleHelper.WriteIntoDebugLogger("joinNetwork() Error: " + str(e)) def getNetworkFragmentID(self): """get current partition id of Thread Network Partition from LeaderData Returns: The Thread network Partition Id """ print '%s call getNetworkFragmentID' % self.port if not self.__isOpenThreadRunning(): print 'OpenThread is not running' return None leaderData = [] leaderData = self.__sendCommand('leaderdata') return int(leaderData[0].split()[2], 16) def getParentAddress(self): """get Thread device's parent extended address and rloc16 short address Returns: The extended address of parent in hex format """ print '%s call getParentAddress' % self.port parentInfo = [] parentInfo = self.__sendCommand('parent') for line in parentInfo: if 'Done' in line: break elif 'Ext Addr' in line: eui = line.split()[2] print eui #elif 'Rloc' in line: # rloc16 = line.split()[1] # print rloc16 else: pass return int(eui, 16) def powerDown(self): """power down the Thread device""" print '%s call powerDown' % self.port self._sendline('reset') self.isPowerDown = True def powerUp(self): """power up the Thread device""" print '%s call powerUp' % self.port if not self.handle: self._connect() self.isPowerDown = False if not self.__isOpenThreadRunning(): self.__startOpenThread() def reboot(self): """reset and rejoin to Thread Network without any timeout Returns: True: successful to reset and rejoin the Thread Network False: fail to reset and rejoin the Thread Network """ print '%s call reboot' % self.port try: self._sendline('reset') self.isPowerDown = True time.sleep(3) self.__startOpenThread() time.sleep(3) if self.__sendCommand('state')[0] == 'disabled': print '[FAIL] reboot' return False else: return True except Exception, e: ModuleHelper.WriteIntoDebugLogger("reboot() Error: " + str(e)) def ping(self, destination, length=20): """ send ICMPv6 echo request with a given length to a unicast destination address Args: destination: the unicast destination address of ICMPv6 echo request length: the size of ICMPv6 echo request payload """ print '%s call ping' % self.port print 'destination: %s' %destination try: cmd = 'ping %s %s' % (destination, str(length)) print cmd self._sendline(cmd) self._expect(cmd) # wait echo reply time.sleep(1) except Exception, e: ModuleHelper.WriteIntoDebugLogger("ping() Error: " + str(e)) def multicast_Ping(self, destination, length=20): """send ICMPv6 echo request with a given length to a multicast destination address Args: destination: the multicast destination address of ICMPv6 echo request length: the size of ICMPv6 echo request payload """ print '%s call multicast_Ping' % self.port print 'destination: %s' % destination try: cmd = 'ping %s %s' % (destination, str(length)) print cmd self._sendline(cmd) self._expect(cmd) # wait echo reply time.sleep(1) except Exception, e: ModuleHelper.WriteIntoDebugLogger("multicast_ping() Error: " + str(e)) def getVersionNumber(self): """get OpenThread stack firmware version number""" print '%s call getVersionNumber' % self.port return self.__sendCommand('version')[0] def setPANID(self, xPAN): """set Thread Network PAN ID Args: xPAN: a given PAN ID in hex format Returns: True: successful to set the Thread Network PAN ID False: fail to set the Thread Network PAN ID """ print '%s call setPANID' % self.port print xPAN panid = '' try: if not isinstance(xPAN, str): panid = str(hex(xPAN)) print panid cmd = 'panid %s' % panid datasetCmd = 'dataset panid %s' % panid self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' and self.__sendCommand(datasetCmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setPANID() Error: " + str(e)) def getPANID(self): """get current Thread Network PAN ID""" print '%s call getPANID' % self.port return self.__sendCommand('panid')[0] def reset(self): """factory reset""" print '%s call reset' % self.port try: self._sendline('factoryreset') self._read() except Exception, e: ModuleHelper.WriteIntoDebugLogger("reset() Error: " + str(e)) def removeRouter(self, xRouterId): """kickoff router with a given router id from the Thread Network Args: xRouterId: a given router id in hex format Returns: True: successful to remove the router from the Thread Network False: fail to remove the router from the Thread Network """ print '%s call removeRouter' % self.port print xRouterId routerId = '' routerId = self.__convertRlocToRouterId(xRouterId) print routerId if routerId == None: print 'no matched xRouterId' return False try: cmd = 'releaserouterid %s' % routerId return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("removeRouter() Error: " + str(e)) def setDefaultValues(self): """set default mandatory Thread Network parameter value""" print '%s call setDefaultValues' % self.port # initialize variables self.networkName = ModuleHelper.Default_NwkName self.networkKey = ModuleHelper.Default_NwkKey self.channel = ModuleHelper.Default_Channel self.channelMask = "0x7fff800" #(0xffff << 11) self.panId = ModuleHelper.Default_PanId self.xpanId = ModuleHelper.Default_XpanId self.localprefix = ModuleHelper.Default_MLPrefix self.pskc = "00000000000000000000000000000000" # OT only accept hex format PSKc for now self.securityPolicySecs = ModuleHelper.Default_SecurityPolicy self.securityPolicyFlags = "onrcb" self.activetimestamp = ModuleHelper.Default_ActiveTimestamp #self.sedPollingRate = ModuleHelper.Default_Harness_SED_Polling_Rate self.sedPollingRate = 3 self.deviceRole = None self.provisioningUrl = '' self.hasActiveDatasetToCommit = False self.logThread = Queue() self.logThreadStatus = self.logStatus['stop'] self.joinCommissionedStatus = self.joinStatus['notstart'] self.networkDataRequirement = '' # indicate Thread device requests full or stable network data self.isPowerDown = False # indicate if Thread device experiences a power down event self._addressfilterMode = 'disable' # indicate AddressFilter mode ['disable', 'whitelist', 'blacklist'] self._addressfilterSet = set() # cache filter entries self.isActiveCommissioner = False # indicate if Thread device is an active commissioner self._lines = None # buffered lines read from device # initialize device configuration try: self.setMAC(self.mac) self.__setChannelMask(self.channelMask) self.__setSecurityPolicy(self.securityPolicySecs, self.securityPolicyFlags) self.setChannel(self.channel) self.setPANID(self.panId) self.setXpanId(self.xpanId) self.setNetworkName(self.networkName) self.setNetworkKey(self.networkKey) self.setMLPrefix(self.localprefix) self.setPSKc(self.pskc) self.setActiveTimestamp(self.activetimestamp) except Exception, e: ModuleHelper.WriteIntoDebugLogger("setDefaultValue() Error: " + str(e)) def getDeviceConncetionStatus(self): """check if serial port connection is ready or not""" print '%s call getDeviceConnectionStatus' % self.port return self.deviceConnected def getPollingRate(self): """get data polling rate for sleepy end device""" print '%s call getPollingRate' % self.port return self.__sendCommand('pollperiod')[0] def setPollingRate(self, iPollingRate): """set data polling rate for sleepy end device Args: iPollingRate: data poll period of sleepy end device Returns: True: successful to set the data polling rate for sleepy end device False: fail to set the data polling rate for sleepy end device """ print '%s call setPollingRate' % self.port print iPollingRate try: cmd = 'pollperiod %s' % str(iPollingRate) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setPollingRate() Error: " + str(e)) def setLinkQuality(self, EUIadr, LinkQuality): """set custom LinkQualityIn for all receiving messages from the specified EUIadr Args: EUIadr: a given extended address LinkQuality: a given custom link quality link quality/link margin mapping table 3: 21 - 255 (dB) 2: 11 - 20 (dB) 1: 3 - 9 (dB) 0: 0 - 2 (dB) Returns: True: successful to set the link quality False: fail to set the link quality """ print '%s call setLinkQuality' % self.port print EUIadr print LinkQuality try: # process EUIadr euiHex = hex(EUIadr) euiStr = str(euiHex) euiStr = euiStr.rstrip('L') address64 = '' if '0x' in euiStr: address64 = euiStr.lstrip('0x') # prepend 0 at the beginning if len(address64) < 16: address64 = address64.zfill(16) print address64 cmd = 'macfilter rss add-lqi %s %s' % (address64, str(LinkQuality)) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setLinkQuality() Error: " + str(e)) def setOutBoundLinkQuality(self, LinkQuality): """set custom LinkQualityIn for all receiving messages from the any address Args: LinkQuality: a given custom link quality link quality/link margin mapping table 3: 21 - 255 (dB) 2: 11 - 20 (dB) 1: 3 - 9 (dB) 0: 0 - 2 (dB) Returns: True: successful to set the link quality False: fail to set the link quality """ print '%s call setOutBoundLinkQuality' % self.port print LinkQuality try: cmd = 'macfilter rss add-lqi * %s' % str(LinkQuality) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setOutBoundLinkQuality() Error: " + str(e)) def removeRouterPrefix(self, prefixEntry): """remove the configured prefix on a border router Args: prefixEntry: a on-mesh prefix entry Returns: True: successful to remove the prefix entry from border router False: fail to remove the prefix entry from border router """ print '%s call removeRouterPrefix' % self.port print prefixEntry prefix = self.__convertIp6PrefixStringToIp6Address(str(prefixEntry)) try: prefixLen = 64 cmd = 'prefix remove %s/%d' % (prefix, prefixLen) print cmd if self.__sendCommand(cmd)[0] == 'Done': # send server data ntf to leader return self.__sendCommand('netdataregister')[0] == 'Done' else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("removeRouterPrefix() Error: " + str(e)) def resetAndRejoin(self, timeout): """reset and join back Thread Network with a given timeout delay Args: timeout: a timeout interval before rejoin Thread Network Returns: True: successful to reset and rejoin Thread Network False: fail to reset and rejoin the Thread Network """ print '%s call resetAndRejoin' % self.port print timeout try: self._sendline('reset') self.isPowerDown = True time.sleep(timeout) if self.deviceRole == Thread_Device_Role.SED: self.setPollingRate(self.sedPollingRate) self.__startOpenThread() time.sleep(3) if self.__sendCommand('state')[0] == 'disabled': print '[FAIL] reset and rejoin' return False return True except Exception, e: ModuleHelper.WriteIntoDebugLogger("resetAndRejoin() Error: " + str(e)) def configBorderRouter(self, P_Prefix, P_stable=1, P_default=1, P_slaac_preferred=0, P_Dhcp=0, P_preference=0, P_on_mesh=1, P_nd_dns=0): """configure the border router with a given prefix entry parameters Args: P_Prefix: IPv6 prefix that is available on the Thread Network P_stable: is true if the default router is expected to be stable network data P_default: is true if border router offers the default route for P_Prefix P_slaac_preferred: is true if Thread device is allowed auto-configure address using P_Prefix P_Dhcp: is true if border router is a DHCPv6 Agent P_preference: is two-bit signed integer indicating router preference P_on_mesh: is true if P_Prefix is considered to be on-mesh P_nd_dns: is true if border router is able to supply DNS information obtained via ND Returns: True: successful to configure the border router with a given prefix entry False: fail to configure the border router with a given prefix entry """ print '%s call configBorderRouter' % self.port prefix = self.__convertIp6PrefixStringToIp6Address(str(P_Prefix)) print prefix try: parameter = '' prf = '' if P_slaac_preferred == 1: parameter += 'p' parameter += 'a' if P_stable == 1: parameter += 's' if P_default == 1: parameter += 'r' if P_Dhcp == 1: parameter += 'd' if P_on_mesh == 1: parameter += 'o' if P_preference == 1: prf = 'high' elif P_preference == 0: prf = 'med' elif P_preference == -1: prf = 'low' else: pass cmd = 'prefix add %s/64 %s %s' % (prefix, parameter, prf) print cmd if self.__sendCommand(cmd)[0] == 'Done': # if prefix configured before starting OpenThread stack # do not send out server data ntf pro-actively if not self.__isOpenThreadRunning(): return True else: # send server data ntf to leader return self.__sendCommand('netdataregister')[0] == 'Done' else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("configBorderRouter() Error: " + str(e)) def setNetworkIDTimeout(self, iNwkIDTimeOut): """set networkid timeout for Thread device Args: iNwkIDTimeOut: a given NETWORK_ID_TIMEOUT Returns: True: successful to set NETWORK_ID_TIMEOUT False: fail to set NETWORK_ID_TIMEOUT """ print '%s call setNetworkIDTimeout' % self.port print iNwkIDTimeOut iNwkIDTimeOut /= 1000 try: cmd = 'networkidtimeout %s' % str(iNwkIDTimeOut) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setNetworkIDTimeout() Error: " + str(e)) def setKeepAliveTimeOut(self, iTimeOut): """set keep alive timeout for device has been deprecated and also set SED polling rate Args: iTimeOut: data poll period for sleepy end device Returns: True: successful to set the data poll period for SED False: fail to set the data poll period for SED """ print '%s call setKeepAliveTimeOut' % self.port print iTimeOut try: cmd = 'pollperiod %s' % str(iTimeOut) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setKeepAliveTimeOut() Error: " + str(e)) def setKeySequenceCounter(self, iKeySequenceValue): """ set the Key sequence counter corresponding to Thread Network master key Args: iKeySequenceValue: key sequence value Returns: True: successful to set the key sequence False: fail to set the key sequence """ print '%s call setKeySequenceCounter' % self.port print iKeySequenceValue try: cmd = 'keysequence counter %s' % str(iKeySequenceValue) if self.__sendCommand(cmd)[0] == 'Done': time.sleep(1) return True else: return False except Exception, e: ModuleHelper.WriteIntoDebugLogger("setKeySequenceCounter() Error; " + str(e)) def getKeySequenceCounter(self): """get current Thread Network key sequence""" print '%s call getKeySequenceCounter' % self.port keySequence = '' keySequence = self.__sendCommand('keysequence counter')[0] return keySequence def incrementKeySequenceCounter(self, iIncrementValue=1): """increment the key sequence with a given value Args: iIncrementValue: specific increment value to be added Returns: True: successful to increment the key sequence with a given value False: fail to increment the key sequence with a given value """ print '%s call incrementKeySequenceCounter' % self.port print iIncrementValue currentKeySeq = '' try: currentKeySeq = self.getKeySequenceCounter() keySequence = int(currentKeySeq, 10) + iIncrementValue print keySequence return self.setKeySequenceCounter(keySequence) except Exception, e: ModuleHelper.WriteIntoDebugLogger("incrementKeySequenceCounter() Error: " + str(e)) def setNetworkDataRequirement(self, eDataRequirement): """set whether the Thread device requires the full network data or only requires the stable network data Args: eDataRequirement: is true if requiring the full network data Returns: True: successful to set the network requirement """ print '%s call setNetworkDataRequirement' % self.port print eDataRequirement if eDataRequirement == Device_Data_Requirement.ALL_DATA: self.networkDataRequirement = 'n' return True def configExternalRouter(self, P_Prefix, P_stable, R_Preference=0): """configure border router with a given external route prefix entry Args: P_Prefix: IPv6 prefix for the route P_Stable: is true if the external route prefix is stable network data R_Preference: a two-bit signed integer indicating Router preference 1: high 0: medium -1: low Returns: True: successful to configure the border router with a given external route prefix False: fail to configure the border router with a given external route prefix """ print '%s call configExternalRouter' % self.port print P_Prefix stable = '' prefix = self.__convertIp6PrefixStringToIp6Address(str(P_Prefix)) try: if R_Preference == 1: prf = 'high' elif R_Preference == 0: prf = 'med' elif R_Preference == -1: prf = 'low' else: pass if P_stable: stable += 's' cmd = 'route add %s/64 %s %s' % (prefix, stable, prf) else: cmd = 'route add %s/64 %s' % (prefix, prf) print cmd if self.__sendCommand(cmd)[0] == 'Done': # send server data ntf to leader return self.__sendCommand('netdataregister')[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("configExternalRouter() Error: " + str(e)) def getNeighbouringRouters(self): """get neighboring routers information Returns: neighboring routers' extended address """ print '%s call getNeighbouringRouters' % self.port try: routerInfo = [] routerList = [] routerList = self.__sendCommand('router list')[0].split() print routerList if 'Done' in routerList: print 'no neighbouring routers' return None for index in routerList: router = [] cmd = 'router %s' % index router = self.__sendCommand(cmd) for line in router: if 'Done' in line: break #elif 'Rloc' in line: # rloc16 = line.split()[1] elif 'Ext Addr' in line: eui = line.split()[2] routerInfo.append(int(eui, 16)) #elif 'LQI In' in line: # lqi_in = line.split()[1] #elif 'LQI Out' in line: # lqi_out = line.split()[1] else: pass print routerInfo return routerInfo except Exception, e: ModuleHelper.WriteIntoDebugLogger("getNeighbouringDevice() Error: " + str(e)) def getChildrenInfo(self): """get all children information Returns: children's extended address """ print '%s call getChildrenInfo' % self.port try: childrenInfoAll = [] childrenInfo = {'EUI': 0, 'Rloc16': 0, 'MLEID': ''} childrenList = self.__sendCommand('child list')[0].split() print childrenList if 'Done' in childrenList: print 'no children' return None for index in childrenList: cmd = 'child %s' % index child = [] child = self.__sendCommand(cmd) for line in child: if 'Done' in line: break elif 'Rloc' in line: rloc16 = line.split()[1] elif 'Ext Addr' in line: eui = line.split()[2] #elif 'Child ID' in line: # child_id = line.split()[2] #elif 'Mode' in line: # mode = line.split()[1] else: pass childrenInfo['EUI'] = int(eui, 16) childrenInfo['Rloc16'] = int(rloc16, 16) #children_info['MLEID'] = self.getMLEID() childrenInfoAll.append(childrenInfo['EUI']) #childrenInfoAll.append(childrenInfo) print childrenInfoAll return childrenInfoAll except Exception, e: ModuleHelper.WriteIntoDebugLogger("getChildrenInfo() Error: " + str(e)) def setXpanId(self, xPanId): """set extended PAN ID of Thread Network Args: xPanId: extended PAN ID in hex format Returns: True: successful to set the extended PAN ID False: fail to set the extended PAN ID """ xpanid = '' print '%s call setXpanId' % self.port 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 else: xpanid = xPanId cmd = 'extpanid %s' % xpanid datasetCmd = 'dataset extpanid %s' % xpanid self.xpanId = xpanid self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' and self.__sendCommand(datasetCmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setXpanId() Error: " + str(e)) def getNeighbouringDevices(self): """gets the neighboring devices' extended address to compute the DUT extended address automatically Returns: A list including extended address of neighboring routers, parent as well as children """ print '%s call getNeighbouringDevices' % self.port neighbourList = [] # get parent info parentAddr = self.getParentAddress() if parentAddr != 0: neighbourList.append(parentAddr) # get ED/SED children info childNeighbours = self.getChildrenInfo() if childNeighbours != None and len(childNeighbours) > 0: for entry in childNeighbours: neighbourList.append(entry) # get neighboring routers info routerNeighbours = self.getNeighbouringRouters() if routerNeighbours != None and len(routerNeighbours) > 0: for entry in routerNeighbours: neighbourList.append(entry) print neighbourList return neighbourList def setPartationId(self, partationId): """set Thread Network Partition ID Args: partitionId: partition id to be set by leader Returns: True: successful to set the Partition ID False: fail to set the Partition ID """ print '%s call setPartationId' % self.port print partationId cmd = 'leaderpartitionid %s' %(str(hex(partationId)).rstrip('L')) print cmd return self.__sendCommand(cmd)[0] == 'Done' def getGUA(self, filterByPrefix=None): """get expected global unicast IPv6 address of Thread device Args: filterByPrefix: a given expected global IPv6 prefix to be matched Returns: a global IPv6 address that matches with filterByPrefix or None if no matched GUA """ print '%s call getGUA' % self.port print filterByPrefix globalAddrs = [] try: # get global addrs set if multiple globalAddrs = self.getGlobal() if filterByPrefix is None: return globalAddrs[0] else: for line in globalAddrs: fullIp = ModuleHelper.GetFullIpv6Address(line) if fullIp.startswith(filterByPrefix): return fullIp print 'no global address matched' return None except Exception, e: ModuleHelper.WriteIntoDebugLogger("getGUA() Error: " + str(e)) def getShortAddress(self): """get Rloc16 short address of Thread device""" print '%s call getShortAddress' % self.port return self.getRloc16() def getULA64(self): """get mesh local EID of Thread device""" print '%s call getULA64' % self.port return self.getMLEID() def setMLPrefix(self, sMeshLocalPrefix): """set mesh local prefix""" print '%s call setMLPrefix' % self.port try: cmd = 'dataset meshlocalprefix %s' % sMeshLocalPrefix self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setMLPrefix() Error: " + str(e)) def getML16(self): """get mesh local 16 unicast address (Rloc)""" print '%s call getML16' % self.port return self.getRloc() def downgradeToDevice(self): pass def upgradeToRouter(self): pass def forceSetSlaac(self, slaacAddress): """force to set a slaac IPv6 address to Thread interface Args: slaacAddress: a slaac IPv6 address to be set Returns: True: successful to set slaac address to Thread interface False: fail to set slaac address to Thread interface """ print '%s call forceSetSlaac' % self.port print slaacAddress try: cmd = 'ipaddr add %s' % str(slaacAddress) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("forceSetSlaac() Error: " + str(e)) def setSleepyNodePollTime(self): pass def enableAutoDUTObjectFlag(self): """set AutoDUTenable flag""" print '%s call enableAutoDUTObjectFlag' % self.port self.AutoDUTEnable = True def getChildTimeoutValue(self): """get child timeout""" print '%s call getChildTimeoutValue' % self.port childTimeout = self.__sendCommand('childtimeout')[0] return int(childTimeout) def diagnosticGet(self, strDestinationAddr, listTLV_ids=[]): if not listTLV_ids: return if not len(listTLV_ids): return cmd = 'networkdiagnostic get %s %s' % (strDestinationAddr, ' '.join([str(tlv) for tlv in listTLV_ids])) print cmd return self._sendline(cmd) def diagnosticReset(self, strDestinationAddr, listTLV_ids=[]): if not listTLV_ids: return if not len(listTLV_ids): return cmd = 'networkdiagnostic reset %s %s' % (strDestinationAddr, ' '.join([str(tlv) for tlv in listTLV_ids])) print cmd return self.__sendCommand(cmd) def diagnosticQuery(self,strDestinationAddr, listTLV_ids = []): self.diagnosticGet(strDestinationAddr, listTLV_ids) def startNativeCommissioner(self, strPSKc='GRLpassWord'): #TODO: Support the whole Native Commissioner functionality # Currently it only aims to trigger a Discovery Request message to pass Certification test 5.8.4 print '%s call startNativeCommissioner' % self.port self.__sendCommand('ifconfig up') cmd = 'joiner start %s' %(strPSKc) print cmd if self.__sendCommand(cmd)[0] == "Done": return True else: return False def startCollapsedCommissioner(self): """start Collapsed Commissioner Returns: True: successful to start Commissioner False: fail to start Commissioner """ print '%s call startCollapsedCommissioner' % self.port if self.__startOpenThread(): time.sleep(20) cmd = 'commissioner start' print cmd if self.__sendCommand(cmd)[0] == 'Done': self.isActiveCommissioner = True time.sleep(20) # time for petition process return True return False def setJoinKey(self, strPSKc): pass def scanJoiner(self, xEUI='*', strPSKd='threadjpaketest'): """scan Joiner Args: xEUI: Joiner's EUI-64 strPSKd: Joiner's PSKd for commissioning Returns: True: successful to add Joiner's steering data False: fail to add Joiner's steering data """ print '%s call scanJoiner' % self.port # long timeout value to avoid automatic joiner removal (in seconds) 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 else: eui64 = xEUI cmd = 'commissioner joiner add %s %s %s' % (eui64, strPSKd, str(timeout)) print cmd if self.__sendCommand(cmd)[0] == 'Done': if self.logThreadStatus == self.logStatus['stop']: self.logThread = ThreadRunner.run(target=self.__readCommissioningLogs, args=(120,)) return True else: return False def setProvisioningUrl(self, strURL='grl.com'): """set provisioning Url Args: strURL: Provisioning Url string Returns: True: successful to set provisioning Url False: fail to set provisioning Url """ print '%s call setProvisioningUrl' % self.port cmd = 'commissioner provisioningurl %s' %(strURL) self.provisioningUrl = strURL print cmd return self.__sendCommand(cmd)[0] == "Done" def allowCommission(self): """start commissioner candidate petition process Returns: True: successful to start commissioner candidate petition process False: fail to start commissioner candidate petition process """ print '%s call allowCommission' % self.port try: cmd = 'commissioner start' print cmd if self.__sendCommand(cmd)[0] == 'Done': self.isActiveCommissioner = True time.sleep(40) # time for petition process and at least one keep alive return True else: return False except Exception, e: ModuleHelper.writeintodebuglogger("allowcommission() error: " + str(e)) def joinCommissioned(self, strPSKd='threadjpaketest', waitTime=20): """start joiner Args: strPSKd: Joiner's PSKd Returns: True: successful to start joiner False: fail to start joiner """ print '%s call joinCommissioned' % self.port self.__sendCommand('ifconfig up') cmd = 'joiner start %s %s' %(strPSKd, self.provisioningUrl) print cmd if self.__sendCommand(cmd)[0] == "Done": maxDuration = 150 # seconds self.joinCommissionedStatus = self.joinStatus['ongoing'] if self.logThreadStatus == self.logStatus['stop']: self.logThread = ThreadRunner.run(target=self.__readCommissioningLogs, args=(maxDuration,)) t_end = time.time() + maxDuration while time.time() < t_end: if self.joinCommissionedStatus == self.joinStatus['succeed']: break elif self.joinCommissionedStatus == self.joinStatus['failed']: return False time.sleep(1) self.__sendCommand('thread start') time.sleep(30) return True else: return False def getCommissioningLogs(self): """get Commissioning logs Returns: Commissioning logs """ rawLogs = self.logThread.get() ProcessedLogs = [] payload = [] while not rawLogs.empty(): rawLogEach = rawLogs.get() print rawLogEach if "[THCI]" not in rawLogEach: continue EncryptedPacket = PlatformDiagnosticPacket() infoList = rawLogEach.split('[THCI]')[1].split(']')[0].split('|') for eachInfo in infoList: print eachInfo info = eachInfo.split("=") infoType = info[0].strip() infoValue = info[1].strip() if "direction" in infoType: EncryptedPacket.Direction = PlatformDiagnosticPacket_Direction.IN if 'recv' in infoValue \ else PlatformDiagnosticPacket_Direction.OUT if 'send' in infoValue \ else PlatformDiagnosticPacket_Direction.UNKNOWN elif "type" in infoType: EncryptedPacket.Type = PlatformDiagnosticPacket_Type.JOIN_FIN_req if 'JOIN_FIN.req' in infoValue \ else PlatformDiagnosticPacket_Type.JOIN_FIN_rsp if 'JOIN_FIN.rsp' in infoValue \ else PlatformDiagnosticPacket_Type.JOIN_ENT_req if 'JOIN_ENT.ntf' in infoValue \ else PlatformDiagnosticPacket_Type.JOIN_ENT_rsp if 'JOIN_ENT.rsp' in infoValue \ else PlatformDiagnosticPacket_Type.UNKNOWN elif "len" in infoType: bytesInEachLine = 16 EncryptedPacket.TLVsLength = int(infoValue) payloadLineCount = (int(infoValue) + bytesInEachLine - 1)/bytesInEachLine while payloadLineCount > 0: payloadLineCount = payloadLineCount - 1 payloadLine = rawLogs.get() payloadSplit = payloadLine.split('|') for block in range(1, 3): payloadBlock = payloadSplit[block] payloadValues = payloadBlock.split(' ') for num in range(1, 9): if ".." not in payloadValues[num]: payload.append(int(payloadValues[num], 16)) EncryptedPacket.TLVs = PlatformPackets.read(EncryptedPacket.Type, payload) if payload != [] else [] ProcessedLogs.append(EncryptedPacket) return ProcessedLogs def MGMT_ED_SCAN(self, sAddr, xCommissionerSessionId, listChannelMask, xCount, xPeriod, xScanDuration): """send MGMT_ED_SCAN message to a given destinaition. Args: sAddr: IPv6 destination address for this message xCommissionerSessionId: commissioner session id listChannelMask: a channel array to indicate which channels to be scaned xCount: number of IEEE 802.15.4 ED Scans (milliseconds) xPeriod: Period between successive IEEE802.15.4 ED Scans (milliseconds) xScanDuration: IEEE 802.15.4 ScanDuration to use when performing an IEEE 802.15.4 ED Scan (milliseconds) Returns: True: successful to send MGMT_ED_SCAN message. False: fail to send MGMT_ED_SCAN message """ print '%s call MGMT_ED_SCAN' % self.port channelMask = '' channelMask = '0x' + self.__convertLongToString(self.__convertChannelMask(listChannelMask)) try: cmd = 'commissioner energy %s %s %s %s %s' % (channelMask, xCount, xPeriod, xScanDuration, sAddr) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.writeintodebuglogger("MGMT_ED_SCAN() error: " + str(e)) def MGMT_PANID_QUERY(self, sAddr, xCommissionerSessionId, listChannelMask, xPanId): """send MGMT_PANID_QUERY message to a given destination Args: xPanId: a given PAN ID to check the conflicts Returns: True: successful to send MGMT_PANID_QUERY message. False: fail to send MGMT_PANID_QUERY message. """ print '%s call MGMT_PANID_QUERY' % self.port panid = '' channelMask = '' channelMask = '0x' + self.__convertLongToString(self.__convertChannelMask(listChannelMask)) if not isinstance(xPanId, str): panid = str(hex(xPanId)) try: cmd = 'commissioner panid %s %s %s' % (panid, channelMask, sAddr) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.writeintodebuglogger("MGMT_PANID_QUERY() error: " + str(e)) def MGMT_ANNOUNCE_BEGIN(self, sAddr, xCommissionerSessionId, listChannelMask, xCount, xPeriod): """send MGMT_ANNOUNCE_BEGIN message to a given destination Returns: True: successful to send MGMT_ANNOUNCE_BEGIN message. False: fail to send MGMT_ANNOUNCE_BEGIN message. """ print '%s call MGMT_ANNOUNCE_BEGIN' % self.port channelMask = '' channelMask = '0x' + self.__convertLongToString(self.__convertChannelMask(listChannelMask)) try: cmd = 'commissioner announce %s %s %s %s' % (channelMask, xCount, xPeriod, sAddr) print cmd return self.__sendCommand(cmd) == 'Done' except Exception, e: ModuleHelper.writeintodebuglogger("MGMT_ANNOUNCE_BEGIN() error: " + str(e)) def MGMT_ACTIVE_GET(self, Addr='', TLVs=[]): """send MGMT_ACTIVE_GET command Returns: True: successful to send MGMT_ACTIVE_GET False: fail to send MGMT_ACTIVE_GET """ print '%s call MGMT_ACTIVE_GET' % self.port try: cmd = 'dataset mgmtgetcommand active' if Addr != '': cmd += ' address ' cmd += Addr if len(TLVs) != 0: tlvs = "".join(hex(tlv).lstrip("0x").zfill(2) for tlv in TLVs) cmd += ' binary ' cmd += tlvs print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_ACTIVE_GET() Error: " + str(e)) def MGMT_ACTIVE_SET(self, sAddr='', xCommissioningSessionId=None, listActiveTimestamp=None, listChannelMask=None, xExtendedPanId=None, sNetworkName=None, sPSKc=None, listSecurityPolicy=None, xChannel=None, sMeshLocalPrefix=None, xMasterKey=None, xPanId=None, xTmfPort=None, xSteeringData=None, xBorderRouterLocator=None, BogusTLV=None, xDelayTimer=None): """send MGMT_ACTIVE_SET command Returns: True: successful to send MGMT_ACTIVE_SET False: fail to send MGMT_ACTIVE_SET """ print '%s call MGMT_ACTIVE_SET' % self.port try: cmd = 'dataset mgmtsetcommand active' if listActiveTimestamp != None: cmd += ' activetimestamp ' cmd += str(listActiveTimestamp[0]) if xExtendedPanId != None: cmd += ' extpanid ' xpanid = self.__convertLongToString(xExtendedPanId) if len(xpanid) < 16: xpanid = xpanid.zfill(16) cmd += xpanid if sNetworkName != None: cmd += ' networkname ' cmd += str(sNetworkName) if xChannel != None: cmd += ' channel ' cmd += str(xChannel) if sMeshLocalPrefix != None: cmd += ' localprefix ' cmd += str(sMeshLocalPrefix) if xMasterKey != None: cmd += ' masterkey ' key = self.__convertLongToString(xMasterKey) if len(key) < 32: key = key.zfill(32) cmd += key if xPanId != None: cmd += ' panid ' cmd += str(xPanId) if listChannelMask != None: cmd += ' channelmask ' cmd += '0x' + self.__convertLongToString(self.__convertChannelMask(listChannelMask)) if sPSKc != None or listSecurityPolicy != None or \ xCommissioningSessionId != None or xTmfPort != None or xSteeringData != None or xBorderRouterLocator != None or \ BogusTLV != None: cmd += ' binary ' if sPSKc != None: cmd += '0410' stretchedPskc = Thread_PBKDF2.get(sPSKc,ModuleHelper.Default_XpanId,ModuleHelper.Default_NwkName) pskc = hex(stretchedPskc).rstrip('L').lstrip('0x') if len(pskc) < 32: pskc = pskc.zfill(32) cmd += pskc if listSecurityPolicy != None: cmd += '0c03' rotationTime = 0 policyBits = 0 # previous passing way listSecurityPolicy=[True, True, 3600, False, False, True] if (len(listSecurityPolicy) == 6): rotationTime = listSecurityPolicy[2] # the last three reserved bits must be 1 policyBits = 0b00000111 if listSecurityPolicy[0]: policyBits = policyBits | 0b10000000 if listSecurityPolicy[1]: policyBits = policyBits | 0b01000000 if listSecurityPolicy[3]: policyBits = policyBits | 0b00100000 if listSecurityPolicy[4]: policyBits = policyBits | 0b00010000 if listSecurityPolicy[5]: policyBits = policyBits | 0b00001000 else: # new passing way listSecurityPolicy=[3600, 0b11001111] rotationTime = listSecurityPolicy[0] policyBits = listSecurityPolicy[1] policy = str(hex(rotationTime))[2:] if len(policy) < 4: policy = policy.zfill(4) cmd += policy cmd += str(hex(policyBits))[2:] if xCommissioningSessionId != None: cmd += '0b02' sessionid = str(hex(xCommissioningSessionId))[2:] if len(sessionid) < 4: sessionid = sessionid.zfill(4) cmd += sessionid if xBorderRouterLocator != None: cmd += '0902' locator = str(hex(xBorderRouterLocator))[2:] if len(locator) < 4: locator = locator.zfill(4) cmd += locator if xSteeringData != None: steeringData = self.__convertLongToString(xSteeringData) cmd += '08' + str(len(steeringData)/2).zfill(2) cmd += steeringData if BogusTLV != None: cmd += "8202aa55" print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_ACTIVE_SET() Error: " + str(e)) def MGMT_PENDING_GET(self, Addr='', TLVs=[]): """send MGMT_PENDING_GET command Returns: True: successful to send MGMT_PENDING_GET False: fail to send MGMT_PENDING_GET """ print '%s call MGMT_PENDING_GET' % self.port try: cmd = 'dataset mgmtgetcommand pending' if Addr != '': cmd += ' address ' cmd += Addr if len(TLVs) != 0: tlvs = "".join(hex(tlv).lstrip("0x").zfill(2) for tlv in TLVs) cmd += ' binary ' cmd += tlvs print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_PENDING_GET() Error: " + str(e)) def MGMT_PENDING_SET(self, sAddr='', xCommissionerSessionId=None, listPendingTimestamp=None, listActiveTimestamp=None, xDelayTimer=None, xChannel=None, xPanId=None, xMasterKey=None, sMeshLocalPrefix=None, sNetworkName=None): """send MGMT_PENDING_SET command Returns: True: successful to send MGMT_PENDING_SET False: fail to send MGMT_PENDING_SET """ print '%s call MGMT_PENDING_SET' % self.port try: cmd = 'dataset mgmtsetcommand pending' if listPendingTimestamp != None: cmd += ' pendingtimestamp ' cmd += str(listPendingTimestamp[0]) if listActiveTimestamp != None: cmd += ' activetimestamp ' cmd += str(listActiveTimestamp[0]) if xDelayTimer != None: cmd += ' delaytimer ' cmd += str(xDelayTimer) #cmd += ' delaytimer 3000000' if xChannel != None: cmd += ' channel ' cmd += str(xChannel) if xPanId != None: cmd += ' panid ' cmd += str(xPanId) if xMasterKey != None: cmd += ' masterkey ' key = self.__convertLongToString(xMasterKey) if len(key) < 32: key = key.zfill(32) cmd += key if sMeshLocalPrefix != None: cmd += ' localprefix ' cmd += str(sMeshLocalPrefix) if sNetworkName != None: cmd += ' networkname ' cmd += str(sNetworkName) if xCommissionerSessionId != None: cmd += ' binary ' cmd += '0b02' sessionid = str(hex(xCommissionerSessionId))[2:] if len(sessionid) < 4: sessionid = sessionid.zfill(4) cmd += sessionid print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_PENDING_SET() Error: " + str(e)) def MGMT_COMM_GET(self, Addr='ff02::1', TLVs=[]): """send MGMT_COMM_GET command Returns: True: successful to send MGMT_COMM_GET False: fail to send MGMT_COMM_GET """ print '%s call MGMT_COMM_GET' % self.port try: cmd = 'commissioner mgmtget' if len(TLVs) != 0: tlvs = "".join(hex(tlv).lstrip("0x").zfill(2) for tlv in TLVs) cmd += ' binary ' cmd += tlvs print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_COMM_GET() Error: " + str(e)) def MGMT_COMM_SET(self, Addr='ff02::1', xCommissionerSessionID=None, xSteeringData=None, xBorderRouterLocator=None, xChannelTlv=None, ExceedMaxPayload=False): """send MGMT_COMM_SET command Returns: True: successful to send MGMT_COMM_SET False: fail to send MGMT_COMM_SET """ print '%s call MGMT_COMM_SET' % self.port try: cmd = 'commissioner mgmtset' if xCommissionerSessionID != None: # use assigned session id cmd += ' sessionid ' cmd += str(xCommissionerSessionID) elif xCommissionerSessionID is None: # use original session id if self.isActiveCommissioner is True: cmd += ' sessionid ' cmd += self.__getCommissionerSessionId() else: pass if xSteeringData != None: cmd += ' steeringdata ' cmd += str(hex(xSteeringData)[2:]) if xBorderRouterLocator != None: cmd += ' locator ' cmd += str(hex(xBorderRouterLocator)) if xChannelTlv != None: cmd += ' binary ' cmd += '000300' + hex(xChannelTlv).lstrip('0x').zfill(4) print cmd return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("MGMT_COMM_SET() Error: " + str(e)) def setActiveDataset(self, listActiveDataset=[]): print '%s call setActiveDataset' % self.port def setCommisionerMode(self): print '%s call setCommissionerMode' % self.port def setPSKc(self, strPSKc): print '%s call setPSKc' % self.port try: cmd = 'dataset pskc %s' % strPSKc self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setPSKc() Error: " + str(e)) def setActiveTimestamp(self, xActiveTimestamp): print '%s call setActiveTimestamp' % self.port try: self.activetimestamp = xActiveTimestamp cmd = 'dataset activetimestamp %s' % str(xActiveTimestamp) self.hasActiveDatasetToCommit = True return self.__sendCommand(cmd)[0] == 'Done' except Exception, e: ModuleHelper.WriteIntoDebugLogger("setActiveTimestamp() Error: " + str(e)) def setUdpJoinerPort(self, portNumber): """set Joiner UDP Port Args: portNumber: Joiner UDP Port number Returns: True: successful to set Joiner UDP Port False: fail to set Joiner UDP Port """ print '%s call setUdpJoinerPort' % self.port cmd = 'joinerport %d' % portNumber print cmd return self.__sendCommand(cmd)[0] == 'Done' def commissionerUnregister(self): """stop commissioner Returns: True: successful to stop commissioner False: fail to stop commissioner """ print '%s call commissionerUnregister' % self.port cmd = 'commissioner stop' print cmd return self.__sendCommand(cmd)[0] == 'Done' def sendBeacons(self, sAddr, xCommissionerSessionId, listChannelMask, xPanId): print '%s call sendBeacons' % self.port self._sendline('scan') return True def updateRouterStatus(self): """force update to router as if there is child id request""" print '%s call updateRouterStatus' % self.port cmd = 'state' while True: state = self.__sendCommand(cmd)[0] if state == 'detached': continue elif state == 'child': break else: return False cmd = 'state router' return self.__sendCommand(cmd)[0] == 'Done' def setRouterThresholdValues(self, upgradeThreshold, downgradeThreshold): print '%s call setRouterThresholdValues' % self.port self.__setRouterUpgradeThreshold(upgradeThreshold) self.__setRouterDowngradeThreshold(downgradeThreshold) def setMinDelayTimer(self, iSeconds): print '%s call setMinDelayTimer' % self.port cmd = 'delaytimermin %s' % iSeconds print cmd return self.__sendCommand(cmd)[0] == 'Done' def ValidateDeviceFirmware(self): print '%s call ValidateDeviceFirmware' % self.port if "OPENTHREAD" in self.UIStatusMsg: return True else: return False