unitree_sdk2_python/unitree_sdk2py/rpc/client.py at 8a835a6c18bc8a048b728819b01b8afe446f69cd · unitreerobotics/unitree_sdk2_python · GitHub

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
from .client_base import ClientBase
from .lease_client import LeaseClient
from .internal import *

"""
" class Client
"""
class Client(ClientBase):
    def __init__(self, serviceName: str, enabaleLease: bool = False):
        super().__init__(serviceName)

        self.__apiMapping = {}
        self.__apiVersion = None
        self.__leaseClient = None
        self.__enableLease = enabaleLease

        if (self.__enableLease):
            self.__leaseClient = LeaseClient(serviceName)
            self.__leaseClient.Init()

    def WaitLeaseApplied(self):
        if self.__enableLease:
            self.__leaseClient.WaitApplied()

    def GetLeaseId(self):
        if self.__enableLease:
            return self.__leaseClient.GetId()
        else:
            return None

    def GetApiVersion(self):
        return self.__apiVersion

    def GetServerApiVersion(self):
        code, apiVerson = self._CallBase(RPC_API_ID_INTERNAL_API_VERSION, "{}", 0, 0)
        if code != 0:
            print("[Client] get server api version error:", code)
            return code, None
        else:
            return code, apiVerson

    def _SetApiVerson(self, apiVersion: str):
        self.__apiVersion = apiVersion

    def _Call(self, apiId: int, parameter: str):
        ret, proirity, leaseId = self.__CheckApi(apiId)
        if ret == 0:
            return self._CallBase(apiId, parameter, proirity, leaseId)
        else:
            return RPC_ERR_CLIENT_API_NOT_REG, None

    def _CallNoReply(self, apiId: int, parameter: str):
        ret, proirity, leaseId = self.__CheckApi(apiId)
        if ret == 0:
            return self._CallNoReplyBase(apiId, parameter, proirity, leaseId)
        else:
            return RPC_ERR_CLIENT_API_NOT_REG

    def _CallBinary(self, apiId: int, parameter: list):
        ret, proirity, leaseId = self.__CheckApi(apiId)
        if ret == 0:
            return self._CallBinaryBase(apiId, parameter, proirity, leaseId)
        else:
            return RPC_ERR_CLIENT_API_NOT_REG, None

    def _CallBinaryNoReply(self, apiId: int, parameter: list):
        ret, proirity, leaseId = self.__CheckApi(apiId)
        if ret == 0:
            return self._CallBinaryNoReplyBase(apiId, parameter, proirity, leaseId)
        else:
            return RPC_ERR_CLIENT_API_NOT_REG

    def _CallParamBinary(self, apiId: int, parameter: str, binary: list):
        ret, proirity, leaseId = self.__CheckApi(apiId)
        if ret == 0:
            return self._CallParamBinaryBase(apiId, parameter, binary, proirity, leaseId)
        else:
            return RPC_ERR_CLIENT_API_NOT_REG, None

    def _RegistApi(self, apiId: int, proirity: int):
        self.__apiMapping[apiId] = proirity

    def __CheckApi(self, apiId: int):
        proirity = 0
        leaseId = 0

        if apiId > RPC_INTERNAL_API_ID_MAX:
            proirity = self.__apiMapping.get(apiId)

            if proirity is None:
                return RPC_ERR_CLIENT_API_NOT_REG, proirity, leaseId

            if self.__enableLease:
                leaseId = self.__leaseClient.GetId()

        return 0, proirity, leaseId