栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Python

基于python+telnetlib的设备通信+固件自动升级

Python 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

基于python+telnetlib的设备通信+固件自动升级

主要完成的功能流程包括
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下载的情况

1、获取本地网卡的ip,

用之前要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

***** 6月1日 更新 判断ping的方法添加 *******
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
转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/942254.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号