程序员经验分享-hwhale

python脚本——通过telnet连接设备

2023-05-16

文章目录

    • 一、说明
    • 二、代码
    • 三、用法总结

一、说明

通过telnetlib库，telnet到设备上并做一些测试。包括重启设备、等待重启完成、其它测试操作等。

二、代码

#! /usr/bin/python3.6
import logging
import telnetlib
import time


host_addr = '10.229.105.14'
port_num = '10017'


# 环境初始化
def init_env():
    global host_addr, port_num
    host_addr = input('请输入串口服务器地址：')
    if len(host_addr) == 0:
        host_addr = '10.229.105.14'
    print('串口服务器：', host_addr)

    port_num = input('请输入端口号：')
    if len(port_num) == 0:
        port_num = '10017'
    print('端口号: ', port_num)


class TelnetClient:
    def __init__(self,):
        self.session = telnetlib.Telnet()

    # 此函数实现telnet登录主机
    def login_host(self, host_addr, port_num):
        try:
            # self.session = telnetlib.Telnet(host_addr, port=port_num)
            self.session.open(host_addr, port=port_num)
        except:
            logging.warning('%s网络连接失败'%host_addr)
            return False
        # 延时两秒再收取返回结果，给服务端足够响应时间
        time.sleep(2)
        # 获取登录结果
        # read_very_eager()获取到的是的是上次获取之后本次获取之前的所有输出
        command_result = self.session.read_very_eager().decode('ascii')
        print('telnet result:', command_result)
        # if 'Login incorrect' not in command_result:
        #     logging.warning('%s登录成功'%host_addr)
        #     return True
        # else:
        #     logging.warning('%s登录失败，用户名或密码错误'%host_addr)
        #     return False
        return True

    # 退出telnet
    def logout_host(self):
        self.session.close()

	# 执行命令，不关心返回值
    def execute_command_without_res(self, command):
        # 执行命令
        self.session.write(command.encode('ascii')+b'\n')

    # 此函数实现执行传过来的命令，并输出其执行结果
    def execute_some_command(self, command, time_wait):
        # 执行命令
        self.execute_command_without_res(command)
        time.sleep(time_wait)
        # 获取命令结果
        command_result = self.session.read_very_eager().decode()
        # logging.warning(command + '执行结果：\n%s' % command_result)
        return command_result

	# 在timeout时间内，等待一些返回值（某些命令执行时间很长用得到）
    def wait_some_res(self, res, timeout):
        while timeout > 0:
            load_res = self.execute_some_command('\r', 1)
            # enter_res = self.session.read_very_eager().decode('ascii')
            if res in load_res:
                return True
            time.sleep(2)
            timeout -= 3
        return False

	# 执行一些命令，并在timeout时间内等待返回值
    def execute_command_wait_some_res(self, command, res, timeout):
        self.session.write(command.encode('ascii')+b'\n')
        while timeout>0:
            load_res = self.session.read_very_eager().decode('ascii')
            if res in load_res:
                return True
            time.sleep(2)
            timeout -= 2
        return False

	# 测试命令，升级boot
    def upgrade_boot(self, times):
        msg = 'echo upgrade boot the %d times' % (times)
        self.execute_command_without_res('\r')
        self.execute_command_without_res(msg)
        now_ver_cmd = 'dmidecode -t 0 |grep "BIOS Revision"'
        now_ver_res = self.execute_some_command(now_ver_cmd, 5)
        if '2.3' in now_ver_res:
            print('now version is 2.3')
            up_boot_cmd = 'echo 1 /home/admin/ZBOOT-TXCPC-X86-REL_4_4.bin > /sys/bsp/bootflash/std_program'
            # self.execute_some_command(up_boot_cmd, 2)
            if self.execute_command_wait_some_res(up_boot_cmd, 'Bootflash program success', 10*60):
                switch_cmd = 'echo 1 > /sys/bsp/bootflash/std_switch'
                self.execute_command_wait_some_res(switch_cmd, 'Bootflash switch success', 5)

        elif '4.4' in now_ver_res:
            print('now version is 4.4')
            up_boot_cmd = 'echo 1 /home/admin/ZBOOT-TXCPC-X86-REL_3_8.bin > /sys/bsp/bootflash/std_program'
            # self.execute_some_command(up_boot_cmd, 2)
            if self.execute_command_wait_some_res(up_boot_cmd, 'Bootflash program success', 10*60):
                switch_cmd = 'echo 1 > /sys/bsp/bootflash/std_switch'
                self.execute_command_wait_some_res(switch_cmd, 'Bootflash switch success', 5)
        else:
            print('get boot version error')

	# 测试命令，升级dbg版本boot
    def upgrade_dbg_boot(self, times):
        msg = 'echo upgrade boot the %d times' % (times)
        self.execute_command_without_res('\r')
        self.execute_command_without_res(msg)
        now_ver_cmd = 'dmidecode -t 0 |grep "BIOS Revision"'
        now_ver_res = self.execute_some_command(now_ver_cmd, 5)

        up_boot_cmd = 'echo 1 /home/admin/ZBOOT-TXCPC-X86-DBG_4_9.bin > /sys/bsp/bootflash/std_program'
        # self.execute_some_command(up_boot_cmd, 2)
        if self.execute_command_wait_some_res(up_boot_cmd, 'Bootflash program success', 10*60):
            switch_cmd = 'echo 1 > /sys/bsp/bootflash/std_switch'
            self.execute_command_wait_some_res(switch_cmd, 'Bootflash switch success', 5)

	# 测试命令，升级bootepld
    def upgrade_bootepld(self, times):
        msg = 'echo upgrade bootepld the %d times' % (times)
        self.execute_command_without_res('\r')
        self.execute_command_without_res(msg)
        bootepld_ver_cmd = 'cat /sys/bsp/epldflash/bootepld0/info |grep epldver'
        bootepld_ver_res = self.execute_some_command(bootepld_ver_cmd, 3)
        if '1.03' in bootepld_ver_res:
            print('now version is 1.03')
            up_epld_cmd = 'echo 2 0 /home/admin/TXCPC_00_210900_EPLD_D3_CFG2_106.vpd >/sys/bsp/epldflash/std_program'
            self.execute_command_without_res('\r')
            self.execute_command_without_res(up_epld_cmd)
        elif '1.06' in bootepld_ver_res:
            print('now version is 1.03')
            up_epld_cmd = 'echo 2 0 /home/admin/TXCPC_00_210900_EPLD_D3_CFG2_103.vpd >/sys/bsp/epldflash/std_program'
            self.execute_command_without_res('\r')
            self.execute_command_without_res(up_epld_cmd)
        else:
            print('get bootepld version error')
            return

        self.wait_some_res('root@kis:', 5)
        if self.wait_some_res('root@kis:', 2*60):
            success_msg = 'upgrade bootepld the %d times success!!' % (times)
            print(success_msg)

	# 测试命令，升级workepld
    def upgrade_workepld(self, times):
        msg = 'echo upgrade workepld the %d times' % (times)
        self.execute_command_without_res('\r')
        self.execute_command_without_res(msg)
        workepld_ver_cmd = 'cat /sys/bsp/epldflash/workepld1/info |grep epldver'
        workepld_ver_res = self.execute_some_command(workepld_ver_cmd, 3)
        if '1.08' in workepld_ver_res:
            print('now version is 1.08')
            up_epld_cmd = 'echo 1 0 /home/admin/ZXR10_5960M-8M-HX-TVMCCB-210900-D1A15-107.vpd > /sys/bsp/epldflash/std_program'
            self.execute_command_without_res('\r')
            self.execute_command_without_res(up_epld_cmd)
        elif '1.07' in workepld_ver_res:
            print('now version is 1.07')
            up_epld_cmd = 'echo 1 0 /home/admin/ZXR10_5960M-8M-HX-TVMCCB-210900-D1A15-108.vpd > /sys/bsp/epldflash/std_program'
            self.execute_command_without_res('\r')
            self.execute_command_without_res(up_epld_cmd)
        else:
            print('get bootepld version error')
            return

        self.wait_some_res('root@kis:', 5)
        if self.wait_some_res('root@kis:', 2*60):
            success_msg = 'upgrade bootepld the %d times success!!' % (times)
            print(success_msg)

	# 测试命令，重启环境
    def reboot_test(self):
        reboot_cmd = '/usr/sbin/reboot'
        if self.execute_command_wait_some_res(reboot_cmd, 'login:', 20*60):
            self.execute_command_wait_some_res('admin', 'Password:', 5)
            self.execute_command_without_res('kis1234')
            if self.wait_some_res('kis#', 10):
                self.execute_command_without_res('exit')
                self.execute_command_without_res('sudo su')
            elif self.wait_some_res('admin@kis:', 10):
                self.execute_command_without_res('sudo su')
        if self.wait_some_res('root@kis:', 10):
            self.execute_command_without_res('echo reboot success!!!!!!!')
        else:
            self.logout_host()


if __name__ == '__main__':
    # init_env()
    telnet_client = TelnetClient()
    # 如果登录结果返加True，则执行命令，然后退出
    if telnet_client.login_host(host_addr, port_num):
        for i in range(0, 500):
            # telnet_client.upgrade_dbg_boot(i)
            telnet_client.upgrade_bootepld(i)
            # telnet_client.upgrade_workepld(i)
            telnet_client.reboot_test()
    telnet_client.logout_host()

三、用法总结

具体用法看手册吧😊😊😊😊

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

python

telnet

连接设备

python脚本——通过telnet连接设备 的相关文章

随机推荐

  • Cannot convert a symbolic Tensor (simple_rnn/strided_slice:0) to a numpy array. 报错 （解决方法）

    1 报错 问题 xff1a xff08 来自 Python深度学习 P164 165 xff09 在运行以下代码的时候 xff0c 报错 xff1a Cannot convert a symbolic Tensor simple rnn s

  • 图像几何校正

    几何校正中混淆的概念 名词描述几何校正几何畸变会给基于遥感图像的定量分析 变化检测 图像融合 地图测量或更新等处理带来误差 xff08 主要指二维平面坐标 xff09 xff0c 所以需要针对图像的几何畸变进行校正 xff0c 也就是几何校

  • 解决VS2015 安装失败问题，如安装包损坏或丢失

    搜索包路径选择下面这个路径

  • 1.ODBC连接Postgresql

    SQLDriverConnect SQLDriverConnect 是 SQLConnect 的替代方法 它支持需要比 SQLConnect 中的三个参数更多的连接信息的数据源 对话框来提示用户输入所有连接信息 xff0c 以及系统信息中未

  • ImportError: DLL load failed while importing Qsci

    pyqt5报错 xff1b ImportError span class token operator span DLL load failed span class token keyword while span importing Q

  • 关于编译QGIS3.22.12配置的python库

    注意 xff1a 安装了多个python环境的很容易编译出错 xff1b 其中3 22需要配置python3 9的环境 在编译QGIS过程中的报错 报错1 xff1a No module named span class token cha

  • Qt中设置删除提示框

    Qt中设置删除提示框 QMessageBox StandardButton sButton 61 QMessageBox question NULL QObject tr 34 提示 34 QObject tr 34 该数据是否删除 34

  • 你真的对图像格式了解么？

    图像是人类视觉的基础 xff0c 是自然景物的客观反映 xff0c 是人类认识世界和人类本身的重要源泉 图 是物体反射或透射光的分布 xff0c 像 是人的视觉系统所接受的图在人脑中所形成的印象或认识 xff0c 照片 绘画 剪贴画 地图

  • 配置Qt中pro、pri文件

    002helloWorld pro span class token macro property span class token directive hash span span class token directive keywor

  • 构建一个字符串类

    文章目录 1 构建字符串数据 2 创建那些函数给外界调用 1 构造函数以及设计构造函数初值 2 类内带有指针 考虑3个特殊重要函数 3 考虑辅助函数 3 定义具体函数 1 设计构造函数内容

  • PostgreSQL查询

    PostgreSQL 数据库连接 QT xff1a 建立到数据库的连接 QSqlDatabase db 61 QSqlDatabase addDatabase 34 QPSQL 34 db setHostName 34 localhost

  • Conda install 报错：An HTTP error occurred when trying to retrieve this URL. HTTP errors are often...

    1 问题描述 xff1a 准备在Anaconda prompt执行以下命令 xff1a conda install c stellargraph stellargraph 报错 xff1a An HTTP error occurred wh

  • svn原理----revert，回滚

    一 子命令Svn revert 取消所有的本地编辑 下面我们来看一下子命令Svn revert例子 xff1a 丢弃对一个文件的修改 xff1a Svn revert foo c Reverted foo c 如果你希望恢复一整个目录的文件

  • Qt 自定义控件提升，头文件找不到的问题

    Qt 自定义控件提升 xff0c 头文件找不到的问题 在附加包含目录添加 xff1a

  • 分析int(*p)[4] = a

    面试题 xff1a 二级指针 include lt iostream gt int main int a 3 4 61 0 1 2 3 4 5 6 7 8 9 10 11 int p 4 61 a std cout lt lt p 43 1

  • af::convolve在CUDA中局限性

    使用在Cuda出现访问冲突问题 xff08 opengcl正常 xff09 xff1a af convolve I I kernel 报错 xff1a 0x00007FFC6443ADAC af dll 处 位于 XXXX exe 中 引发

  • 2016

    2016 最近 xff0c 许多朋友兴起总结2016了 xff0c 看得我心痒 xff0c 心热 我自己不禁也总结起来了 别人的总结要么是 2016XXXX 要么是 2016OOOO 我苦思2秒 xff0c 却想不起一个标题来 xff0c

  • gdb反汇编disassemble

    GDB Command Reference disassemble command gdb反汇编可用disassemble disass命令 用法如下 xff1a disassemble disassemble Function 指定要反汇

  • S.M.A.R.T. 参数(smartctl)计算硬盘精确健康值

    参考 xff1a Acronis Drive Monitor Disk Health Calculation 文章目录 1 背景2 smartctl a dev sda3 计算健康值3 1 关键参数3 1 1 公式说明3 2 2 计算举例

  • python脚本——通过telnet连接设备

    文章目录 一 说明二 代码三 用法总结 一 说明 通过telnetlib库 xff0c telnet到设备上并做一些测试 包括重启设备 等待重启完成 其它测试操作等 二 代码 span class token comment usr bin

热门标签