Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions src/paho/mqtt/reasoncodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ def __init__(self, packetType: int, aName: str ="Success", identifier: int =-1):
self.value = identifier
self.getName() # check it's good

def __getName__(self, packetType, identifier):
def __getName__(self, packetType: int, identifier: int) -> str:
"""
Get the reason code string name for a specific identifier.
The name can vary by packet type for the same identifier, which
Expand All @@ -144,7 +144,7 @@ def __getName__(self, packetType, identifier):
raise ValueError(f"Expected exactly one name, found {namelist!r}")
return namelist[0]

def getId(self, name):
def getId(self, name: str) -> int:
"""
Get the numeric id corresponding to a reason code name.

Expand All @@ -157,21 +157,21 @@ def getId(self, name):
return code
raise KeyError(f"Reason code name not found: {name}")

def set(self, name):
def set(self, name: str) -> None:
self.value = self.getId(name)

def unpack(self, buffer):
def unpack(self, buffer: bytearray) -> int:
c = buffer[0]
name = self.__getName__(self.packetType, c)
self.value = self.getId(name)
return 1

def getName(self):
def getName(self) -> str:
"""Returns the reason code name corresponding to the numeric value which is set.
"""
return self.__getName__(self.packetType, self.value)

def __eq__(self, other):
def __eq__(self, other: Any) -> bool:
if isinstance(other, int):
return self.value == other
if isinstance(other, str):
Expand All @@ -180,28 +180,28 @@ def __eq__(self, other):
return self.value == other.value
return False

def __lt__(self, other):
def __lt__(self, other: Any) -> bool:
if isinstance(other, int):
return self.value < other
if isinstance(other, ReasonCode):
return self.value < other.value
return NotImplemented

def __repr__(self):
def __repr__(self) -> str:
try:
packet_name = PacketTypes.Names[self.packetType]
except IndexError:
packet_name = "Unknown"

return f"ReasonCode({packet_name}, {self.getName()!r})"

def __str__(self):
def __str__(self) -> str:
return self.getName()

def json(self):
def json(self) -> str:
return self.getName()

def pack(self):
def pack(self) -> bytearray:
return bytearray([self.value])

@property
Expand Down