【Python】OPC UA 服务器扫描工具

引言

 OPC UA(开放平台通信统一架构)是一种跨平台的、开放的数据交换标准,广泛用于工业自动化领域。在工业4.0的大背景下,OPC UA服务器在网络中的部署日益增多,如何快速有效地发现这些服务器成为了一个实际需求。本文将介绍如何使用Python编写一个简单的OPC UA服务器扫描工具。

准备工作

在开始编写代码之前,需要确保您的Python环境中安装了opcua库。可以通过以下命令进行安装:

pip install opcua

代码实现

我们的脚本将从扫描本地网络中的一系列IP地址开始,尝试连接到默认的OPC UA端口(4840),并记录下能够成功连接的服务器地址和端口。

导入库

首先,我们需要导入必要的库。

import sys
import time
from opcua import Client
import threading
import logging

日志设置

为了记录错误信息,我们设置了日志记录功能,将错误信息保存到文件中。

logger = logging.getLogger()
fh = logging.FileHandler("error.log", encoding="utf-8", mode="a")
formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.setLevel(logging.ERROR)

进度条函数

为了提供友好的用户界面,我们实现了一个打印进度条的函数。

def print_progress_bar(completed, total, length=50):
    progress = int(length * completed / total)
    bar = '扫描进度 [' + '=' * progress + '-' * (length - progress) + ']'
    percent = round(100.0 * completed / total, 1)
    sys.stdout.write(f'\r{bar} {percent}%')
    sys.stdout.flush()

扫描服务器线程函数

扫描服务器的操作将在线程中执行,这样可以同时扫描多个地址,提高效率。

def scan_server(ip_address, default_port, servers, i, length):
    try:
        print_progress_bar(i, length)
        client = Client(f"opc.tcp://{ip_address}:{default_port}")
        client.connect()
        # 如果成功连接,则将服务器地址和端口添加到列表中
        servers.append((ip_address, default_port))
        client.disconnect()
    except Exception as e:
        # 如果连接失败,则忽略该地址
        logging.error("%s 异常信息: %s", str(ip_address), str(e))

主扫描函数

主函数scan_opc_servers负责创建和管理工作线程,并收集扫描结果。

def scan_opc_servers():
    # 扫描本地网络上的所有 OPC UA 服务器
    servers = []
    # 默认的 OPC UA 服务器端口为 4840
    default_port = 4840

    # 遍历本地网络上的可能的 IP 地址范围
    ip_addresses = [f"192.168.1.{i}" for i in range(1, 256)] +  [f"192.168.168.{i}" for i in range(1, 256)]  # 您可能需要更改此处的 IP 地址范围

    # 创建一个线程池
    thread_pool = []
    max_threads = 75

    for i, ip_address in enumerate(ip_addresses):
        time.sleep(0.05)
        
        # 如果当前线程数量达到最大值,则等待一个线程完成
        if len(thread_pool) >= max_threads:
            thread_pool[0].join()
            thread_pool.pop(0)

        # 创建一个新的线程来扫描服务器
        thread = threading.Thread(target=scan_server, args=(ip_address, default_port, servers, i, len(ip_addresses)),)
        thread.start()
        thread_pool.append(thread)

    # 等待所有线程完成
    for thread in thread_pool:
        thread.join()

    return servers

主程序入口

最后,我们定义了程序的入口点,调用扫描函数,并打印出找到的服务器列表。

if __name__ == "__main__":
    opc_servers = scan_opc_servers()
    print_progress_bar(1, 1)
    print("\nFound OPC UA Servers:")
    for server in opc_servers:
        print(f"Address: {server[0]}, Port: {server[1]}")

结果

请添加图片描述
日志:
在这里插入图片描述

总结

通过以上代码,我们实现了一个简单的OPC UA服务器扫描工具。它可以在本地网络中快速扫描并发现OPC UA服务器,帮助工程师和开发人员更高效地进行网络管理和系统维护。


完整代码:

import sys
import time
from opcua import Client
import threading
import logging

logger = logging.getLogger()
fh = logging.FileHandler("error.log", encoding="utf-8", mode="a")
formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.setLevel(logging.ERROR)


def print_progress_bar(completed, total, length=50):
    progress = int(length * completed / total)
    bar = '扫描进度 [' + '=' * progress + '-' * (length - progress) + ']'
    percent = round(100.0 * completed / total, 1)
    sys.stdout.write(f'\r{bar} {percent}%')
    sys.stdout.flush()

# 定义一个线程工作函数
def scan_server(ip_address, default_port, servers, i, length):
    try:
        print_progress_bar(i, length)
        client = Client(f"opc.tcp://{ip_address}:{default_port}")
        client.connect()
        # 如果成功连接,则将服务器地址和端口添加到列表中
        servers.append((ip_address, default_port))
        client.disconnect()
    except Exception as e:
        # 如果连接失败,则忽略该地址
        logging.error("%s 异常信息: %s", str(ip_address), str(e))

def scan_opc_servers():
    # 扫描本地网络上的所有 OPC UA 服务器
    servers = []
    # 默认的 OPC UA 服务器端口为 4840
    default_port = 4840

    # 遍历本地网络上的可能的 IP 地址范围
    ip_addresses = [f"192.168.1.{i}" for i in range(1, 256)] +  [f"192.168.168.{i}" for i in range(1, 256)]  # 您可能需要更改此处的 IP 地址范围

    # 创建一个线程池
    thread_pool = []
    max_threads = 75

    for i, ip_address in enumerate(ip_addresses):
        time.sleep(0.05)
        
        # 如果当前线程数量达到最大值,则等待一个线程完成
        if len(thread_pool) >= max_threads:
            thread_pool[0].join()
            thread_pool.pop(0)

        # 创建一个新的线程来扫描服务器
        thread = threading.Thread(target=scan_server, args=(ip_address, default_port, servers, i, len(ip_addresses)),)
        thread.start()
        thread_pool.append(thread)

    # 等待所有线程完成
    for thread in thread_pool:
        thread.join()

    return servers

if __name__ == "__main__":
    opc_servers = scan_opc_servers()
    print_progress_bar(1, 1)
    print("\nFound OPC UA Servers:")
    for server in opc_servers:
        print(f"Address: {server[0]}, Port: {server[1]}")

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/527390.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【opencv】示例-create_mask.cpp 通过鼠标交互产生掩膜图像(黑白图像)

/* * create_mask.cpp * * Author: * Siddharth Kherada <siddharthkherada27[at]gmail[dot]com> * * 这个教程演示了如何制作掩膜图像&#xff08;黑白图像&#xff09;。 * 该程序将输入图像作为源图像&#xff0c;并输出对应的掩膜图像。 */#include "opencv2/im…

【opencv】示例-demhist.cpp 调整图像的亮度和对比度,并在GUI窗口中实时显示调整后的图像以及其直方图。...

#include "opencv2/core/utility.hpp" // 包含OpenCV核心工具库的头文件 #include "opencv2/imgproc.hpp" // 包含OpenCV图像处理的头文件 #include "opencv2/imgcodecs.hpp" // 包含OpenCV图像编码解码的头文件 #include "opencv2/highgui…

3.1_2 内存管理的概念

3.1_2 内存管理的概念 重要考点

不使用 Docker 构建 Triton 服务器并在 Google Colab 平台上部署 HuggingFace 模型

Build Triton server without docker and deploy HuggingFace models on Google Colab platform EnvironmentBuilding Triton serverDeploying HuggingFace models客户端推荐阅读参考 Environment 根据Triton 环境对应表 &#xff0c;Colab 环境缺少 tensorrt-8.6.1&#xff0…

matlab读/写文本数据(矩阵格式)

1、背景介绍 文本格式的点云数据是一种常见的形式,其形式一般如下:x y z。后面可能带有标签、颜色等信息。在使用matlab进行数据处理时,常用的2个操作:读取与写入。有时,点云之间用逗号‘,’进行间隔。 2、数据读取 对于上述两种形式,均可使用load函数进行读取,如下:…

Kali系统的常用安装后开机配置(都是实用的配置)

目录 一.查看当前操作系统的所有版本信息 二.开启root并修改密码 二.启动 Apache 服务&#xff08;用不到可忽略&#xff09; 三.开启Secure Shell&#xff08;ssh&#xff09;服务&#xff0c;才可以使用远程工具&#xff08;xshell&#xff0c;finalshell&#xff09;连接…

(虚拟DOM)前端八股文修炼Day10

一 虚拟 DOM 是什么 虚拟 DOM (Virtual DOM) 本质上是真实 DOM 的一个轻量级的 JavaScript 表示形式。它是一个在内存中的抽象&#xff0c;用于描述真实 DOM 的结构和内容。虚拟 DOM 提供了一种机制&#xff0c;允许开发者通过操作 JavaScript 对象来间接更新页面&#xff0c;…

Clip论文学习

学习过程如上 学习时&#xff0c;图片与文本对应&#xff08;句子&#xff09;&#xff0c;共有n个句子与n个文本&#xff0c;进行对比学习 测试时&#xff0c;将标签句子化&#xff0c;再将其放入encoder中操作。最后找出特征最相似的。 特点&#xff1a;泛化能力强&#x…

点击上传文件

一、页面样式&#xff1a; &#xff08;1&#xff09;点击前&#xff1a; &#xff08;2&#xff09;点击后&#xff1a; 设计&#xff1a;①自定义elementPlus图标&#xff1b;②使用Tooltip实现鼠标悬浮按钮上出现文字提示&#xff1b;③上传与更换的切换样式&#xff1b;…

蓝桥备赛——素数

素数定义 试除法判断 from math import * def if_prime(n):if n==1:return Falsem=int(sqrt(n)+1)for i in range(2,m): #范围是2到根下n+1if n%i==0:return Falsereturn True print(if_prime(5)) 试除法判断素数。就是判断是否除了1和本身还有其他可以整除自己的数,…

模块十三 异常_Object

回顾与重点 模块十二回顾 权限修饰符&#xff1a; public → protected → 默认 → private a. 构造一般用public&#xff1a;便于new对象 b. 成员方法一般用public&#xff1a;便于调用 c. 属性一般用private&#xff1a;封装思想 final&#xff1a;最终的 a. 修饰类&#x…

如何从iPhone恢复已删除的照片

这些照片是我们今天生活中最珍贵的回忆。它们捕捉了我们一生珍惜的时刻。但是&#xff0c;丢失这些记忆可能是毁灭性的&#xff0c;尤其是在不小心从iPhone中删除它们时。 幸运的是&#xff0c;有一些方法可以从iPhone恢复已删除的照片。在这篇博文中&#xff0c;我们将讨论从…

【opencv】示例-detect_mser.cpp 使用 MSER 算法来检测图像中的极值区域

#include <opencv2/core.hpp> // 包含OpenCV核心操作的头文件 #include <opencv2/imgproc.hpp> // 包含图像处理功能的头文件 #include <opencv2/highgui.hpp> // 包含图形用户界面的头文件 #include <opencv2/features2d.hpp> // 包含特征检测相关功能…

【数据结构】考研真题攻克与重点知识点剖析 - 第 5 篇:树与二叉树

前言 本文基础知识部分来自于b站&#xff1a;分享笔记的好人儿的思维导图与王道考研课程&#xff0c;感谢大佬的开源精神&#xff0c;习题来自老师划的重点以及考研真题。此前我尝试了完全使用Python或是结合大语言模型对考研真题进行数据清洗与可视化分析&#xff0c;本人技术…

C# + OpencvSharp4 错误信息收集

异常1&#xff1a; 初次使用&#xff0c;如下代码报错&#xff0c;OpenCvSharp.OpenCvSharpException:“imread failed.” Mat src Cv2.ImRead("Source.png", ImreadModes.Unchanged); 原因&#xff1a;检查Nuget包与OpencvSharp4库相关安装是否完整&#xff0c;…

虚拟地址的设置与应用全攻略

在计算机网络和操作系统中&#xff0c;虚拟地址是一个核心概念&#xff0c;它允许程序在逻辑上拥有连续的内存空间&#xff0c;而无需关心实际的物理内存布局。虚拟地址的引入极大地简化了程序的开发和内存管理&#xff0c;提高了系统的灵活性和安全性。虎观代理将详细解析虚拟…

Spring Boot中整合JodConverter实现文件在线预览

Spring Boot中整合JodConverter实现文件在线预览 1.安装LibreOffice 24.2 下载地址 LibreOffice 是一款功能强大的办公软件&#xff0c;默认使用开放文档格式 (OpenDocument Format , ODF), 并支持 *.docx, *.xlsx, *.pptx 等其他格式。 它包含了 Writer, Calc, Impress, Dra…

java基于微服务的智慧工地管理云平台SaaS源码 数据大屏端 APP移动端

目录 智慧工地云平台概述 智慧工地功能 &#xff08;项目端&#xff09; 一、劳务管理&#xff1a; 二、施工安全管理&#xff1a; 三、视频监控管理&#xff1a; 四、机械安全管理&#xff1a; 五、施工质量管理&#xff1a; 六、绿色文明施工&#xff1a; 七、施工综…

CSS3如何实现雷达扫描图(动态样式)

动态样式控制雷达扫描和暂停&#xff1a; //html部分&#xff1a; <view class"radar" :style"{--state:animationPlayState}"></view>data部分&#xff1a; animationPlayState: paused, methods: changeStatus(){this.animationPlayState …

Failed to start docker.service: Unit is not loaded properly: Invalid argument.

Failed to start docker.service: Unit is not loaded properly: Invalid argument. 未知原因&#xff1a;docker服务无法正常load 解决方式&#xff1a; 卸载docker&#xff0c; 删除docker.service 重新安装docker Docker是一种相对使用较简单的容器&#xff0c;我们可以通过…