当前位置： 首页 > wzjs >正文

做室内意向图的网站长春网站推广排名

wzjs 2025/8/1 15:00:10
做室内意向图的网站,长春网站推广排名,温州网站建设制作公司,松江企业做网站一、项目背景与功能概述 本工具专为华为ONU设备运维设计，支持通过Telnet协议进行设备连接、命令下发与状态监控。核心功能包括： 多预设IP/账号/密码快速连接 实时命令控制台与自动周期指令 系统日志记录与可视化展示 集成Ping网络测试工具 二、代码…

一、项目背景与功能概述

本工具专为华为ONU设备运维设计，支持通过Telnet协议进行设备连接、命令下发与状态监控。核心功能包括：

  • 多预设IP/账号/密码快速连接

  • 实时命令控制台与自动周期指令

  • 系统日志记录与可视化展示

  • 集成Ping网络测试工具

二、代码架构解析

1. GUI框架搭建（PyQt5）
# GUI主窗口定义
class HuaweiONTGUI(QWidget):def __init__(self):super().__init__()self.client = HuaweiONTClient()  # Telnet客户端实例self.init_ui()                    # 初始化界面self.setup_logging()             # 日志系统绑定# 日志处理组件（桥接logging与GUI）
class GuiLogHandler(QObject, logging.Handler):log_signal = pyqtSignal(str)  # 自定义信号用于跨线程通信

关键布局：

  • 采用QHBoxLayout与QVBoxLayout构建左右分栏布局

  • 通过QGroupBox实现功能模块化分组

  • 使用qasync库实现异步事件循环，解决GUI阻塞问题

2. 连接管理模块
# Telnet客户端核心类
class HuaweiONTClient:class LoginState(Enum):  # 登录状态机INIT = auto()WAIT_PASSWORD = auto()COMPLETE = auto()async def connect(self, host, port):self.reader, self.writer = await asyncio.open_connection(host, port)await self._process_welcome_messages()  # 处理欢迎信息async def adaptive_login(self, username, password):return await self._telnet_login(username, password)

技术要点：

  • 状态机管理登录流程（INIT → WAIT_PASSWORD → COMPLETE）

  • 异步处理Telnet选项协商（_process_telnet_options）

  • 智能识别多语言登录提示（支持Login:/Username:等变体）

3. 命令交互系统
# 自动发送功能实现
def toggle_auto_send(self, state):if state == Qt.Checked:interval = int(self.interval_input.text()) * 1000self.auto_send_timer = QTimer()  # 创建定时器self.auto_send_timer.timeout.connect(self.on_auto_send)@asyncSlot()
async def on_auto_send(self):output = await self.client.write(cmd)  # 异步发送指令self.update_cmd_output(output)

交互布置：

  • 支持命令历史记录（QTextEdit持久化显示）

  • 输入框returnPressed事件绑定即时发送

  • 自动命令支持自定义间隔与指令（如定期查询版本信息）

4. 异常处理与日志系统
# 日志信号传递机制
class GuiLogHandler(logging.Handler):def emit(self, record):msg = self.format(record)self.log_signal.emit(msg)  # 信号触发GUI更新# 客户端异常捕获
async def _telnet_login(self, username, password):try:# ...登录流程...except Exception as e:self.logger.error(f"Login error: {str(e)}")return False

可靠性保障：

  • 双日志通道：文件（huawei_ont.log）+ GUI实时显示

  • 异步操作异常捕获与状态回显

  • 网络超时机制（asyncio.TimeoutError处理）

三、项目应用与扩展

典型使用场景：

  • 批量ONU设备固件版本检查

  • 光模块状态实时监控

  • 快速配置下发与故障排查

扩展方向：

  1. 增加SSH协议支持

  2. 实现配置模板化批量部署

  3. 集成SNMP监控模块

  4. 添加拓扑自动发现功能

四、一个成品原码分享

Telnet带GUI客户端原码：

import asyncio
import logging
import re
import sys
import time
from enum import Enum, auto
from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QHBoxLayout,QTextEdit, QLineEdit, QPushButton, QLabel, QGroupBox, QFormLayout, QComboBox, QCheckBox)
from PyQt5.QtCore import Qt, QObject, pyqtSignal, QTimer
from PyQt5.QtGui import QFont, QTextOption
from qasync import QEventLoop, asyncSlotclass GuiLogHandler(QObject, logging.Handler):log_signal = pyqtSignal(str)def __init__(self):super().__init__()logging.Handler.__init__(self)self.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))def emit(self, record):msg = self.format(record)self.log_signal.emit(msg)class HuaweiONTGUI(QWidget):def __init__(self):super().__init__()self.client = HuaweiONTClient()self.auto_send_timer = Noneself.init_ui()self.setup_logging()def init_ui(self):self.setWindowTitle('ONT Manager v2.2')self.setGeometry(300, 300, 1000, 700)self.setStyleSheet("""QWidget { background-color: #F5F5F5; }QGroupBox { border: 2px solid #0078D4;border-radius: 5px;margin-top: 1ex;font-weight: bold;}QGroupBox::title { subcontrol-origin: margin; padding: 0 3px; }QTextEdit, QLineEdit { border: 1px solid #CCCCCC;border-radius: 3px;padding: 3px;}QPushButton {background-color: #0078D4;color: white;border: none;padding: 5px 15px;border-radius: 4px;}QPushButton:hover { background-color: #006CBB; }QPushButton:disabled { background-color: #CCCCCC; }""")main_layout = QHBoxLayout()# Left Panelleft_panel = QVBoxLayout()# Connection Infoconn_group = QGroupBox("Connection Settings")form_layout = QFormLayout()self.ip_combo = QComboBox()self.ip_combo.addItems(['192.168.1.1'])self.ip_combo.setEditable(True)self.ip_combo.setCurrentIndex(0)# userself.user_combo = QComboBox()self.user_combo.addItems(['root'])self.user_combo.setEditable(True)# passself.pass_combo = QComboBox()self.pass_combo.addItems(['admin'])self.pass_combo.setEditable(True)self.pass_combo.setInsertPolicy(QComboBox.NoInsert)self.port_input = QLineEdit('23')form_layout.addRow(QLabel('IP:'), self.ip_combo)form_layout.addRow(QLabel('Port:'), self.port_input)form_layout.addRow(QLabel('User:'), self.user_combo)form_layout.addRow(QLabel('Password:'), self.pass_combo)conn_group.setLayout(form_layout)# Buttonsbtn_layout = QHBoxLayout()self.connect_btn = QPushButton('Connect')self.ping_btn = QPushButton('Ping Test')btn_layout.addWidget(self.connect_btn)btn_layout.addWidget(self.ping_btn)# Command Consolecmd_group = QGroupBox("Command Console")cmd_layout = QVBoxLayout()auto_send_layout = QHBoxLayout()self.auto_send_check = QCheckBox("Auto Send (sec)")self.interval_input = QLineEdit('5')self.auto_cmd_input = QLineEdit('display version')self.auto_cmd_input.setPlaceholderText("Auto command...")self.interval_input.setFixedWidth(60)self.auto_cmd_input.setFixedWidth(180)auto_send_layout.addWidget(self.auto_send_check)auto_send_layout.addWidget(self.interval_input)auto_send_layout.addWidget(QLabel("Command:"))auto_send_layout.addWidget(self.auto_cmd_input)auto_send_layout.addStretch()self.cmd_input = QLineEdit()self.cmd_input.setPlaceholderText("Enter command...")self.cmd_input.returnPressed.connect(self.on_send_command)self.cmd_output = QTextEdit()self.cmd_output.setReadOnly(True)self.send_btn = QPushButton('Send Command')cmd_layout.addLayout(auto_send_layout)cmd_layout.addWidget(self.cmd_input)cmd_layout.addWidget(self.send_btn)cmd_layout.addWidget(self.cmd_output)cmd_group.setLayout(cmd_layout)left_panel.addWidget(conn_group)left_panel.addLayout(btn_layout)left_panel.addWidget(cmd_group)# Right Panel (Logs)right_panel = QVBoxLayout()log_group = QGroupBox("System Logs")log_layout = QVBoxLayout()self.log_display = QTextEdit()self.log_display.setReadOnly(True)self.log_display.setFont(QFont("Consolas", 9))self.log_display.setWordWrapMode(QTextOption.WrapAnywhere)log_layout.addWidget(self.log_display)log_group.setLayout(log_layout)right_panel.addWidget(log_group)main_layout.addLayout(left_panel, 60)main_layout.addLayout(right_panel, 40)self.setLayout(main_layout)# Signalsself.connect_btn.clicked.connect(self.on_connect)self.ping_btn.clicked.connect(self.on_ping)self.send_btn.clicked.connect(self.on_send_command)self.auto_send_check.stateChanged.connect(self.toggle_auto_send)def toggle_auto_send(self, state):if state == Qt.Checked:interval = int(self.interval_input.text()) * 1000self.auto_send_timer = QTimer()self.auto_send_timer.timeout.connect(self.on_auto_send)self.auto_send_timer.start(interval)else:if self.auto_send_timer:self.auto_send_timer.stop()self.auto_send_timer = None@asyncSlot()async def on_auto_send(self):cmd = self.auto_cmd_input.text().strip()if not cmd:returnself.update_cmd_output(f"> [AUTO] {cmd}")try:output = await self.client.write(cmd)if output:self.update_cmd_output(output)except Exception as e:self.update_cmd_output(f"Auto Command Error: {str(e)}")def setup_logging(self):self.gui_handler = GuiLogHandler()self.gui_handler.log_signal.connect(self.update_log_display)self.client.logger.addHandler(self.gui_handler)def update_log_display(self, text):self.log_display.append(text)self.log_display.verticalScrollBar().setValue(self.log_display.verticalScrollBar().maximum())@asyncSlot()async def on_connect(self):self.connect_btn.setEnabled(False)try:success = await self.client.connect(host=self.ip_combo.currentText(),port=int(self.port_input.text()))if success:login_success = await self.client.adaptive_login(username=self.user_combo.currentText(),password=self.pass_combo.currentText())if login_success:self.update_cmd_output("\nLogin successful! Ready to send commands.")except Exception as e:self.log_display.append(f"Error: {str(e)}")finally:self.connect_btn.setEnabled(True)@asyncSlot()async def on_ping(self):self.ping_btn.setEnabled(False)try:host = self.ip_combo.currentText()proc = await asyncio.create_subprocess_shell(f"ping -n 4 {host}" if sys.platform == 'win32' else f"ping -c 4 {host}",stdout=asyncio.subprocess.PIPE,stderr=asyncio.subprocess.PIPE)stdout, _ = await proc.communicate()output = stdout.decode('gbk' if sys.platform == 'win32' else 'utf-8', errors='replace')self.log_display.append(f"\nPing Results:\n{output}")except Exception as e:self.log_display.append(f"Ping Error: {str(e)}")finally:self.ping_btn.setEnabled(True)def update_cmd_output(self, text):self.cmd_output.append(text)self.cmd_output.verticalScrollBar().setValue(self.cmd_output.verticalScrollBar().maximum())@asyncSlot()async def on_send_command(self):cmd = self.cmd_input.text().strip()if not cmd:returnself.cmd_input.clear()self.update_cmd_output(f"> {cmd}")try:output = await self.client.write(cmd)if output:self.update_cmd_output(output)except Exception as e:self.update_cmd_output(f"Command Error: {str(e)}")class HuaweiONTClient:class LoginState(Enum):INIT = auto()WAIT_USERNAME = auto()WAIT_PASSWORD = auto()COMPLETE = auto()def __init__(self):self.reader = Noneself.writer = Noneself.buffer = b''self.login_state = self.LoginState.INITself.telnet_commands = {b'\xff\xfb\x01': b'\xff\xfd\x01',b'\xff\xfb\x03': b'\xff\xfd\x03',b'\xff\xfb\x18': b'\xff\xfd\x18'}self.logger = logging.getLogger('HuaweiONT')self.logger.setLevel(logging.DEBUG)formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')file_handler = logging.FileHandler('huawei_ont.log', mode='a')file_handler.setFormatter(formatter)console_handler = logging.StreamHandler()console_handler.setFormatter(formatter)self.logger.addHandler(file_handler)self.logger.addHandler(console_handler)async def _process_telnet_options(self):processed = Falsefor cmd, response in self.telnet_commands.items():while cmd in self.buffer:pos = self.buffer.find(cmd)self.writer.write(response)self.buffer = self.buffer[:pos] + self.buffer[pos+len(cmd):]self.logger.debug(f"Responded to Telnet option: {cmd!r}")processed = Trueawait self.writer.drain()return processedasync def _detect_login_prompt(self):patterns = [b'Login:',b'Username:',b'User name:',b'login:',b'\r\nlogin:',b'\r\nUser:']for pattern in patterns:if pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]self.logger.debug(f"Detected login prompt: {pattern!r}")return patternif b'\r\n' in self.buffer and len(self.buffer.split(b'\r\n')[-1].strip()) == 0:self.logger.debug("Detected empty line prompt")return b'implicit_prompt'return Noneasync def _process_welcome_messages(self):welcome_patterns = [b'Welcome Visiting Huawei',b'Copyright by Huawei',b'Huawei Home Gateway',b'Huawei Technologies Co',b'\r\n\r\nlogin:',b'\r\n\r\nLogin:']cleaned = Falsefor pattern in welcome_patterns:if pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]self.logger.debug(f"Cleaned welcome pattern: {pattern!r}")cleaned = Truereturn cleanedasync def _smart_expect(self, patterns, timeout=10):if not isinstance(patterns, list):patterns = [patterns]compiled_patterns = []for pattern in patterns:if isinstance(pattern, str):pattern = pattern.encode('ascii')compiled_patterns.append(re.compile(pattern) if b'(' in pattern else pattern)end_time = asyncio.get_event_loop().time() + timeoutwhile True:await self._process_telnet_options()for i, pattern in enumerate(compiled_patterns):if isinstance(pattern, re.Pattern):match = pattern.search(self.buffer)if match:self.buffer = self.buffer[match.end():]return ielif pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]return iif asyncio.get_event_loop().time() > end_time:raise asyncio.TimeoutError(f"No patterns matched: {patterns}")if not await self._raw_read(timeout=0.5):await asyncio.sleep(0.1)async def _raw_read(self, timeout=1):try:data = await asyncio.wait_for(self.reader.read(4096), timeout=timeout)if data:self.buffer += dataself.logger.debug(f"Received: {data!r}")return bool(data)except asyncio.TimeoutError:return Falseasync def connect(self, host='192.168.1.1', port=23):try:self.reader, self.writer = await asyncio.open_connection(host, port)self.logger.info(f"Connected to {host}:{port}")await self._raw_read(timeout=2)await self._process_welcome_messages()return Trueexcept Exception as e:self.logger.error(f"Connection failed: {str(e)}")return Falseasync def write(self, data, hide=False):if not isinstance(data, bytes):data = data.encode('ascii')if not data.endswith(b'\r\n'):data += b'\r\n'self.writer.write(data)await self.writer.drain()if hide:self.logger.debug("Sent hidden data (password)")else:self.logger.debug(f"Sent: {data!r}")await asyncio.sleep(0.5)await self._raw_read(timeout=2)return self.buffer.decode('ascii', errors='ignore')async def adaptive_login(self, username='root', password='admin'):return await self._telnet_login(username, password)async def _telnet_login(self, username, password):try:self.login_state = self.LoginState.INITstart_time = asyncio.get_event_loop().time()while self.login_state != self.LoginState.COMPLETE:if asyncio.get_event_loop().time() - start_time > 15:raise asyncio.TimeoutError("Login timeout")if self.login_state == self.LoginState.INIT:prompt = await self._detect_login_prompt()if prompt:await self.write(username)self.login_state = self.LoginState.WAIT_PASSWORDself.logger.debug("Username sent, waiting for password...")else:await self._process_welcome_messages()await asyncio.sleep(0.5)elif self.login_state == self.LoginState.WAIT_PASSWORD:prompt = await self._detect_login_prompt()if prompt == b'implicit_prompt':await self.write(password, hide=True)self.login_state = self.LoginState.COMPLETEelif any(p in self.buffer.lower() for p in [b'password:', b'passwd:']):await self.write(password, hide=True)self.login_state = self.LoginState.COMPLETEawait self._raw_read(timeout=0.5)await asyncio.sleep(1)if b'incorrect' in self.buffer.lower():raise PermissionError("Invalid credentials")self.logger.info("Telnet Login successful")return Trueexcept Exception as e:self.logger.error(f"Login error: {str(e)}")return Falseif __name__ == '__main__':app = QApplication(sys.argv)loop = QEventLoop(app)asyncio.set_event_loop(loop)window = HuaweiONTGUI()window.show()with loop:sys.exit(loop.run_forever())

http://www.dtcms.com/wzjs/156281.html

相关文章：

  • 烟草电子商务网站企业网站模板html
  • 本田汽车网站建设如何让别人在百度上搜到自己公司
  • 做网站要学哪些全球最大的磁力搜索引擎
  • 无锡高端网站建设机构友情链接代码模板
  • html5响应式网站开发百度网盘搜索引擎入口哪里
  • 北京网站建设有哪些怎么联系百度人工服务
  • 视觉差网站制作关键词排名怎么做上去
  • 南京电商网站开发百度网站关键词排名查询
  • 住建局现任领导班子优化大师怎么样
  • 做网站贵吗百度推广优化师
  • wordpress表单统计插件下载成都seo经理
  • 网站规划是什么意思北京网
  • 网站建设的网络潍坊网站定制模板建站
  • 怎么网上推广自己的门店排名优化怎么做
  • 广州seo网站开发苹果aso优化
  • 购物网站页面重庆seo顾问服务
  • 做数据图表网站抖音seo优化怎么做
  • 银行网站建设方案视频网站推广优化
  • asp转换手机网站网站的seo
  • 推广模式有几种越秀seo搜索引擎优化
  • 网站建设怎么推广网络推广公司网站
  • 太原营销型网站凡科网站建站教程
  • 常州企业建站系统模板想要网站导航推广页
  • 营销型网站建设的特点表现苏州百度 seo
  • 重庆网站备案系统网络营销促销策略有哪些
  • 防止网站独立ip被别人绑定如何推广
  • 德州做网站的公司有哪些seo自动发布外链工具
  • java做的网站放哪打开百度网站
  • 襄州区住房和城乡建设局网站近期新闻热点大事件
  • 网页设计模板流程图昆明seo工资