DCS_V1.0/protocol/RPC/RpcClient.py

import pika
import uuid
import json
import threading
import socket

class RpcClient:
    def __init__(self, clientName, rabbitHost='localhost', protocolManager=None):
        """
        初始化RPC客户端
        :param clientName: 客户端名称
        :param rabbitHost: RabbitMQ服务器地址
        :param protocolManager: 协议管理器实例，用于处理读写操作
        """
        self.clientName = clientName
        self.protocolManager = protocolManager
        self.rabbitHost = rabbitHost
        self.variables = {}
        # self.variables = {
        #     "temp": {"type": "AI", "min": 0, "max": 100, "value": 25.5},
        #     "status": {"type": DO", "min": None, "max": None, "value": "OK"},
        # }
        self.credentials = pika.PlainCredentials('dcs', '123456')
        self.connection = pika.BlockingConnection(
                pika.ConnectionParameters(host=rabbitHost, credentials=self.credentials)
            )
        self.channel = self.connection.channel()
        self.queueName = f"rpc_queue_{self.clientName}"
        self.channel.queue_declare(queue=self.queueName)
        # print(f"[{self.clientName}] 等待服务端指令...")

    def onRequest(self, ch, method, props, body):
        request = json.loads(body)
        cmd = request.get("cmd")
        response = {}
        
        if cmd == "ping":
            # 响应ping命令，带本机IP
            ip = self.getLocalIp()
            response = {"client": self.clientName, "result": "pong", "ip": ip}
        elif cmd == "read":
            response = {"client": self.clientName, "variables": self.variables}
        elif cmd == "write":
            # 使用协议管理器处理写入操作
            response = self.handleWriteRequest(request)
        else:
            # 未知命令的响应
            response = {"client": self.clientName, "result": "fail", "reason": "unknown command"}
        
        ch.basic_publish(
            exchange='',
            routing_key=props.reply_to,
            properties=pika.BasicProperties(correlation_id=props.correlation_id),
            body=json.dumps(response)
        )
        ch.basic_ack(delivery_tag=method.delivery_tag)
    

    
    def handleWriteRequest(self, request):
        """
        处理写入请求
        :param request: 请求数据
        :return: 响应数据
        """
        varName = request.get("varName")
        value = request.get("value")
        
        # 如果有协议管理器，使用它来处理写入
        if self.protocolManager:
            try:
                success = self.protocolManager.writeVariableValue(varName, value)
                if success:
                    return {
                        "client": self.clientName, 
                        "result": "success",
                        "varName": varName,
                        "value": value
                    }
                else:
                    return {
                        "client": self.clientName, 
                        "result": "fail", 
                        "reason": "write failed"
                    }
            except Exception as e:
                return {
                    "client": self.clientName, 
                    "result": "fail", 
                    "reason": f"write error: {str(e)}"
                }
    
    def setVarContent(self, variableName, value, min, max, type):
        if type in ['DI', 'DO']:
            min = 0
            max = 1
        variableName = variableName + '.' + self.clientName
        self.variables[variableName]["value"] = value + '.' + self.clientName
        self.variables[variableName]["min"] = min
        self.variables[variableName]["max"] = max
        self.variables[variableName]["type"] = type


    def start(self):
        """启动客户端监听（阻塞方法）"""
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(queue=self.queueName, on_message_callback=self.onRequest)
        self.channel.start_consuming()
    
    def startNonBlocking(self):
        """非阻塞启动客户端（在后台线程中运行）"""
        import threading
        self.client_thread = threading.Thread(target=self.start, daemon=True)
        self.client_thread.start()
        print(f"[{self.clientName}] RPC客户端已在后台线程启动")
    
    def isRunning(self):
        """检查客户端是否正在运行"""
        return hasattr(self, 'client_thread') and self.client_thread.is_alive()
    
    def close(self):
        """关闭RPC连接"""
        try:
            if self.channel and not self.channel.is_closed:
                self.channel.close()
            if self.connection and not self.connection.is_closed:
                self.connection.close()
            print(f"[{self.clientName}] 连接已关闭")
        except Exception as e:
            print(f"[{self.clientName}] 关闭连接时出错: {e}")

    def getLocalIp(self):
        # 获取本机第一个非回环IPv4地址
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(('8.8.8.8', 80))
            ip = s.getsockname()[0]
            s.close()
            return ip
        except:
            return '127.0.0.1'

if __name__ == "__main__":
    import sys
    clientName = sys.argv[1] if len(sys.argv) > 1 else "Client1"
    client = RpcClient(clientName)
    client.start()