使用示例

Warning

由于版本更新，以下示例中的RPC接口可能已经废弃，使用时请参考最新的RPC接口

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
# 测试RPC接口
import grpc
import robot_pb2
import robot_pb2_grpc


def print_resp_data(rpc_resp):
    """
    把所有RPC返回值打印
    """
    if hasattr(rpc_resp, "data"):
        print(rpc_resp.data)
    if hasattr(rpc_resp, "ret"):
        print(rpc_resp.ret)
    if hasattr(rpc_resp, "data1"):
        print(rpc_resp.data1)
    if hasattr(rpc_resp, "data2"):
        print(rpc_resp.data2)


def getPosAndPose():
    """
    TODO 暂未实现
    获取当前位姿
    """
    resp = stub.getPosAndPose(robot_pb2.ReqInt(req=0))
    print_resp_data(resp.data)


def MoveA():
    """
    MoveA
    """
    resp = stub.MoveA(robot_pb2.ReqMoveA(
        rjoint=joint,
        rspeed=speed
    ))
    print_resp_data(resp)


def MoveJ():
    """
    MoveJ
    """
    resp = stub.MoveJ(robot_pb2.ReqMoveJ(
        rpose=pose,
        rspeed=speed,
        rzone=zone,
        rtool=tool,
        wobj=wobj
    ))
    print_resp_data(resp)


def TGetRobotNum():
    """
    获取机器人个数
    """
    resp = stub.TGetRobotNum(robot_pb2.ReqVoid())
    print_resp_data(resp)


def TStreamRobotJoint():
    """
    流式传输，获取机械臂关节
    """
    resp = stub.TStreamRobotJoint(robot_pb2.ReqVoid())
    for r in resp:
        print_resp_data(r)


def TStreamRobotCartesian():
    """
    """
    resp = stub.TStreamRobotCartesian(robot_pb2.ReqVoid())
    for r in resp:
        print_resp_data(r)


def TRobotHome():
    """
    机械臂回零点位置
    """
    resp = stub.TRobotHome(robot_pb2.ReqVoid())
    print_resp_data(resp)


def ConnectRPC():
    """
    测试RPC服务状态
    """
    resp = stub.ConnectRPC(robot_pb2.ReqVoid())
    print_resp_data(resp)


def MoveStart():
    """
    机械臂上电
    """
    resp = stub.MoveStart(robot_pb2.ReqVoid())
    print_resp_data(resp)


def MoveStop():
    """
    机械臂下电
    """
    resp = stub.MoveStop(robot_pb2.ReqVoid())
    print_resp_data(resp)


def MoveL():
    """
    MoveL
    """
    resp = stub.MoveL(robot_pb2.ReqMoveL(rpose=pose,
                                         rspeed=speed,
                                         rzone=zone,
                                         rtool=tool,
                                         wobj=wobj))
    print_resp_data(resp)


def MoveC():
    """
    MoveC
    """
    resp = stub.MoveC(robot_pb2.ReqMoveC(rpose=pose,
                                         rpose_mid=pose2,
                                         rspeed=speed,
                                         rzone=zone,
                                         rtool=tool,
                                         wobj=wobj))
    print_resp_data(resp)


def getRobotJoint():
    """
    获取机械臂当前关节
    """
    resp = stub.getRobotJoint(robot_pb2.ReqInt(req=1))
    print_resp_data(resp)


def Offs():
    """
    TODO
    """
    resp = stub.Offs(robot_pb2.ReqArraydouble2(reqarr1=joint,
                                               reqarr2=joint))
    print_resp_data(resp)


def robot_getDOF():
    """
    获取机器人自由度
    """
    resp = stub.robot_getDOF(robot_pb2.ReqInt(req=0))
    print_resp_data(resp)


def additionaxis_getDOF():
    """
    获取机器人附加轴数量
    """
    resp = stub.additionaxis_getDOF(robot_pb2.ReqInt(req=0))
    print_resp_data(resp)


def robot_getNUM():
    """
    获取机器人数量
    """
    resp = stub.robot_getNUM(robot_pb2.ReqVoid())
    print_resp_data(resp)


if __name__ == '__main__':
    # 创建RPC Client
    channel = grpc.insecure_channel("192.168.1.167:50051")
    stub = robot_pb2_grpc.RobotServiceStub(channel)
    # 定义一些常量
    speed = robot_pb2.HyySpeed(dof=6, per_flag=2, tcp_flag=0, per=[2] * 10, tcp=0.5, orl=0.5)
    pose = stub.ReadRobpose(robot_pb2.ReqString(reqstr="p2"))
    pose2 = stub.ReadRobpose(robot_pb2.ReqString(reqstr="p3"))
    zone = stub.ReadZone(robot_pb2.ReqString(reqstr="z0"))
    tool = stub.ReadTool(robot_pb2.ReqString(reqstr="tool0"))
    wobj = stub.ReadWobj(robot_pb2.ReqString(reqstr="wobj0"))
    joint = stub.ReadRobjoint(robot_pb2.ReqString(reqstr="jhhh"))
    # 测试各个RPC接口
    # MoveL()
    # MoveC()
    # TRobotHome()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# Webots通过RPC读取数据
import time
from threading import Thread

import grpc
from controller import Robot

import robot_pb2
import robot_pb2_grpc


def get_robot_joint():
    """
    获取机械臂关节
    以线程的方式启动，避免阻塞Webot的仿真线程
    """
    print("线程已经启动")
    stub.MoveStart(robot_pb2.ReqVoid())
    resp = stub.TStreamRobotJoint(robot_pb2.ReqVoid())
    for r in resp:
        print(r.data)
        for i in range(6):
            joints[i].setPosition(-r.data[i])


channel = grpc.insecure_channel("192.168.1.167:50051")
stub = robot_pb2_grpc.RobotServiceStub(channel)

get_joint_thread = Thread(target=get_robot_joint)
get_joint_thread.start()
robot = Robot()
joints = []
for i in range(1,7):
    joints.append(robot.getMotor(f"xb4_joint_{i}"))
timestep = int(robot.getBasicTimeStep())
while robot.step(timestep) != -1:
    time.sleep(0.01)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# RPC控制机械臂移动
import grpc
import robot_pb2
import robot_pb2_grpc
# 三指机械爪
from robot_python_claw import DH3
if __name__ == '__main__':
    channel = grpc.insecure_channel("192.168.1.167:50051")
    stub = robot_pb2_grpc.RobotServiceStub(channel)
    joint_1 = stub.ReadRobjoint(robot_pb2.ReqString(reqstr="j_1"))
    joint_2 = stub.ReadRobjoint(robot_pb2.ReqString(reqstr="j_2"))
    joint_3 = stub.ReadRobjoint(robot_pb2.ReqString(reqstr="j_3"))
    joint_4 = stub.ReadRobjoint(robot_pb2.ReqString(reqstr="j_4"))
    dh3 = DH3()
    dh3.set_force(50)
    dh3.set_angle(60)
    dh3.set_position(90)
    speed = robot_pb2.HyySpeed(dof=6, per_flag=2, tcp_flag=0, per=[3] * 10, tcp=0.5, orl=0.5)
    zone = stub.ReadZone(robot_pb2.ReqString(reqstr="z0"))
    tool = stub.ReadTool(robot_pb2.ReqString(reqstr="tool0"))
    wobj = stub.ReadWobj(robot_pb2.ReqString(reqstr="wobj0"))
    stub.TGetRobotNum(robot_pb2.ReqVoid())
    stub.MoveA(robot_pb2.ReqMoveA(
        rjoint=joint_1,
        rspeed=speed,
        rzone=zone,
        rtool=tool,
        wobj=wobj
    ))
    dh3.set_position(30)
    stub.MoveA(robot_pb2.ReqMoveA(
        rjoint=joint_2,
        rspeed=speed,
        rzone=zone,
        rtool=tool,
        wobj=wobj
    ))
    stub.MoveA(robot_pb2.ReqMoveA(
        rjoint=joint_3,
        rspeed=speed,
        rzone=zone,
        rtool=tool,
        wobj=wobj
    ))
    stub.MoveA(robot_pb2.ReqMoveA(
        rjoint=joint_4,
        rspeed=speed,
        rzone=zone,
        rtool=tool,
        wobj=wobj
    ))
    dh3.set_position(90)
    stub.TRobotHome(robot_pb2.ReqVoid())