Coverage for filip/clients/mqtt/encoder/json.py: 95%

1""" 

2Json encoder class for all IoTA-JSON MQTT message encoders 

3""" 

4 

5import json 

6from typing import Any, Dict, Tuple 

7from filip.clients.mqtt.encoder import BaseEncoder 

8from filip.models.mqtt import IoTAMQTTMessageType 

9 

10 

11class Json(BaseEncoder): 

12 """ 

13 Json encoder class for all IoTA-JSON MQTT message encoders 

14 """ 

15 

16 prefix = "/json" 

17 

18 def __init__(self): 

19 super().__init__() 

20 

21 def decode_message(self, msg, decoder="utf-8") -> Tuple[str, str, Dict]: 

22 apikey, device_id, payload = super().decode_message(msg=msg, decoder=decoder) 

23 payload = json.loads(payload) 

24 return apikey, device_id, payload 

25 

26 def encode_msg(self, device_id, payload: Any, msg_type: IoTAMQTTMessageType) -> str: 

27 if msg_type == IoTAMQTTMessageType.SINGLE: 

28 return payload 

29 elif msg_type == IoTAMQTTMessageType.MULTI: 

30 payload = super()._parse_timestamp(payload=payload) 

31 return json.dumps(payload, default=str) 

32 elif msg_type == IoTAMQTTMessageType.CMDEXE: 

33 return json.dumps(payload) 

34 super()._raise_encoding_error(payload=payload, msg_type=msg_type)