The following code is given as a python example under the API.
I have already cloned the repo to my machine. Performed. make
python setup.py build
and
python setup.py install
This Unimplemented error still persists.
`print("open grpc channel . . .")
try:
# Connect without using TLS.
channel = grpc.insecure_channel(server)
except Exception:
raise ConnectionError('FAILED grpc channel')
else:
print(' grpc channel okay.')
print("open device service client. . .")
try:
# Device-queue API client.
client = api.DeviceServiceStub(channel)
except Exception:
raise ConnectionError('FAILED dev serv client')
else:
print(' dev serv client okay.')
# Define the API key meta-data.
auth_token = [("authorization", "Bearer %s" % api_token)]
# Construct request.
req = api.EnqueueDeviceQueueItemRequest()
req.queue_item.confirmed = False
req.queue_item.data = bytes([0x01, 0x09, 0x05, 0x0A ])
req.queue_item.dev_eui = dev_eui
req.queue_item.f_port = 2 # CS wanted 10 here ?
print(dir(client))
resp = client.Enqueue(req,
metadata=auth_token)
# Print the downlink id
print(resp.id)`
Running the code results in this error,
open grpc channel . . . grpc channel okay. open device service client. . . dev serv client okay. ['Activate', 'Create', 'CreateKeys', 'Deactivate', 'Delete', 'DeleteKeys', 'Enqueue', 'FlushDevNonces', 'FlushQueue', 'Get', 'GetActivation', 'GetKeys', 'GetLinkMetrics', 'GetMetrics', 'GetQueue', 'GetRandomDevAddr', 'List', 'Update', 'UpdateKeys', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__'] Traceback (most recent call last): File "controlapp.py", line 83, in <module> resp = client.Enqueue(req, metadata=auth_token) File "/home/linus/anaconda3/lib/python3.8/site-packages/grpc/_channel.py", line 946, in __call__ return _end_unary_response_blocking(state, call, False, None) File "/home/linus/anaconda3/lib/python3.8/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking raise _InactiveRpcError(state) grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNIMPLEMENTED details = "unknown method Enqueue for service api.DeviceService" debug_error_string = "{"created":"@1668050965.854003706","description":"Error received from peer ipv4:127.0.0.1:8080","file":"src/core/lib/surface/call.cc","file_line":1064,"grpc_message":"unknown method Enqueue for service api.DeviceService","grpc_status":12}"
I found several Enqueue methods located here,
`## github chirpstack-api/python/src/chirpstack_api/as_pb/external/api/deviceQueue_pb2_grpc.py
class DeviceQueueServiceServicer(object):
"""DeviceQueueService is the service managing the downlink data queue.
“”"
def Enqueue(self, request, context):
"""Enqueue adds the given item to the device-queue.
"""
context.set_code(
grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not
implemented!')
raise NotImplementedError('Method not
implemented!')`
Any updates from your side?