检查代码中的错误 class ADB_SHELL: def get_ip(self): # 执行adb shell命令并输出结果 subprocess.check_output( ['adb', 'shell', 'udhcpc'] ) subprocess.check_output( ['adb', 'shell', 'udhcpc -i eth1'] ) self.conf = subprocess.check_output( ['adb', 'shell', 'ifconfig'] ).decode() # conf = str(ip).split(r'\r\r\n') # tmp = conf.replace( "\r\r\n", "\n" ) # print( tmp) self.ip = re.findall( r'addr:(.*?) Bcast', str( self.conf ) ) print(self.ip) for self.i in selfip : speed = subprocess.check_output((['adb', 'shell', f'iperf3 -B {self.i} -c 192.168.102.105'])).decode() print(speed.replace("\r\r\n", "\n")) for i in range(5): write_data = subprocess.check_output(['adb', 'shell', 'time dd if=/dev/zero of=/data/test.data bs=128k count=1024']).decode() print(write_data.replace("\r\r\n", "\n")) read_data = subprocess.check_output(['adb', 'shell', 'time dd if=/data/test.data of=/dev/null bs=128k count=1024']).decode() print(read_data.replace("\r\r\n", "\n")) ls = subprocess.check_output( ['adb', 'shell', 'ls /data'] ).decode() print( ls.replace( '\r', ' ' ) ) dl = subprocess.check_output( ['adb', 'shell', 'rm /data/test.data'] ).decode() l = subprocess.check_output(['adb', 'shell', 'ls /data']).decode() print( l.replace( '\r', ' ' ) ) subprocess.check_output( (['adb', 'shell', f'iperf3 -s']) ) def get_ssh(self): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy) for i in self.ip: ssh.connect(hostname='192.168.102.105',port=22,username='root',password='xiayi123456',timeout=30) stdin,stdout,stderr = ssh.exec_command(f'iperf3 -c {i}') print(stdout.read()) # ssh.close() if __name__ == '__main__': # get_ip() # get_ssh() A = ADB_SHELL t1 = threading.Thread(target=A.get_ssh) t2 = threading.Thread(target=A.get_ip) t1.start() t2.start() t1.join() t2.join()
时间: 2024-04-29 17:19:53 浏览: 258
在第12行代码中,将 `selfip` 改为 `self.ip`。此外,第28行代码中的 `ssh.set_missing_host_key_policy` 应该是 `ssh.set_missing_host_key_policy()`。最后,第39行代码中的 `A = ADB_SHELL` 应该是 `A = ADB_SHELL()`,因为 ADB_SHELL 是一个类,需要实例化才能调用其方法。
相关问题
检查这段代码 adb_path = 'D:\\tools\ADB\platform-tools\\' # 定义adb shell命令 adb_command = adb_path +'adb shell ls /data/' # 执行adb shell命令并输出结果 output = subprocess.check_output(adb_command.split()) print(output.decode('utf-8'))
这段代码有一个小问题,需要将adb命令与adb路径分开,以便正确执行。您可以将代码修改为以下形式:
```python
import subprocess
# 设置adb路径
adb_path = r'D:\tools\ADB\platform-tools\adb'
# 定义adb shell命令
adb_command = adb_path + ' shell ls /data/'
# 执行adb shell命令并输出结果
output = subprocess.check_output(adb_command.split())
print(output.decode('utf-8'))
```
这段代码将执行`adb shell ls /data/`命令并输出结果。请注意,我在`adb_path`变量前加上了`r`前缀,这是为了将特殊字符`\`视为普通字符,而不是转义字符。请将`D:\tools\ADB\platform-tools\adb`替换为您实际的adb路径。
使用多线程执行这两个函数 def get_ip(): # 执行adb shell命令并输出结果 subprocess.check_output( ['adb', 'shell', 'udhcpc'] ) subprocess.check_output( ['adb', 'shell', 'udhcpc -i eth1'] ) conf = subprocess.check_output( ['adb', 'shell', 'ifconfig'] ).decode() # conf = str(ip).split(r'\r\r\n') # tmp = conf.replace( "\r\r\n", "\n" ) # print( tmp) ip = re.findall( r'addr:(.*?) Bcast', str( conf ) ) print(ip) for i in ip : speed = subprocess.check_output((['adb', 'shell', f'iperf3 -B {i} -c 192.168.102.105'])).decode() print(speed.replace("\r\r\n", "\n")) def get_ssh(): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy) ssh.connect(hostname='192.168.102.105',port=22,username='root',password='xiayi123456',timeout=30) stdin,stdout,stderr = ssh.exec_command('iperf3 -s') print(stdout.read())
可以使用 Python 的 `threading` 模块来实现多线程执行这两个函数。具体代码如下:
```
import threading
import subprocess
import re
import paramiko
def get_ip():
subprocess.check_output(['adb', 'shell', 'udhcpc'])
subprocess.check_output(['adb', 'shell', 'udhcpc -i eth1'])
conf = subprocess.check_output(['adb', 'shell', 'ifconfig']).decode()
ip = re.findall(r'addr:(.*?) Bcast', str(conf))
print(ip)
for i in ip:
speed = subprocess.check_output((['adb', 'shell', f'iperf3 -B {i} -c 192.168.102.105'])).decode()
print(speed.replace("\r\r\n", "\n"))
def get_ssh():
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
ssh.connect(hostname='192.168.102.105', port=22, username='root', password='xiayi123456', timeout=30)
stdin, stdout, stderr = ssh.exec_command('iperf3 -s')
print(stdout.read())
if __name__ == '__main__':
# 创建两个线程
t1 = threading.Thread(target=get_ip)
t2 = threading.Thread(target=get_ssh)
# 启动线程
t1.start()
t2.start()
# 等待线程执行完毕
t1.join()
t2.join()
```
在主函数中,首先创建了两个线程 `t1` 和 `t2`,分别执行 `get_ip` 和 `get_ssh` 函数。然后通过调用 `start()` 方法启动线程,最后通过调用 `join()` 方法等待线程执行完毕。这样就可以实现多线程执行两个函数。
阅读全文