简介本资源是一个面向工业自动化初学者与嵌入式/工控软件开发者的OPC服务器入门实践项目聚焦OPCOLE for Process Control标准的核心实现原理帮助开发者理解COM组件注册、OPC接口如IOPCServer、IOPCItemMgt、IOPCSyncRead等编程、数据项建模及多线程服务机制。压缩包共57个文件含32个CPP源码涵盖服务器主逻辑、组管理、项操作、连接点实现等、15个H头文件定义接口、辅助类与缓冲结构、1个SOLUTION工程及配套VCXPROJ配置整体仅121KB轻量易读适合VS2012环境直接编译调试。已有1931人学习下载代码结构清晰模块职责分明——如ICpc.cpp实现连接点容器SERVER.CPP承载核心服务生命周期ITEM.CPP与GROUP.CPP分别管理数据项与组订阅配合ReadMe.txt说明可快速掌握OPC Server从注册、接口实现到设备数据桥接的完整链路。1. 为什么“一个简单的OPC Server实现”不是玩具项目而是工业现场调试的刚需入口在某高校自动化实验室调试一条模拟灌装产线时A同学卡在了最后一步上位机HMI死活读不到PLC寄存器里的液位值。排查两整天后发现不是网线松了、IP没配对也不是PLC程序有误——而是HMI只认OPC UA协议而PLC自带的Modbus TCP服务根本不在它的通信白名单里。这时候“一个简单的OPC Server实现”就从教材里的概念变成了必须30分钟内跑通的救命方案。它不追求高并发、不堆集群、不接云平台核心就三点能被标准OPC UA客户端如UaExpert、Prosys OPC UA Client发现、能映射真实或模拟的变量地址、能响应读写请求且不崩。适合刚接触工业协议栈的嵌入式开发者、需要快速验证数据链路的自动化工程师以及在国产化替代中要对接老旧设备的系统集成人员。它不是替代商业OPC服务器的方案而是帮你绕过许可证墙、理解地址空间建模、看清UA安全策略实际生效位置的黑匣子——很多翻车其实发生在证书握手之前连“连接成功”四个字都没机会打印出来。2. 用 Python FreeOpcUa 搭建最小可运行 OPC UA Server从零启动到变量可见FreeOpcUa 是目前 Python 生态中最轻量、文档最贴近实际调试场景的开源 OPC UA 栈。它不依赖 Windows COM 组件纯 Python 实现调试时能直接看到每个 UA 服务调用的原始请求结构对理解协议本质极友好。注意这里不选 opcua-asyncio异步模型对新手不友好、也不用 Node-RED 插件封装过深看不到底层地址空间操作目标是“改一行代码就能让UaExpert里多出一个变量”。2.1 安装与基础骨架避开 wheel 编译陷阱# 推荐使用 conda 创建干净环境避免 pip install 时因 lxml/cryptography 版本冲突导致编译失败 conda create -n opcua-env python3.9 conda activate opcua-env pip install freeopcua提示若遇到lxml编译报错执行conda install lxml后再装freeopcuaWindows 用户务必关闭杀毒软件实时扫描否则opcua-server进程可能被误杀。2.2 最小可运行 Server5 行代码暴露一个浮点数变量from opcua import Server import time # 1. 实例化服务器对象绑定本地任意可用端口默认4840若被占用可改4841 server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) # 2. 设置服务器信息UaExpert 连接列表里显示的名称 server.set_server_name(SimpleTestServer) # 3. 启动服务器此时已可被发现但无变量 server.start() print(fServer started at {server.endpoint}) # 4. 获取默认地址空间根节点Objects文件夹 root server.get_objects_node() # 5. 在 Objects 下添加一个变量节点命名为 Temperature类型为 Float var root.add_variable(ns2;i1, Temperature, 25.6) # 6. 设置变量为可读写默认只读 var.set_writable(True) try: while True: # 模拟温度缓慢变化每2秒0.1℃ current_val var.get_value() var.set_value(current_val 0.1) time.sleep(2) except KeyboardInterrupt: server.stop() print(Server stopped)逻辑说明与参数拆解ns2;i1是节点IDNodeID的字符串表示ns2指命名空间索引2FreeOpcUa自动为自定义节点分配ns2i1是整数型ID确保唯一性实际项目中建议用ns2;sTemperature字符串ID更易维护。add_variable()第三个参数25.6是初始值类型由值自动推断为Float若传42则为Int32传hello则为String。set_writable(True)必须显式调用否则UaExpert里该变量灰显不可写。time.sleep(2)不是占位符——OPC UA 协议要求服务器主动推送值变更需通过MonitoredItem机制此处用轮询模拟仅用于验证读写通路真实场景应结合server.start()后的事件循环或定时器回调。2.3 用 UaExpert 验证确认三件事才算真正跑通发现服务打开UaExpert → “Discovery”页签 → 点击“Find Servers” → 列表中出现SimpleTestServer若未出现检查防火墙是否放行4840端口或尝试ping 127.0.0.1确认本地回环正常。连接成功双击该服务器 → 弹窗选择“Anonymous”认证FreeOpcUa默认禁用用户名密码不配置证书即走匿名→ 点击“Connect”状态栏变绿且显示Connected。读写变量左侧地址空间树展开Objects→ 找到Temperature节点 → 右键“Browse” → 在右侧Value栏看到实时跳动的浮点数如25.6,25.7,25.8右键该节点 → “Write Value” → 输入100.0→ 回车 → 左侧Value立即更新证明写通道有效。这三步缺一不可。很多初学者卡在第1步服务未被发现本质是server.set_endpoint()的URL格式错误漏掉末尾斜杠/或绑定地址写成127.0.0.1导致局域网其他机器无法访问。3. 把模拟变量升级为真实数据源对接 Modbus RTU/TCP、串口传感器与内存数据库一个“简单”的OPC Server 若只能生成随机数工业现场会直接弃用。真实价值在于它作为协议转换桥接层把非OPC设备的数据翻译成UA标准语义。FreeOpcUa 本身不内置Modbus驱动但其架构天然支持外挂数据源——关键在var.set_value()的调用时机和数据获取方式。3.1 对接 Modbus TCP 设备用 pymodbus 读取 PLC 寄存器假设目标PLC IP为192.168.1.100保持线圈地址00001对应Modbus功能码01需映射为OPC变量MotorStatusfrom pymodbus.client import ModbusTcpClient from opcua import Server import time # 初始化Modbus客户端超时设为2秒避免阻塞OPC主线程 modbus_client ModbusTcpClient(192.168.1.100, timeout2) # OPC Server初始化同前 server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) server.set_server_name(ModbusBridge) server.start() root server.get_objects_node() motor_status_var root.add_variable(ns2;sMotorStatus, MotorStatus, False) motor_status_var.set_writable(False) # PLC状态只读 def read_modbus_and_update(): 单次读取Modbus并更新OPC变量 try: # 读取线圈00001地址0数量1 result modbus_client.read_coils(0, 1) if not result.isError(): new_value result.bits[0] # 返回布尔列表取第一个 motor_status_var.set_value(new_value) else: print(fModbus read error: {result}) except Exception as e: print(fModbus connection failed: {e}) # 主循环每500ms读一次避免频繁轮询压垮PLC try: while True: read_modbus_and_update() time.sleep(0.5) except KeyboardInterrupt: modbus_client.close() server.stop()参数说明read_coils(0, 1)中0是起始地址Modbus协议从0开始计数对应线圈000011是读取数量若需读保持寄存器4xxxx改用read_holding_registers(0, 1)。timeout2是关键——若PLC断电或网络中断pymodbus 默认等待7.5秒才超时会拖慢整个OPC Server心跳导致UaExpert显示“BadWaitingForInitialData”。set_writable(False)明确告知客户端此变量不可写避免误操作触发PLC异常。3.2 对接串口温湿度传感器用 pyserial 解析 ASCII 协议某型号传感器通过RS485输出T:23.5,H:45.2\n格式文本需解析后映射为两个OPC变量import serial from opcua import Server import re import time # 打开串口根据实际设备修改 port 和 baudrate ser serial.Serial(/dev/ttyUSB0, 9600, timeout1) server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) server.set_server_name(SensorBridge) server.start() root server.get_objects_node() temp_var root.add_variable(ns2;sSensorTemp, SensorTemp, 0.0) humi_var root.add_variable(ns2;sSensorHumi, SensorHumi, 0.0) def parse_sensor_data(): 读取串口一行用正则提取温度湿度 try: line ser.readline().decode(ascii, errorsignore).strip() if not line: return None, None # 匹配 T:xx.x,H:yy.y 格式 match re.match(rT:(\d\.\d),H:(\d\.\d), line) if match: return float(match.group(1)), float(match.group(2)) except Exception as e: print(fSerial parse error: {e}) return None, None try: while True: temp, humi parse_sensor_data() if temp is not None and humi is not None: temp_var.set_value(temp) humi_var.set_value(humi) time.sleep(1) except KeyboardInterrupt: ser.close() server.stop()关键细节errorsignore防止传感器发送乱码导致decode()报错崩溃strip()去除换行符干扰正则匹配。正则rT:(\d\.\d),H:(\d\.\d)严格匹配小数点格式避免误捕获T:100,H:200整数导致float()失败。若传感器响应慢time.sleep(1)可延长至2但需同步调整UaExpert中订阅的“Sampling Interval”采样间隔否则客户端缓存旧值。3.3 对接内存数据库用 SQLite 实现变量持久化与历史查询OPC UA 规范要求服务器支持HistoryRead服务但FreeOpcUa默认不启用。我们用SQLite记录变量写入历史再通过自定义方法提供历史数据import sqlite3 from datetime import datetime from opcua import Server, ua # 初始化SQLite表结构timestamp TEXT, nodeid TEXT, value REAL conn sqlite3.connect(opc_history.db) conn.execute( CREATE TABLE IF NOT EXISTS history ( timestamp TEXT, nodeid TEXT, value REAL ) ) conn.commit() server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) server.set_server_name(HistoryServer) server.start() root server.get_objects_node() counter_var root.add_variable(ns2;sCounter, Counter, 0) counter_var.set_writable(True) # 重写变量的setter增加写入历史 original_setter counter_var.set_value def history_aware_setter(value): original_setter(value) # 写入SQLite conn.execute( INSERT INTO history (timestamp, nodeid, value) VALUES (?, ?, ?), (datetime.now().isoformat(), ns2;sCounter, float(value)) ) conn.commit() counter_var.set_value history_aware_setter # 添加一个方法节点供客户端调用获取历史UaExpert中右键调用 history_method root.add_method( ns2;sGetHistory, GetHistory, [ua.VariantType.String, ua.VariantType.DateTime, ua.VariantType.DateTime], # 参数nodeid, start, end [ua.VariantType.String, ua.VariantType.Double] # 返回timestamp, value ) uamethod def get_history(parent, nodeid, start_time, end_time): cursor conn.cursor() cursor.execute( SELECT timestamp, value FROM history WHERE nodeid? AND timestamp BETWEEN ? AND ? ORDER BY timestamp, (nodeid, start_time.isoformat(), end_time.isoformat()) ) return cursor.fetchall() server.link_method(history_method, get_history) try: while True: time.sleep(1) except KeyboardInterrupt: conn.close() server.stop()落地要点counter_var.set_value history_aware_setter是Python动态属性替换比继承重写更轻量所有写入均经此函数保证历史不丢。GetHistory方法返回的是元组列表UaExpert调用时会自动解析为表格实际项目中可扩展为分页、聚合AVG/MAX等。SQLite 文件opc_history.db位于脚本同目录无需额外服务适合边缘设备部署。4. OPC UA 安全配置实战从匿名登录到证书双向认证的平滑过渡“简单”不等于“不安全”。当Server从实验室移到产线匿名登录Anonymous必须升级为基于证书的认证否则会被安全审计一票否决。FreeOpcUa 支持X.509证书但配置路径与商业服务器差异较大——它不提供图形向导所有证书操作都在代码里完成反而更透明。4.1 生成自签名证书用 OpenSSL 三步创建可信CA与服务器证书# 1. 创建根证书颁发机构CA私钥和证书有效期10年 openssl genrsa -out ca_key.pem 2048 openssl req -x509 -new -nodes -key ca_key.pem -sha256 -days 3650 -out ca_cert.pem # 2. 创建服务器私钥和证书签名请求CSR openssl genrsa -out server_key.pem 2048 openssl req -new -key server_key.pem -out server_csr.pem # 3. 用CA签发服务器证书关键Subject Alternative Name必须含DNS名/IP cat ext.cnf EOF [req] distinguished_name req_distinguished_name x509_extensions v3_req prompt no [req_distinguished_name] C CN ST Beijing L Beijing O MyFactory OU Automation CN localhost [v3_req] basicConstraints CA:FALSE keyUsage nonRepudiation, digitalSignature, keyEncipherment subjectAltName alt_names [alt_names] DNS.1 localhost IP.1 127.0.0.1 IP.2 192.168.1.50 # 替换为你的服务器实际IP EOF openssl x509 -req -in server_csr.pem -CA ca_cert.pem -CAkey ca_key.pem -CAcreateserial -out server_cert.pem -days 3650 -sha256 -extfile ext.cnf -extensions v3_req注意subjectAltName是UA安全握手强制要求漏掉IP.1或DNS.1会导致UaExpert报错BadCertificateUseNotAllowed。生产环境必须填真实域名或IP不能只写localhost。4.2 在 Server 代码中加载证书并启用安全策略from opcua import Server, ua from opcua.common.security_policies import SecurityPolicy server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) # 加载证书与私钥路径需绝对或相对于脚本位置 server.load_certificate(server_cert.pem) server.load_private_key(server_key.pem) # 启用三种安全策略Basic256Sha256最常用 server.set_security_policy([ SecurityPolicy.SecurityPolicyBasic256Sha256, SecurityPolicy.SecurityPolicyBasic256, SecurityPolicy.SecurityPolicyBasic128Rsa15 ]) # 设置用户认证启用用户名密码同时保留证书认证 server.set_user_manager(user_manager_callback) def user_manager_callback(username, password): # 简单校验生产环境应查数据库或LDAP return username admin and password SecurePass123! # 启动服务器 server.start()参数说明set_security_policy()传入列表客户端连接时会协商最高共同支持的策略Basic256Sha256是当前推荐标准Basic128Rsa15为兼容老客户端保留。set_user_manager()是可选的但建议开启——证书认证解决“你是谁”用户名密码解决“你能做什么”双因子更稳妥。若只启用证书认证注释掉set_user_manager()并在UaExpert连接时选择“Certificate”而非“Username/Password”。4.3 UaExpert 配置证书信任链避免“证书不受信任”警告将ca_cert.pem复制到Windows证书存储双击ca_cert.pem→ “安装证书” → “本地计算机” → “受信任的根证书颁发机构”。在UaExpert中“Settings” → “Security” → “Trust List” → 点击“Add” → 选择ca_cert.pem。“Settings” → “Security” → “Issuer List” → 点击“Add” → 选择server_cert.pem此步非必须但可预加载。重新连接选择“Certificate”认证方式 → 点击“Connect” → 若弹出证书选择框选server_cert.pem对应的条目。提示若仍报错BadCertificateInvalid用openssl x509 -in server_cert.pem -text -noout检查Subject Alternative Name是否包含当前连接IP/DNS时间不同步也会导致证书验证失败确保服务器与客户端系统时间误差 5分钟。5. 避坑指南FreeOpcUa 开发中 4 个血泪经验总结这些坑都是在凌晨两点重启服务器十次后记下的。它们不写在任何文档里但踩中一个就能让你卡住一整天。5.1 现象UaExpert 显示“Connected”但地址空间树为空双击“Objects”无反应原因server.set_endpoint()的URL中协议头opc.tcp://后的IP地址写成了127.0.0.1而UaExpert从另一台机器连接导致OPC UA发现服务mDNS/SD无法将127.0.0.1解析为可路由地址。解决将server.set_endpoint(opc.tcp://127.0.0.1:4840/...)改为server.set_endpoint(opc.tcp://0.0.0.0:4840/...)监听所有接口或明确填写服务器实际IP如192.168.1.50。验证命令telnet 192.168.1.50 4840应能建立TCP连接。5.2 现象变量值在Server端set_value()成功但UaExpert里始终显示(BadWaitingForInitialData)原因变量节点未正确添加到地址空间的Objects文件夹下而是挂在了Types或Views下常见于复制粘贴代码时误写server.get_types_node()。UaExpert默认只展开Objects树。解决确认add_variable()的调用对象是server.get_objects_node()的返回值在UaExpert中点击“View” → “Address Space” → “Show All Nodes”搜索变量名若出现在Types/VariableTypes下说明挂载位置错误需修正代码。5.3 现象启用证书后UaExpert 连接时卡在“Verifying certificate...”最终超时原因服务器证书的Not Before时间晚于客户端当前系统时间证书尚未生效或Not After时间早于客户端时间证书已过期。FreeOpcUa 严格校验时间不接受误差。解决用openssl x509 -in server_cert.pem -dates -noout查看证书有效期同步服务器与客户端NTP时间若测试环境时间难同步生成证书时用-startdate参数指定早于当前时间如-startdate 20200101000000Z。5.4 现象Modbus读取正常但OPC变量值在UaExpert里跳变剧烈如 25.6 → 1000.0 → 0.0原因Modbus读取返回的result.registers是整数列表直接set_value(result.registers[0])会传入Int16而OPC变量定义为Float类型不匹配导致UA栈内部按浮点解释整数比特产生乱码值。解决显式类型转换例如set_value(float(result.registers[0]))更稳妥做法是定义变量时用ua.VariantType.Float明确类型var root.add_variable(ns2;i1, Temperature, ua.Variant(0.0, ua.VariantType.Float))6. 进阶技巧用 OPC UA 方法节点实现远程控制与诊断让Server不止于“数据管道”一个真正可用的OPC Server不该只是被动吐数据的管道。通过方法节点Method Node你可以把Python函数暴露给HMI或SCADA实现一键启停、参数下发、固件升级等操作。这比在HMI里写VB脚本或硬编码IP调用REST API 更符合工业语义。6.1 定义带输入输出参数的方法节点以“设备复位”为例from opcua import Server, ua import os import subprocess server Server() server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) server.set_server_name(ControlServer) server.start() root server.get_objects_node() # 定义方法节点ResetDevice输入参数为复位延时秒输出为执行结果字符串 reset_method root.add_method( ns2;sResetDevice, ResetDevice, [ua.VariantType.Int32], # 输入延时秒数 [ua.VariantType.String] # 输出成功/失败信息 ) uamethod def reset_device(parent, delay_seconds): 远程复位设备执行shell命令重启服务 parent: 调用该方法的节点此处为root delay_seconds: 输入参数复位前等待秒数 try: # 模拟设备复位前的安全检查如确认无运行任务 if delay_seconds 0 or delay_seconds 300: return fError: Delay must be between 0 and 300, got {delay_seconds} # 执行复位操作此处为重启某个服务实际项目替换为具体命令 result subprocess.run( [systemctl, restart, my-device-service], capture_outputTrue, textTrue, timeout30 ) if result.returncode 0: return fSuccess: Device restarted after {delay_seconds}s delay else: return fFailed: systemctl error {result.returncode} - {result.stderr[:100]} except subprocess.TimeoutExpired: return Error: Reset command timed out except Exception as e: return fException: {str(e)} # 将方法注册到服务器 server.link_method(reset_method, reset_device) # 保持服务器运行 try: while True: time.sleep(1) except KeyboardInterrupt: server.stop()关键设计点输入参数[ua.VariantType.Int32]和输出[ua.VariantType.String]必须与uamethod函数签名严格一致否则UaExpert调用时报BadArgumentsInvalid。subprocess.run()加timeout30防止命令卡死capture_outputTrue捕获日志用于返回结果避免后台进程失控。错误处理覆盖超时、权限拒绝、命令不存在等常见场景返回人类可读字符串方便HMI解析提示。6.2 在 UaExpert 中调用方法验证输入输出与错误反馈连接到Server后在地址空间树找到ResetDevice方法节点通常在Objects下。右键 → “Call Method” → 弹窗中Input Arguments输入框填10表示延时10秒。点击“OK”下方Output Arguments显示返回字符串如Success: Device restarted after 10s delay。故意输-5测试错误分支应返回Error: Delay must be between 0 and 300, got -5。提示UaExpert调用方法时输入参数类型由UI自动匹配填数字即Int32但若需传字符串数组需在UI中切换参数类型下拉框。6.3 扩展为诊断服务集成日志查询与健康检查将方法升级为诊断中心支持查询最近10条错误日志、检查磁盘空间、Ping关键设备方法名输入参数输出参数典型用途GetLastErrorLog无String日志行快速定位上次故障原因CheckDiskSpaceString路径如/opt/dataStructfree_gb,used_percent,status预判存储满风险PingDeviceStringIP地址Structsuccess: Boolean,latency_ms: Float验证网络连通性实现CheckDiskSpace示例uamethod def check_disk_space(parent, path): try: stat os.statvfs(path) free_gb (stat.f_bavail * stat.f_frsize) / (1024**3) total_gb (stat.f_blocks * stat.f_frsize) / (1024**3) used_percent (1 - stat.f_bavail / stat.f_blocks) * 100 status OK if free_gb 1.0 else WARNING: Low space return { free_gb: round(free_gb, 2), used_percent: round(used_percent, 1), status: status } except OSError as e: return {free_gb: 0.0, used_percent: 0.0, status: fERROR: {e}}这种结构让HMI开发人员无需关心Python只需按方法名和参数表调用Server端统一处理异常、日志、超时——这才是工业级“简单”的真谛表面接口极简背后鲁棒性拉满。我做这类桥接Server的规矩是第一版永远先跑通匿名明文变量第二版加Modbus/串口第三版上证书第四版加方法节点。每次只动一个模块日志打满用UaExpert逐帧验证。省下的调试时间够你喝三杯咖啡。希望帮到你。本文还有配套的精品资源点击获取