220 lines
6.9 KiB
Python
220 lines
6.9 KiB
Python
import socket
|
|
import json
|
|
from pprint import pprint
|
|
import tkinter
|
|
|
|
from func import *
|
|
|
|
|
|
class SocketRobotArm:
|
|
line_speed = 100.0
|
|
line_smooth = 9
|
|
line_tool = 1
|
|
|
|
global_speed = 100
|
|
physical_speed = 50
|
|
# laser_id = 15
|
|
laser_id = 14
|
|
# filename = 'half-sphere-no-angle'
|
|
filename = "half-sphere"
|
|
|
|
pass_size = 4
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.socket = socket.socket()
|
|
self.host = MODBUS_SERVER_HOST
|
|
self.port = 9760
|
|
|
|
self.host = "127.0.0.1"
|
|
self.port = 65432
|
|
|
|
self.connect()
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
self.close()
|
|
|
|
def update_text(self, m, text):
|
|
label = tkinter.Label(m, text=text)
|
|
label.pack()
|
|
m.update()
|
|
|
|
def connect(self):
|
|
self.socket.connect((self.host, self.port))
|
|
|
|
m = tkinter.Tk()
|
|
m.geometry("500x500")
|
|
|
|
self.get_axis()
|
|
self.update_text(m, text=f"Координаты осей {self.start_axis_coordinates}")
|
|
|
|
self.get_world()
|
|
self.update_text(m, text=f"Мировые координаты {self.start_world_coordinates}")
|
|
|
|
self.get_command_count()
|
|
self.update_text(m, text=f"Команд в очереди {self.remote_command_count}")
|
|
|
|
self.send_data(self.set_global_speed())
|
|
self.update_text(m, text=f"Установили глобальную скорость {self.global_speed}")
|
|
|
|
self.send_data(self.start_cycle())
|
|
self.update_text(m, text=f"Старт одиночного цикла")
|
|
|
|
self.add_rcc_list = (
|
|
[self.set_physical_speed(True), self.set_output_laser(True)]
|
|
+ self.steps_from_file()
|
|
+ [self.set_physical_speed(False), self.set_output_laser(False)]
|
|
)
|
|
|
|
my_label = tkinter.Label(m, text=f"Отправка данных")
|
|
my_label.pack()
|
|
m.update()
|
|
step = 2
|
|
for i in range(0, len(self.add_rcc_list), step):
|
|
print(f"Отправка данных {i}...{i+step-1}")
|
|
# self.update_text(m, text=f"Отправка данных {i}...{i+step-1}")
|
|
my_label.config(text=f"Отправка данных {i}...{i+step-1}")
|
|
m.update()
|
|
|
|
self.send_data(
|
|
make_addrcc_data(self.add_rcc_list[i : i + step], int(not i))
|
|
)
|
|
time.sleep(0.5)
|
|
|
|
def send_data(self, data):
|
|
response = None
|
|
# if data["reqType"] == "AddRCC":
|
|
# return
|
|
self.socket.send(str.encode(json.dumps(data)))
|
|
response_data = self.socket.recv(1024)
|
|
response = json.loads(response_data)
|
|
|
|
if data["reqType"] == "query":
|
|
return response["queryData"]
|
|
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)
|
|
|
|
def get_world(self):
|
|
world_coord_raw = self.send_data(
|
|
make_query_data(
|
|
["world-0", "world-1", "world-2", "world-3", "world-4", "world-5"]
|
|
)
|
|
)
|
|
self.start_world_coordinates = [float(i) for i in world_coord_raw]
|
|
print("start_world_coordinates", self.start_world_coordinates)
|
|
|
|
def get_command_count(self):
|
|
res = self.send_data(make_query_data(["RemoteCmdLen"]))
|
|
self.remote_command_count = res
|
|
print(res)
|
|
|
|
def set_global_speed(self):
|
|
# Изменили глобальную скорость на global_speed%
|
|
return make_command_data(["modifyGSPD", str(self.global_speed * 10)])
|
|
|
|
def start_cycle(self):
|
|
return make_command_data(["actionSingleCycle"])
|
|
|
|
def round(v):
|
|
return round(v, 3)
|
|
|
|
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()
|