Source code for filip.clients.mqtt.encoder.json

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

import json
from typing import Any, Dict, Tuple
from filip.clients.mqtt.encoder import BaseEncoder
from filip.models.mqtt import IoTAMQTTMessageType


class Json(BaseEncoder):
    """
    Json encoder class for all IoTA-JSON MQTT message encoders
    """

    prefix = "/json"

    def __init__(self):
        super().__init__()

    def decode_message(self, msg, decoder="utf-8") -> Tuple[str, str, Dict]:
        apikey, device_id, payload = super().decode_message(msg=msg, decoder=decoder)
        payload = json.loads(payload)
        return apikey, device_id, payload

    def encode_msg(self, device_id, payload: Any, msg_type: IoTAMQTTMessageType) -> str:
        if msg_type == IoTAMQTTMessageType.SINGLE:
            return payload
        elif msg_type == IoTAMQTTMessageType.MULTI:
            payload = super()._parse_timestamp(payload=payload)
            return json.dumps(payload, default=str)
        elif msg_type == IoTAMQTTMessageType.CMDEXE:
            return json.dumps(payload)
        super()._raise_encoding_error(payload=payload, msg_type=msg_type)