make class

client_socket.py

@@ -1,142 +1,195 @@
 import socket
 import json
 from pprint import pprint
+from tkinter import ttk
 
 from func import *
 
-s = socket.socket()
-host = MODBUS_SERVER_HOST
-port = 9760
 
-s.connect((host, port))
+class SocketRobotArm:
-# print(s)
+    line_speed = 100.0
-def send_data(data):    
+    line_smooth = 9
-    response = None
+    line_tool = 1
-    if data["reqType"] == 'AddRCC':
-        # return
-        pass
-    s.send(str.encode(json.dumps(data)))
-    response_data = s.recv(1024)
-    response = json.loads(response_data)
 
-    if data["reqType"] == 'query':
+    global_speed = 100
-        # pprint({"request":data["queryAddr"], "response":response["queryData"]})
+    physical_speed = 50
-        return response["queryData"]
+    # laser_id = 15
-    elif data["reqType"] == 'command':
+    laser_id = 14
-        # pprint({"request":data["cmdData"], "response":response["cmdReply"]})
+    # filename = 'half-sphere-no-angle'
-        return response["cmdReply"]
+    filename = "half-sphere"
-    elif data["reqType"] == 'AddRCC' and "cmdReply" in response.keys():
-        return response["cmdReply"]
-    else:
-        pprint(response)
 
-def make_addrcc_data(q, emptyList=0):
+    pass_size = 4
-    # print(q)
-    return {"dsID": "www.hc-system.com.HCRemoteCommand","reqType":"AddRCC","emptyList":str(emptyList),"packID":"0","instructions":q} 
-def make_command_data(q):
-    return {"dsID": "www.hc-system.com.RemoteMonitor","reqType":"command","packID":"0","cmdData":q}
-def make_query_data(q):
-    return {"dsID": "www.hc-system.com.RemoteMonitor","reqType":"query","packID":"0","queryAddr":q}
 
-axis_coord_raw = send_data(make_query_data(['axis-0','axis-1','axis-2','axis-3','axis-4','axis-5']))
+    def __init__(self, *args, **kwargs):
-axis_coord = []
+        self.socket = socket.socket()
-for i in axis_coord_raw:
+        self.host = MODBUS_SERVER_HOST
-    axis_coord.append(float(i))
+        self.port = 9760
 
-world_coord_raw = send_data(make_query_data(['world-0','world-1','world-2','world-3','world-4','world-5']))
+        self.host = "127.0.0.1"
-world_coord = []
+        self.port = 65432
-for i in world_coord_raw:
-    world_coord.append(float(i))
-x,y,z,u,v,w = world_coord
 
-print('axis', axis_coord)
+        self.connect()
-print('world',  world_coord)
-print('remote command count', send_data(make_query_data(['RemoteCmdLen'])))
 
-line_speed = 100.0
+    def __exit__(self, exc_type, exc_value, traceback):
-line_smooth = 9
+        self.close()
-line_tool = 1
 
-global_speed = 100
+    def connect(self):
-physical_speed = 50
+        self.socket.connect((self.host, self.port))
-# laser_id = 15
-laser_id = 14
-# filename = 'half-sphere-no-angle'
-filename = 'half-sphere'
 
-pass_size = 4
+        self.get_axis()
+        self.get_world()
+        self.get_command_count()
 
-def make_world_step(type, p):
+        self.send_data(self.set_global_speed())
-    step = {
-        "oneshot": "0",
-        "delay":"0.0",
-        "speed":str(line_speed),"smooth":str(line_smooth),
-        "coord":"0","tool":str(line_tool),
-        "ckStatus":"0x3F",
-    }
-    if type == 'line':
-        m0,m1,m2,m3,m4,m5=p
-        step.update({
-            "action": "10",
-            "m0": round(x + m0,3), "m1": round(y + m1,3), "m2": round(z + m2,3),
-            "m3": round(u + m3,3), "m4": round(v + m4,3), "m5": round(w + m5,3),
-            "m6": 0, "m7": 0,
-        })
-    elif type == 'curve':
-        m0,m1,m2,m3,m4,m5, m0_p, m1_p,m2_p,m3_p,m4_p,m5_p=p
-        step.update({
-            "action": "17",
-            "m0": round(x + m0,3), "m1": round(y + m1,3), "m2": round(z + m2,3),
-            "m3": round(u + m3,3), "m4": round(v + m4,3), "m5": round(w + m5,3),
-            "m6": 0, "m7": 0,
-            "m0_p": round(x + m0_p,3), "m1_p": round(y + m1_p,3), "m2_p": round(z + m2_p,3),
-            "m3_p": round(u + m3_p,3), "m4_p": round(v + m4_p,3), "m5_p": round(w + m3_p,3),
-            "m6_p": 0, "m7_p": 0,
-        })
 
-    for s in step:
+        self.add_rcc_list = (
-        step[s] = str(step[s])
+            [self.set_physical_speed(True), self.set_output_laser(True)]
-    return step
+            + self.steps_from_file()
+            + [self.set_physical_speed(False), self.set_output_laser(False)]
+        )
+        step = 2
+        for i in range(0, len(self.add_rcc_list), step):
+            print(self.add_rcc_list[i : i + step])
+            self.send_data(
+                make_addrcc_data(self.add_rcc_list[i : i + step], int(not i))
+            )
 
-result = []
+        self.send_data(self.start_cycle())
-with open(f'data/{filename}.nc.result', 'r') as fp:
+
-    for line in fp:
+    def send_data(self, data):
-        data = line.strip().split(' ')
+        response = None
-        prep = {}
+        # if data["reqType"] == "AddRCC":
-        for item in data:
+        #     return
-            prep[item[:1]] = float(item[1:])
+        self.socket.send(str.encode(json.dumps(data)))
-        result.append(
+        response_data = self.socket.recv(1024)
-            make_world_step(
+        response = json.loads(response_data)
-                'line', 
+
-                (
+        if data["reqType"] == "query":
-                    prep.get('X', 0), prep.get('Y', 0), prep.get('Z', 0), 
+            return response["queryData"]
-                    prep.get('U', 0), prep.get('V', 0), prep.get('W', 0)
+        elif data["reqType"] == "command":
-                )
+            return response["cmdReply"]
+        elif data["reqType"] == "AddRCC" and "cmdReply" in response.keys():
+            return response["cmdReply"]
+        else:
+            pprint(response)
+
+    def get_axis(self):
+        axis_coord_raw = self.send_data(
+            make_query_data(
+                ["axis-0", "axis-1", "axis-2", "axis-3", "axis-4", "axis-5"]
             )
         )
+        self.start_axis_coordinates = [float(i) for i in axis_coord_raw]
+        print("start_axis_coordinates", self.start_axis_coordinates)
 
-# Изменили глобальную скорость на global_speed%
+    def get_world(self):
-send_data(make_command_data(['modifyGSPD', str(global_speed  * 10)]))
+        world_coord_raw = self.send_data(
-send_data(make_addrcc_data([
+            make_query_data(
-    # Включили физическую скорость
+                ["world-0", "world-1", "world-2", "world-3", "world-4", "world-5"]
-    {"oneshot":"0", "action":"51","isUse":"1","speed":str(physical_speed*1000)},
+            )
-    # Поставили Y026 (laser id + 11) в TRUE -- ЛАЗЕР!!!!
+        )
-    {"oneshot":"0", "action":"200","type":"0","io_status":"1", "point":str(laser_id)},
+        self.start_world_coordinates = [float(i) for i in world_coord_raw]
-], 1))
+        print("start_world_coordinates", self.start_world_coordinates)
 
-res_list = len(result) + 1
+    def get_command_count(self):
-if res_list > 300:
+        res = self.send_data(make_query_data(["RemoteCmdLen"]))
-    res_list = 300
+        self.remote_command_count = res
+        print(res)
 
-for i in range(pass_size, res_list, pass_size):
+    def set_global_speed(self):
-    print(f"Отправили на устройство команды {i} ... {i + 1}")
+        # Изменили глобальную скорость на global_speed%
-    send_data(make_addrcc_data(result[i:i+1]))
+        return make_command_data(["modifyGSPD", str(self.global_speed * 10)])
 
-send_data(make_addrcc_data([
+    def start_cycle(self):
-    {"oneshot":"0", "action":"51","isUse":"0"},
+        return make_command_data(["actionSingleCycle"])
-    {"oneshot":"0", "action":"200","type":"0","io_status":"0", "point":str(laser_id)},
-]))
 
-print('remote command count', send_data(make_query_data(['RemoteCmdLen'])))
+    def round(v):
-send_data(make_command_data(['actionSingleCycle']))
+        return round(v, 3)
-# send_data(make_command_data(['actionStop']))
+
+    def make_world_step(self, type, point):
+        step = {
+            "oneshot": "0",
+            "delay": "0.0",
+            "speed": str(self.line_speed),
+            "smooth": str(self.line_smooth),
+            "coord": "0",
+            "tool": str(self.line_tool),
+            "ckStatus": "0x3F",
+        }
+        if type == "line":
+            pairs = zip(self.start_world_coordinates, point)
+            m0, m1, m2, m3, m4, m5 = [round(sum(i), 3) for i in pairs]
+            step.update({"action": "10"})
+            step.update({"m0": m0, "m1": m1, "m2": m2, "m3": m3, "m4": m4, "m5": m5})
+            step.update({"m6": 0, "m7": 0})
+        elif type == "curve":
+            pairs = zip(self.start_world_coordinates, point[:5])
+            m0, m1, m2, m3, m4, m5 = [round(sum(i), 3) for i in pairs]
+
+            pairs_p = zip(self.start_world_coordinates, point[6:])
+            m0_p, m1_p, m2_p, m3_p, m4_p, m5_p = [round(sum(i), 3) for i in pairs_p]
+            step.update({"action": "17"})
+            step.update({"m0": m0, "m1": m1, "m2": m2, "m3": m3, "m4": m4, "m5": m5})
+            step.update({"m6": 0, "m7": 0})
+            step.update(
+                {
+                    "m0_p": m0_p,
+                    "m1_p": m1_p,
+                    "m2_p": m2_p,
+                    "m3_p": m3_p,
+                    "m4_p": m4_p,
+                    "m5_p": m5_p,
+                }
+            )
+            step.update({"m6_p": 0, "m7_p": 0})
+
+        for s in step:
+            step[s] = str(step[s])
+        return step
+
+    def set_physical_speed(self, status: bool = False):
+        return (
+            {
+                "oneshot": "0",
+                "action": "51",
+                "isUse": str(int(status)),
+                "speed": str(self.physical_speed * 1000),
+            },
+        )
+
+    def set_output_laser(self, status: bool = False):
+        return (
+            {
+                "oneshot": "0",
+                "action": "200",
+                "type": "0",
+                "io_status": str(int(status)),
+                "point": str(self.laser_id),
+            },
+        )
+
+    def steps_from_file(self):
+        result = []
+        with open(f"data/{self.filename}.nc.result", "r") as fp:
+            for line in fp:
+                data = line.strip().split(" ")
+                prep = {}
+                for item in data:
+                    prep[item[:1]] = float(item[1:])
+                result.append(
+                    self.make_world_step(
+                        "line",
+                        (
+                            prep.get("X", 0),
+                            prep.get("Y", 0),
+                            prep.get("Z", 0),
+                            prep.get("U", 0),
+                            prep.get("V", 0),
+                            prep.get("W", 0),
+                        ),
+                    )
+                )
+        return result
+
+
+SocketRobotArm()

func.py

@@ -113,3 +113,11 @@ def stop_immediately(client):
     
 def multiply_1000(n):
     return int(n * 1000)
+
+def make_addrcc_data(q, emptyList=0):
+    # print(q)
+    return {"dsID": "www.hc-system.com.HCRemoteCommand","reqType":"AddRCC","emptyList":str(emptyList),"packID":"0","instructions":q} 
+def make_command_data(q):
+    return {"dsID": "www.hc-system.com.RemoteMonitor","reqType":"command","packID":"0","cmdData":q}
+def make_query_data(q):
+    return {"dsID": "www.hc-system.com.RemoteMonitor","reqType":"query","packID":"0","queryAddr":q}

test_socket_server.py

@@ -0,0 +1,27 @@
+import socket
+import json
+
+HOST = "127.0.0.1"  # Standard loopback interface address (localhost)
+PORT = 65432  # Port to listen on (non-privileged ports are > 1023)
+
+with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+    s.bind((HOST, PORT))
+    s.listen()
+    conn, addr = s.accept()
+    with conn:
+        print(f"Connected by {addr}")
+        while True:
+            data = conn.recv(1024)
+            if data:
+                req = json.loads(data)
+                res = {"queryData": ["ok"]}
+
+                if "queryAddr" in req.keys() and "axis-0" in req["queryAddr"]:
+                    res["queryData"] = [10, 11, 12, 13, 14, 15]
+
+                if "queryAddr" in req and "world-0" in req["queryAddr"]:
+                    res["queryData"] = [100, 101, 102, 103, 104, 105]
+
+                if req["reqType"] == "command":
+                    res["cmdReply"] = ['ok']
+                conn.sendall(json.dumps(res).encode())