主要完成的功能流程包括
1.先检查是不是能和设备ping通
2.通过telnet协议与设备通信
3.通过tftp将用于升级的bin包和fupdate文件传到设备内部
4.修改权限并且固件上传,升级完成
5.python程序用pyinstaller打包成exe,方便使用
ps:为了提高代码的复用性将待升级设备的ip和对应的升级的固件写成json,搞一个函数自动获取本机的ip地址,除了telnet登录的用户名和密码是写死的,其他关键数据都变成global形式,
pps:关于等待程序运行的时间和telnet登出踩了坑,json文件读取的时候地址转换也踩坑,获取ip地址也踩坑,把telnetlib这个给库里的方法翻了个底朝天…
ppps:有本地包的情况下,可以直接用tftp传,可能会有不能联网计算机不能在url下载的情况
用之前要import socket,另外这个方法只能直接当个变量使用。如果为了简洁写了另外一个名字(global变量),就会诡异的找不到= =
import socket
# 获取本地网卡的ip地址
def get_ip_address():
res = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
res.connect(("8.8.8.8", 80))
return res.getsockname()[0]
2、等待方法
等待+及时读取执行时候输出的信息
def wait_read(client, nwait):
time.sleep(nwait)
client.readState();
3.整一个TelnetClient类
封装一些telnet连接方法、命令执行方法和读取状态的方法。
-
对于telnetlib中的read_very_eager().decode(‘ascii’)是要指定编码格式的
(read_very_eager(b’XXXX’)读取上次更新之后的最新的输出信息) -
read_until()中的参数也是要规定为byte类型可以在字符串前面加个b,换行符同理(b’n’)
(read_until(match,timeout)是知道读取到特定的字符才去进行对应的操作或等待超时) -
write(USER.encode(‘ascii’) + b’n’)
(write函数也类似,要规定编码方式,将byte类型的内容写到socket里面)
class TelnetClient:
def __init__(self):
self.session = telnetlib.Telnet()
# telnet连接
def telnet_connect(self, client_addr, port_num):
try:
self.session = telnetlib.Telnet(client_addr, port=port_num)
self.session.open(client_addr, port=port_num) # telnet默认开启本行注释掉
except:
logging.warning('%s网络连接失败' % client_addr)
return False
# 延时两秒再收取返回结果,给服务端足够响应时间
time.sleep(2)
# 获取登录结果
command_result = self.session.read_very_eager().decode('ascii')
print('telnet result:', command_result)
try:
#读到login时候输入用户名
self.session.read_until(b'Login:', timeout=2)
self.session.write(USER.encode('ascii') + b'n')
command_result = self.session.read_very_eager().decode('ascii')
print(command_result)
time.sleep(2)
#读到password的时候输入密码
self.session.read_until(b'Password:', timeout=2)
self.session.write(PSW.encode('ascii') + b'n')
command_result = self.session.read_very_eager().decode('ascii')
print(command_result)
time.sleep(2)
# 打印登陆后的输出信息,如果是熟悉的welcome linux就连接成功
command_result = self.session.read_very_eager().decode('ascii')
print(command_result)
print('telnet登陆成功')
except:
logging.warning('telnet 登陆失败')
return False
return True
# telnet 退出登录,慎用,可能会让session提前被停止
def logout_telnet(self):
self.session.close()
# 打印最新输出的信息
def readState(self):
command_result = self.session.read_very_eager().decode('ascii')
print(command_result)
# 执行需要返回值的指令
def execute_command(self, command, time_wait):
# 执行命令并且等待
self.session.write(command + b'n')
time.sleep(time_wait)
# 获取命令结果
command_result = self.session.read_very_eager().decode('ascii')
# logging.warning(command + 'execute result:n %s' % command_result+b'n')
print(command_result)
return command_result
4.开始telnet连接以及命令执行
在升级或传文件需要较长时间的时候一定要抓到信息,给出合理的等待时间,否则会出现任务没有完成就导致telenet登出闪退的尴尬错误
def start_telnet():
telnet_client = TelnetClient()
# 1、 telnet连接
if telnet_client.telnet_connect(updateIp, port):
time.sleep(3)
telnet_client.execute_command(b'cd /tmp/', 1)
# print("cd到tmp")
time.sleep(3)
# 2、tftp传递
# tftp_str = "tftp -gl fupdate " + get_ip_address()
bin_str = "tftp -gl " + updateBinName + " " + get_ip_address()
telnet_client.execute_command(bytes(bin_str, "utf-8"), 2)
# bin文件上传很慢,需要多等待一些时间
wait_read(telnet_client, 5)
wait_read(telnet_client, 15)
wait_read(telnet_client, 15)
wait_read(telnet_client, 15)
wait_read(telnet_client, 8)
wait_read(telnet_client, 8)
tftp_str = "tftp -gl fupdate " + get_ip_address()
telnet_client.execute_command(bytes(tftp_str, "utf-8"), 1)
wait_read(telnet_client, 3)
wait_read(telnet_client, 3)
# print('tftp执行')
# 3、修改权限 并升级固件
telnet_client.execute_command(b'chmod 777 fupdate', 2)
time.sleep(3)
# update_str = "fupdate -l " + get_ip_address() + "/"+updateBinName+" /tmp/app.bin"
update_str = "fupdate /tmp/" + updateBinName
telnet_client.execute_command(bytes(update_str, "utf-8"), 2)
# logging.warning('升级成功')
time.sleep(3)
wait_read(telnet_client, 5)
wait_read(telnet_client, 5)
wait_read(telnet_client, 5)
wait_read(telnet_client, 5)
input("程序执行完成,点击Enter继续升级");
go_work();
else:
telnet_client.logout_telnet()
#pass
5.两个用来执行的函数
整理执行流程,把乱七八糟的方法从main函数解放出来!直接执行这两个流程就可以!
#检查是否能与设备ping通,正式开始连接和升级
def start_work():
# 规定设备回应五次视为ping成功
# ping_result = os.system('ping -c<5> %s' % updateIp)
ping_result = pingHost(updateIp)
# 代码更新pingHost
if ping_result:
print("*********************PING %s SUCCESS*********************" % updateIp)
print("***********************START TELNET CONNECT***********************")
start_telnet()
else:
print("*********************PING %s FAILED*********************" % updateIp)
# 其实是按任意键可以再次尝试ping
input("*********************PRESS ENTER TO TRY AGAIN**********************")
start_work()
# 开始升级前的准备工作:拿到当前pc的ip
def go_work():
if get_ip_address() != "":
print("********************当前电脑IP:%s *********************" % get_ip_address())
print("******************************************************************")
start_work()
else:
print("*******************获取电脑IP失败,请检查网络连接*******************")
6.main函数
打开文件拿到写在json里面的ip和包名,开始work!
if __name__ == '__main__':
# 读取JSON文件,升级目前IP和升级包名
print("************************升级程序已启动****************************")
print("*********************BUILD DATE 2022-5-30************************")
print("*****************************************************************")
# 读取json文件地址
filepath = os.path.dirname(os.path.realpath(__file__)) + os.sep + "update.json"
# 和\的替换
filepath = filepath.replace("\", "\")
# 打开写好的json文件,读取需要的设备ip和bin包的名字
with open(os.getcwd() + os.sep + "update.json", 'r', encoding="utf-8") as load_f:
updateDic = json.load(load_f)
updateIp = updateDic["updateIP"]
updateBinName = updateDic["updateFileName"]
print("================== 自动升级程序已启动成功 ===================")
# 用ip和包名是否存在来判断是否读取成功
if len(updateIp) > 0 and len(updateBinName) > 0:
print("t升级配置文件读取成功")
print("t升级目标IP地址:", updateIp)
print("t升级目标型号:", updateBinName)
#都读取成功之后进行下一步设备连接和升级
go_work()
else:
print("t升级配置文件读取失败,请检查目录文件%s%supdate.json" % (os.getcwd(), os.sep))
print("t检测文件内容无误后在重新运行本程序")
input("按回车退出")
json中的内容
{
"//":"updateIP:升级的目标IP地址 updateFileName:升级包的文件名",
"updateIP":"192.168.1.xxx",
"updateFileName":"update_xxxxxx.bin"
}
7.py打包成exe
将.py搞成可以直接执行的exe文件,在文件目录(一般可以在script)下面cmd
pip install pyinstaller # 安装完成后直接打包需要的文件就好,重名会自动覆盖 pyinstaller -F xxxx.py
生成的exe文件就会在程序文件目录下面的dist里面出现,双击即可运行。
8.一些哔哔赖赖1、芯片对应的固件、fupdate文件、tftp64.exe和json文件都要可以对应版本,放在一个目录下面
2、另外升级结果除了设备会重启之外,具体版本号要手动查看确认是都升级成功!
3、fupdate是自己写的升级程序,这个版本的程序采取的是将bin包传到设备里面再进行升级,设备的flash可能不够用,而且升级慢一倍。但是fupdate不是的是不是因为版本不对应的问题一直不能远程升级。【找到原因了!是因为服务器url和本地ip搞错了!!!】
4、一定要给够系统和设备的反应时间,即使反馈消息
5、try和except的使用不是很熟悉,不过对于一些可以用条件判断的还是要用if和else
def pingHost(host):
#for i in range(1, 256):
success= 'PING SUCCESS'
failed= 'PING FAILED'
nowTime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
p = os.popen("ping "+ host + " -n 2")
line = p.read()
if "无法访问目标主机" in line:
print("***********"+nowTime, host, failed+"**********")
return False
else:
print("***********"+nowTime, host, success+"**********")
return True



