fischertechnik-factory-twin/simulator/data_models/mqtt_message.py

import json


class MqttMessage:
    """
    Represents an MQTT message with a topic, payload, and origin.

    This class is used to construct and manage MQTT messages, including their
    serialization to JSON format. It provides an interface to handle MQTT
    messages, allowing for easy manipulation of the message topic, payload, and
    origin.

    Attributes:
        topic (str): The topic associated with the MQTT message.
        payload (dict): The payload data of the MQTT message.
        origin (str): The origin of the MQTT message, 'sim' or 'real'.
    """

    def __init__(self, message: str | None = None, origin: str = 'sim'):
        self.topic: str = ""
        self.payload: dict = dict()
        self.origin: str = origin
        if message is not None:
            message = json.loads(message)
            self.topic = message['topic']
            self.payload = json.loads(message['payload'])
            if 'origin' in message:
                self.origin = message['origin']

    def to_json(self) -> str:
        total_msg = dict()
        total_msg['topic'] = self.topic
        total_msg['payload'] = self.payload
        total_msg['origin'] = self.origin
        return json.dumps(total_msg)

    def __str__(self) -> str:
        return f"{type(self).__name__}(origin='{self.origin}', topic='{self.topic}', payload={self.payload})"

    def __repr__(self) -> str:
        return f"{type(self).__name__}(origin='{self.origin}', topic='{self.topic}', payload={self.payload})"