#!/usr/bin/env python3 """ 简单的OPC UA服务器示例 用于工业自动化数据通信 """ from opcua import Server, ua import time import random import threading from datetime import datetime # from core.system import SystemState # from config.ini_manager import ini_manager class SimpleOPCUAServer: # def __init__(self, state, endpoint=ini_manager.opcua_endpoint, name="Feed_Server"): def __init__(self, endpoint='', name="Feed_Server"): """ 初始化OPC UA服务器 Args: endpoint: 服务器端点地址 name: 服务器名称 """ self.server = Server() self.server.set_endpoint(endpoint) self.server.set_server_name(name) # self.state = state # 设置服务器命名空间 self.namespace = self.server.register_namespace("Feed_Control_System") # 获取对象节点 self.objects = self.server.get_objects_node() # 创建自定义对象 self.create_object_structure() # 运行标志 self.running = False # 订阅和监控项 self.subscription = None self.monitored_items = [] # 记录上次值用于检测变化 self._last_values = {} def create_object_structure(self): """创建OPC UA对象结构""" # 创建上料斗对象 self.upper = self.objects.add_object(self.namespace, "upper") self.lower=self.objects.add_object(self.namespace, "lower") self.sys=self.objects.add_object(self.namespace, "sys") self.mould=self.objects.add_object(self.namespace, "mould") self.pd=self.objects.add_object(self.namespace, "pd") # 创建变量 self.create_variables() def create_variables(self): """创建OPC UA变量""" # 创建变量时显式指定数据类型和初始值 #上料斗 self.upper_weight = self.upper.add_variable(self.namespace, "upper_weight", ua.Variant(0.0, ua.VariantType.Float)) self.upper_is_arch = self.upper.add_variable(self.namespace, "upper_is_arch", ua.Variant(False, ua.VariantType.Boolean)) self.upper_door_closed = self.upper.add_variable(self.namespace, "upper_door_closed", ua.Variant(False, ua.VariantType.Boolean)) self.upper_volume = self.upper.add_variable(self.namespace, "upper_volume", ua.Variant(0.0, ua.VariantType.Float)) self.upper_door_position = self.upper.add_variable(self.namespace, "upper_door_position", ua.Variant(0, ua.VariantType.Int16)) #下料斗 self.lower_weight = self.lower.add_variable(self.namespace, "lower_weight", ua.Variant(0.0, ua.VariantType.Float)) self.lower_is_arch = self.lower.add_variable(self.namespace, "lower_is_arch", ua.Variant(False, ua.VariantType.Boolean)) #模具车 self.mould_finish_weight = self.mould.add_variable(self.namespace, "mould_finish_weight", ua.Variant(0.0, ua.VariantType.Float)) self.mould_need_weight = self.mould.add_variable(self.namespace, "mould_need_weight", ua.Variant(0.0, ua.VariantType.Float)) self.mould_frequency = self.mould.add_variable(self.namespace, "mould_frequency", ua.Variant(230, ua.VariantType.Int32)) self.mould_vibrate_status = self.mould.add_variable(self.namespace, "mould_vibrate_status", ua.Variant(False, ua.VariantType.Boolean)) self.feed_status = self.mould.add_variable(self.namespace, "feed_status", ua.Variant(0, ua.VariantType.Int16)) self.pd_data=self.pd.add_variable(self.namespace, "pd_data", ua.Variant("", ua.VariantType.String)) # 在创建变量后立即设置可写权限（不需要等待服务器启动） self.upper_weight.set_writable(True) self.lower_weight.set_writable(True) self.upper_is_arch.set_writable(True) self.upper_door_closed.set_writable(True) self.upper_volume.set_writable(True) self.upper_door_position.set_writable(True) self.lower_is_arch.set_writable(True) self.mould_finish_weight.set_writable(True) self.mould_need_weight.set_writable(True) self.mould_frequency.set_writable(True) self.mould_vibrate_status.set_writable(True) self.feed_status.set_writable(True) self.pd_data.set_writable(True) print("[变量创建] 变量创建完成，AccessLevel权限已设置") # 验证并打印当前的AccessLevel属性 # try: # al = self.upper_weight.get_attribute(ua.AttributeIds.AccessLevel) # ual = self.upper_weight.get_attribute(ua.AttributeIds.UserAccessLevel) # print(f"[变量创建] upper_weight AccessLevel: {al.Value.Value}, UserAccessLevel: {ual.Value.Value}") # al2 = self.lower_weight.get_attribute(ua.AttributeIds.AccessLevel) # ual2 = self.lower_weight.get_attribute(ua.AttributeIds.UserAccessLevel) # print(f"[变量创建] lower_weight AccessLevel: {al2.Value.Value}, UserAccessLevel: {ual2.Value.Value}") # except Exception as e: # print(f"[变量创建] 获取权限属性失败: {e}") def setup_variable_permissions(self): """设置变量权限 - 在服务器启动后调用""" try: # 重新设置变量为可写，确保权限生效 self.upper_weight.set_writable(True) self.lower_weight.set_writable(True) print("[权限设置] 变量权限已重新设置") # 验证权限 try: al = self.upper_weight.get_attribute(ua.AttributeIds.AccessLevel) ual = self.upper_weight.get_attribute(ua.AttributeIds.UserAccessLevel) print(f"[权限设置] upper_weight AccessLevel: {al.Value.Value}, UserAccessLevel: {ual.Value.Value}") except Exception as e: print(f"[权限设置] 验证失败: {e}") except Exception as e: print(f"[权限设置] 设置权限失败: {e}") print("[权限设置] 尝试强制设置...") # def setup_state_listeners(self): # """设置状态监听器 - 事件驱动更新""" # if hasattr(self.state, 'state_updated'): # self.state.state_updated.connect(self.on_state_changed) # print("状态监听器已设置 - 事件驱动模式") # def on_state_changed(self, property_name, value): # """状态变化时的回调函数""" # try: # # 根据属性名更新对应的OPC UA变量 # if property_name == "upper_weight": # self.upper_weight.set_value(value) # elif property_name == "lower_weight": # self.lower_weight.set_value(value) # # 可以在这里添加更多状态映射 # print(f"状态更新: {property_name} = {value}") # except Exception as e: # print(f"状态更新错误: {e}") def start(self): """启动服务器""" try: self.server.start() self.running = True print(f"服务器端点: opc.tcp://0.0.0.0:4840/freeopcua/server/") print("=" * 60) # 【关键修复】在设置监听器之前，先设置变量权限 # 这确保 AccessLevel 属性在客户端写入前已正确设置 # self.setup_variable_permissions() print("=" * 60) # 设置客户端写入监听器 # self.setup_write_listeners() print("=" * 60) # 初始化当前值 # if self.state: # self.upper_weight.set_value(self.state._upper_weight) # self.lower_weight.set_value(self.state._lower_weight) # print("已同步初始状态值") # 设置状态监听器 - 关键步骤！ # self.setup_state_listeners() # # 只有在没有状态系统时才使用模拟线程 # if not self.state: # print("使用模拟数据模式") # self.simulation_thread = threading.Thread(target=self.simulate_data) # self.simulation_thread.daemon = True # self.simulation_thread.start() except Exception as e: print(f"启动服务器失败: {e}") def stop(self): """停止服务器""" # 移除监听器 # self.remove_write_listeners() self.running = False self.server.stop() print("OPC UA服务器已停止") # 断开状态监听器 # if hasattr(self.state, 'state_updated'): # try: # self.state.state_updated.disconnect(self.on_state_changed) # except: # pass def main(): """主函数""" # 创建系统状态实例 # state = SystemState() # 创建并启动服务器 server = SimpleOPCUAServer( endpoint="opc.tcp://0.0.0.0:4840/freeopcua/server/", name="工业自动化 OPC UA 服务器" ) try: server.start() print("服务器正在运行，按 Ctrl+C 停止...") # 保持服务器运行 while True: time.sleep(1) except KeyboardInterrupt: print("\n正在停止服务器...") server.stop() except Exception as e: print(f"服务器运行错误: {e}") server.stop() if __name__ == "__main__": main()