ksenia_mikhailova
/
modbus_test
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

format

This commit is contained in:
Kseninia Mikhaylova 2024-10-01 12:32:20 +03:00
parent 3bea136c0c
commit b58db72f74
4 changed files with 73 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
client_socket.py
View File

@ -25,66 +25,78 @@ class SocketRobotArm:
self.socket = socket.socket()
self.host = MODBUS_SERVER_HOST
self.port = 9760
if len(sys.argv) and "--test" in sys.argv:
self.host = "127.0.0.1"
self.port = 65432
self.tkinter_root = tkinter.Tk()
self.tkinter_info_label = tkinter.Label(self.tkinter_root)
self.tkinter_exit = tkinter.Button(
self.tkinter_root, text="Exit", command=self.close
)
self.connect()
self.socket.connect((self.host, self.port))
self.cycle_base()
self.cycle_start()
def __exit__(self, exc_type, exc_value, traceback):
print("exiting")
self.socket.close()
def close(self):
print("close")
self.tkinter_info_label.destroy()
# self.tkinter_root.destroy()
self.socket.close()
self.tkinter_root.destroy()
sys.exit()
def set_text(self, text):
label = tkinter.Label(self.tkinter_root, text=text)
label.pack()
self.tkinter_root.update()
def connect(self):
self.socket.connect((self.host, self.port))
self.socket = None
# sys.exit()
def cycle_base(self):
self.tkinter_root.geometry("500x300")
self.tkinter_exit.pack(pady=20)
self.get_axis()
self.set_text(text=f"Координаты осей {self.start_axis_coordinates}")
time.sleep(0.5)
self.get_world()
self.set_text(text=f"Мировые координаты {self.start_world_coordinates}")
time.sleep(0.5)
self.get_command_count()
self.set_text(text=f"Команд в очереди {self.remote_command_count}")
time.sleep(0.5)
self.send_data(self.set_global_speed())
self.set_text(text=f"Установили глобальную скорость {self.global_speed}")
time.sleep(0.5)
def cycle_start(self):
self.send_data(self.start_cycle())
self.set_text(text=f"Старт одиночного цикла")
time.sleep(0.5)
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(self.tkinter_root, text=f"Отправка данных")
my_label.pack()
self.tkinter_info_label.config(text=f"Отправка данных...")
self.tkinter_info_label.pack()
self.tkinter_root.update()
exit_button = tkinter.Button(self.tkinter_root, text="Exit", command=self.close)
exit_button.pack(pady=20)
step = 2
for i in range(0, len(self.add_rcc_list), step):
if not self.tkinter_info_label:
return
if not self.socket:
return
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}")
self.tkinter_info_label.config(text=f"Отправка данных {i}...{i+step-1}")
self.tkinter_root.update()
self.send_data(
@ -92,10 +104,15 @@ class SocketRobotArm:
)
time.sleep(0.5)
def set_text(self, text):
label = tkinter.Label(self.tkinter_root, text=text)
label.pack()
self.tkinter_root.update()
def send_data(self, data):
response = None
# if data["reqType"] == "AddRCC":
# return
if not self.socket:
return
self.socket.send(str.encode(json.dumps(data)))
response_data = self.socket.recv(1024)
response = json.loads(response_data)

18
poetry.lock generated
View File

@ -158,6 +158,22 @@ files = [
{file = "numpy-2.0.1.tar.gz", hash = "sha256:485b87235796410c3519a699cfe1faab097e509e90ebb05dcd098db2ae87e7b3"},
]
[[package]]
name = "pybullet"
version = "3.2.6"
description = "Official Python Interface for the Bullet Physics SDK specialized for Robotics Simulation and Reinforcement Learning"
optional = false
python-versions = "*"
files = [
{file = "pybullet-3.2.6-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6f71d0bed6824875b521066571d612274f1d2dead3ede50f8fbdde1946ec0c40"},
{file = "pybullet-3.2.6-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a476d51726efa7aabe1dc8c4ee676b78bcd9fc19d4a2ce9776f2c9a9dacd9e8b"},
{file = "pybullet-3.2.6-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8af791050e927f407a85fa9de15754fbe0973f8c798fd2893773b52fa1fc70e0"},
{file = "pybullet-3.2.6-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:72be1e68c61bbe49af5b7c609d59ede92d3fbbfca072e9b31d4819575c8f332b"},
{file = "pybullet-3.2.6-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ba46f9d8ae76d2bcae4bf4b5768b1dccc0b3b15110b69ba3d78cd0b11f8f21cf"},
{file = "pybullet-3.2.6-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f928444471186c5d6092d71371b03436e23ec96acd7800dfefcd8e96572c6581"},
{file = "pybullet-3.2.6.tar.gz", hash = "sha256:da27525433c88698dc9fd8bc20fa4ae4d07738b4656633659ebd82c2d2884e08"},
]
[[package]]
name = "pymodbus"
version = "3.7.0"
@ -468,4 +484,4 @@ testing = ["coverage (>=5.0.3)", "zope.event", "zope.testing"]
[metadata]
lock-version = "2.0"
python-versions = "3.10.12"
content-hash = "a224728a18ef19c1f754e77db7d8cba413724a9a4257af5704b0fad93f833c58"
content-hash = "5d6ac608b7d4cd83ce992399f2263bf6356902a177e910b7419966ec5eb20e3f"

1
pyproject.toml
View File

@ -13,6 +13,7 @@ pymodbus = "^3.7.0"
pymodbus3 = "^1.0.0"
svgpathtools = "^1.6.1"
tk = "^0.1.0"
pybullet = "^3.2.6"
[build-system]

25
test_socket_server.py
View File

@ -11,17 +11,20 @@ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
with conn:
print(f"Connected by {addr}")
while True:
data = conn.recv(1024)
if data:
req = json.loads(data)
res = {"queryData": ["ok"]}
try:
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.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 "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())
if req["reqType"] == "command":
res["cmdReply"] = ['ok']
conn.sendall(json.dumps(res).encode())
except Exception as e:
print(f"by {addr} error {e}")