MqttMonitor/MqttReasonCodes.py

# -*- coding: utf-8 -*-

# Copyright (c) 2021 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the translated MQTT v5 reason codes.
"""

from PyQt6.QtCore import QCoreApplication

from paho.mqtt.packettypes import PacketTypes

MqttReasonCodeNames = {
    0: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Success"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK, PacketTypes.PUBREC,
            PacketTypes.PUBREL, PacketTypes.PUBCOMP, PacketTypes.UNSUBACK,
            PacketTypes.AUTH],
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Normal disconnection"
        ): [PacketTypes.DISCONNECT],
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Granted QoS 0"
        ): [PacketTypes.SUBACK]},
    1: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Granted QoS 1"
        ): [PacketTypes.SUBACK]},
    2: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Granted QoS 2"
        ): [PacketTypes.SUBACK]},
    4: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Disconnect with will message"
        ): [PacketTypes.DISCONNECT]},
    16: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "No matching subscribers"
        ): [PacketTypes.PUBACK, PacketTypes.PUBREC]},
    17: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "No subscription found"
        ): [PacketTypes.UNSUBACK]},
    24: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Continue authentication"
        ): [PacketTypes.AUTH]},
    25: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Re-authenticate"
        ): [PacketTypes.AUTH]},
    128: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Unspecified error"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK,
            PacketTypes.PUBREC, PacketTypes.SUBACK,
            PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]},
    129: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Malformed packet"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    130: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Protocol error"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    131: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Implementation specific error"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK,
            PacketTypes.PUBREC, PacketTypes.SUBACK,
            PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]},
    132: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Unsupported protocol version"
        ): [PacketTypes.CONNACK]},
    133: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Client identifier not valid"
        ): [PacketTypes.CONNACK]},
    134: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Bad user name or password"
        ): [PacketTypes.CONNACK]},
    135: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Not authorized"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK,
            PacketTypes.PUBREC, PacketTypes.SUBACK,
            PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]},
    136: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Server unavailable"
        ): [PacketTypes.CONNACK]},
    137: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Server busy"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    138: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Banned"
        ): [PacketTypes.CONNACK]},
    139: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Server shutting down"
        ): [PacketTypes.DISCONNECT]},
    140: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Bad authentication method"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    141: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Keep alive timeout"
        ): [PacketTypes.DISCONNECT]},
    142: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Session taken over"
        ): [PacketTypes.DISCONNECT]},
    143: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Topic filter invalid"
        ): [PacketTypes.SUBACK, PacketTypes.UNSUBACK,
            PacketTypes.DISCONNECT]},
    144: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Topic name invalid"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK,
            PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
    145: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Packet identifier in use"
        ): [PacketTypes.PUBACK, PacketTypes.PUBREC,
            PacketTypes.SUBACK, PacketTypes.UNSUBACK]},
    146: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Packet identifier not found"
        ): [PacketTypes.PUBREL, PacketTypes.PUBCOMP]},
    147: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Receive maximum exceeded"
        ): [PacketTypes.DISCONNECT]},
    148: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Topic alias invalid"
        ): [PacketTypes.DISCONNECT]},
    149: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Packet too large"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    150: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Message rate too high"
        ): [PacketTypes.DISCONNECT]},
    151: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Quota exceeded"
        ): [PacketTypes.CONNACK, PacketTypes.PUBACK,
            PacketTypes.PUBREC, PacketTypes.SUBACK,
            PacketTypes.DISCONNECT], },
    152: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Administrative action"
        ): [PacketTypes.DISCONNECT]},
    153: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Payload format invalid"
        ): [PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
    154: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Retain not supported"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    155: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "QoS not supported"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    156: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Use another server"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    157: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Server moved"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    158: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Shared subscription not supported"
        ): [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
    159: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Connection rate exceeded"
        ): [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
    160: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Maximum connect time"
        ): [PacketTypes.DISCONNECT]},
    161: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Subscription identifiers not supported"
        ): [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
    162: {
        QCoreApplication.translate(
            "MqttReasonCodeNames", "Wildcard subscription not supported"
        ): [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
}

def mqttReasonCode(rc, packetType):
    """
    Function to get the readable reason code string given the result code and
    the packet type.
    
    @param rc result code
    @type int
    @param packetType packet type
    @type PacketTypes (= int)
    """
    if rc not in MqttReasonCodeNames:
        return QCoreApplication.translate(
            "MqttReasonCodeNames", "Unknown result code ({0})").format(rc)
    
    messages = MqttReasonCodeNames[rc]
    messagesList = [message for message in messages.keys()
                    if packetType in messages[message]]
    if len(messagesList) == 0:
        return QCoreApplication.translate(
            "MqttReasonCodeNames",
            "Unknown result code ({0}) for packet type '{1}'"
        ).format(rc, packetType)
    
    return messagesList[0]