mirror/thingsboard-gateway
1
0
Fork 0
mirror of https://github.com/thingsboard/thingsboard-gateway synced 2025-10-26 22:31:42 +08:00
Code Issues Releases Wiki Activity

Added processiong for GRPC connector (PostTelemetryRequest and PostAttributesRequest)

Browse Source
This commit is contained in:
zbeacon
2021-12-01 16:09:28 +02:00
parent 9affaab677
commit 0233980065
4 changed files with 83 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
thingsboard_gateway/gateway/grpc_service/grpc_connector.py
View File

@@ -14,11 +14,13 @@
from thingsboard_gateway.connectors.connector import Connector
from thingsboard_gateway.gateway.grpc_service.tb_grpc_manager import TBGRPCServerManager
class GrpcConnector(Connector):
def __init__(self, gateway, config, context):
def __init__(self, gateway, config, tb_grpc_server_manager: TBGRPCServerManager):
self.name = None
self.__server_manager = tb_grpc_server_manager
def setName(self, name):
self.name = name

65
thingsboard_gateway/gateway/grpc_service/grpc_uplink_converter.py
View File

@@ -11,12 +11,11 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import Union
from thingsboard_gateway.connectors.converter import Converter, log
from thingsboard_gateway.gateway.constant_enums import UplinkMessageType
from thingsboard_gateway.gateway.proto.messages_pb2 import ConnectMsg, DisconnectMsg, GatewayAttributesMsg, GatewayAttributesRequestMsg, GatewayClaimMsg, \
GatewayRpcResponseMsg, GatewayTelemetryMsg, RegisterConnectorMsg, Response, UnregisterConnectorMsg
GatewayRpcResponseMsg, GatewayTelemetryMsg, RegisterConnectorMsg, Response, UnregisterConnectorMsg, KeyValueProto, KeyValueType
class GrpcUplinkConverter(Converter):
@@ -26,8 +25,8 @@ class GrpcUplinkConverter(Converter):
UplinkMessageType.GatewayTelemetryMsg: self.__convert_gateway_telemetry_msg,
UplinkMessageType.GatewayAttributesMsg: self.__convert_gateway_attributes_msg,
UplinkMessageType.GatewayClaimMsg: self.__convert_gateway_claim_msg,
UplinkMessageType.RegisterConnectorMsg: self.__convert_register_connector_msg,
UplinkMessageType.UnregisterConnectorMsg: self.__convert_unregister_connector_msg,
# UplinkMessageType.RegisterConnectorMsg: self.__convert_register_connector_msg,
# UplinkMessageType.UnregisterConnectorMsg: self.__convert_unregister_connector_msg,
UplinkMessageType.ConnectMsg: self.__convert_connect_msg,
UplinkMessageType.DisconnectMsg: self.__convert_disconnect_msg,
UplinkMessageType.GatewayRpcResponseMsg: self.__convert_gateway_rpc_response_msg,
@@ -45,34 +44,68 @@ class GrpcUplinkConverter(Converter):
log.exception("[GRPC] ", e)
return {}
@staticmethod
def __convert_response_msg(self, msg: Response):
pass
log.debug("Converted response: %r", msg.response)
def __convert_gateway_telemetry_msg(self, msg: GatewayTelemetryMsg):
pass
@staticmethod
def __convert_gateway_telemetry_msg(msg: GatewayTelemetryMsg):
result = []
for telemetry_msg in msg.msg:
device_dict = {"deviceName": telemetry_msg.deviceName, "telemetry": []}
for post_telemetry_msg in telemetry_msg.msg:
for ts_kv_list in post_telemetry_msg.tsKvList:
ts_kv_list_dict = {"ts": ts_kv_list.ts, "values": {}}
for kv in ts_kv_list.kv:
ts_kv_list_dict['values'][kv.key] = GrpcUplinkConverter.get_value(kv)
device_dict['telemetry'].append(ts_kv_list_dict)
result.append(device_dict)
return result
def __convert_gateway_attributes_msg(self, msg: GatewayAttributesMsg):
pass
@staticmethod
def __convert_gateway_attributes_msg(msg: GatewayAttributesMsg):
result = []
for attributes_msg in msg.msg:
device_dict = {"deviceName": attributes_msg.deviceName, "attributes": {}}
for post_attribute_msg in attributes_msg.msg:
for kv in post_attribute_msg.kv:
device_dict['attributes'][kv.key] = GrpcUplinkConverter.get_value(kv)
result.append(device_dict)
return result
def __convert_gateway_claim_msg(self, msg: GatewayClaimMsg):
pass
def __convert_register_connector_msg(self, msg: RegisterConnectorMsg):
pass
def __convert_unregister_connector_msg(self, msg: UnregisterConnectorMsg):
pass
# def __convert_register_connector_msg(self, msg: RegisterConnectorMsg):
# pass
#
# def __convert_unregister_connector_msg(self, msg: UnregisterConnectorMsg):
# pass
@staticmethod
def __convert_connect_msg(msg: ConnectMsg) -> dict:
result_dict = {'deviceName': msg.deviceName, 'deviceType': msg.deviceType}
return result_dict
def __convert_disconnect_msg(self, msg: DisconnectMsg):
pass
@staticmethod
def __convert_disconnect_msg(msg: DisconnectMsg):
return {"deviceName": msg.deviceName}
def __convert_gateway_rpc_response_msg(self, msg: GatewayRpcResponseMsg):
pass
def __convert_gateway_attributes_request_msg(self, msg: GatewayAttributesRequestMsg):
pass
@staticmethod
def get_value(msg: KeyValueProto):
if msg.type == KeyValueProto.BOOLEAN_V:
return msg.bool_v
if msg.type == KeyValueProto.LONG_V:
return msg.long_v
if msg.type == KeyValueProto.DOUBLE_V:
return msg.double_v
if msg.type == KeyValueProto.STRING_V:
return msg.string_v
if msg.type == KeyValueProto.JSON_V:
return msg.json_v

22
thingsboard_gateway/gateway/grpc_service/tb_grpc_manager.py
View File

@@ -77,9 +77,19 @@ class TBGRPCServerManager(Thread):
if msg.HasField("response"):
pass
if msg.HasField("gatewayTelemetryMsg"):
pass
data = self.__uplink_converter.convert(UplinkMessageType.GatewayTelemetryMsg, msg.gatewayTelemetryMsg)
result_status = self.__gateway.send_to_storage(self.sessions[context.peer()]['name'], data)
outgoing_message = self.__downlink_converter.convert([DownlinkMessageType.Response], result_status)
if not outgoing_message:
log.error("Cannot convert outgoing message!")
self.__grpc_server.write(context, outgoing_message)
if msg.HasField("gatewayAttributesMsg"):
pass
data = self.__uplink_converter.convert(UplinkMessageType.GatewayAttributesMsg, msg.gatewayAttributesMsg)
result_status = self.__gateway.send_to_storage(self.sessions[context.peer()]['name'], data)
outgoing_message = self.__downlink_converter.convert([DownlinkMessageType.Response], result_status)
if not outgoing_message:
log.error("Cannot convert outgoing message!")
self.__grpc_server.write(context, outgoing_message)
if msg.HasField("gatewayClaimMsg"):
pass
if msg.HasField("registerConnectorMsg"):
@@ -95,7 +105,13 @@ class TBGRPCServerManager(Thread):
log.error("Cannot convert outgoing message!")
self.__grpc_server.write(context, outgoing_message)
if msg.HasField("disconnectMsg"):
pass
data = self.__uplink_converter.convert(UplinkMessageType.DisconnectMsg, msg.connectMsg)
data['name'] = self.sessions[context.peer()]['name']
result_status = self.__gateway.del_device_async(data)
outgoing_message = self.__downlink_converter.convert([DownlinkMessageType.Response], result_status)
if not outgoing_message:
log.error("Cannot convert outgoing message!")
self.__grpc_server.write(context, outgoing_message)
if msg.HasField("gatewayRpcResponseMsg"):
pass
if msg.HasField("gatewayAttributeRequestMsg"):

15
thingsboard_gateway/gateway/tb_gateway_service.py
View File

@@ -337,7 +337,7 @@ class TBGatewayService:
def __register_connector(self, context, connector_key):
if self.__grpc_connectors.get(connector_key) is not None and self.__grpc_connectors[connector_key]['name'] not in self.available_connectors:
target_connector = self.__grpc_connectors.get(connector_key)
connector = GrpcConnector(self, target_connector['config'], context)
connector = GrpcConnector(self, target_connector['config'], self.__grpc_manager)
connector.setName(target_connector['name'])
self.available_connectors[connector.get_name()] = connector
self.__grpc_manager.registration_finished(Status.SUCCESS, context, target_connector)
@@ -457,7 +457,12 @@ class TBGatewayService:
self._connect_with_connectors()
def send_to_storage(self, connector_name, data):
self.__converted_data_queue.put((connector_name, data), True, 100)
try:
self.__converted_data_queue.put((connector_name, data), True, 100)
return Status.SUCCESS
except Exception as e:
log.exception("Cannot put converted data!", e)
return Status.FAILURE
def __send_to_storage(self):
while True:
@@ -825,7 +830,11 @@ class TBGatewayService:
self.__connected_devices[device_name][event] = content
def del_device_async(self, data):
self.__async_device_actions_queue.put((DeviceActions.DISCONNECT, data))
if data['deviceName'] in self.__saved_devices:
self.__async_device_actions_queue.put((DeviceActions.DISCONNECT, data))
return Status.SUCCESS
else:
return Status.FAILURE
def del_device(self, device_name):
del self.__connected_devices[device_name]