Clean up code redundancy in PacketBuilder/PacketReader

Author: Davnit

bncs/packets.py

@@ -113,15 +113,7 @@
 
 
 class BncsPacket(PacketBuilder):
-    def __init__(self, packet_id):
-        super().__init__(packet_id)
-
-    @property
-    def length(self):
-        return self.__len__()
-
-    def __len__(self):
-        return super().__len__() + 4
+    HEADER_SIZE = 4
 
     def __str__(self):
         return "BNCS " + super().__str__()
@@ -150,17 +142,8 @@ def __init__(self, data):
         self.packet_id = self.get_byte()
         self.length = self.get_word()
 
-    def __len__(self):
-        return self.length
-
     def __str__(self):
         return "BNCS " + super().__str__()
 
     def get_name(self):
         return get_packet_name(self, globals(), "SID_")
-
-    @classmethod
-    async def read_from(cls, reader):
-        packet = cls(await reader.readexactly(4))
-        await packet.fill(reader)
-        return packet

bncs/utils/packet.py

@@ -17,6 +17,8 @@ def get_packet_name(packet, names, prefix=None):
 
 class PacketBuilder(abc.ABC, DataBuffer):
     """Helper class for creating and writing packets."""
+    HEADER_SIZE = 0
+
     def __init__(self, packet_id, data=None):
         """Creates a new packet with the specified ID."""
         self.packet_id = packet_id
@@ -25,7 +27,10 @@ def __init__(self, packet_id, data=None):
 
     @property
     def length(self):
-        return super().__len__()
+        return len(self)
+
+    def __len__(self):
+        return super().__len__() + self.HEADER_SIZE
 
     def __str__(self):
         if str_id := self.get_name():
@@ -82,7 +87,7 @@ def get_name(self):
 
     @property
     def data_len(self):
-        """The actual number of bytes in the packet."""
+        """The actual length of the data in the buffer."""
         return super().__len__()
 
     @property
@@ -104,10 +109,17 @@ def reset(self):
         self.position = self.HEADER_SIZE
 
     @classmethod
-    @abc.abstractmethod
     async def read_from(cls, stream):
         """Reads a full packet from the stream"""
-        pass
+        from asyncio import IncompleteReadError
+
+        try:
+            packet = cls(await stream.readexactly(cls.HEADER_SIZE))
+        except IncompleteReadError:
+            return None
+
+        await packet.fill(stream)
+        return packet
 
     async def fill(self, stream):
         """Reads the remaining packet data from a stream."""

bnls/packets.py

@@ -34,11 +34,7 @@
 
 
 class BnlsPacket(PacketBuilder):
-    def __init__(self, packet_id):
-        super().__init__(packet_id)
-
-    def __len__(self):
-        return super().__len__() + 3
+    HEADER_SIZE = 3
 
     def __str__(self):
         return "BNLS " + super().__str__()
@@ -62,23 +58,8 @@ def __init__(self, data):
         self.length = self.get_word()
         self.packet_id = self.get_byte()
 
-    def __len__(self):
-        return self.length
-
     def __str__(self):
         return "BNLS " + super().__str__()
 
     def get_name(self):
         return get_packet_name(self, globals(), "BNLS_")
-
-    @classmethod
-    async def read_from(cls, reader):
-        from asyncio import IncompleteReadError
-
-        try:
-            packet = BnlsReader(await reader.readexactly(3))
-        except IncompleteReadError:
-            return None
-
-        await packet.fill(reader)
-        return packet

botnet/packets.py

@@ -26,11 +26,7 @@
 
 
 class BotNetPacket(PacketBuilder):
-    def __init__(self, packet_id):
-        super().__init__(packet_id)
-
-    def __len__(self):
-        return super().__len__() + 4
+    HEADER_SIZE = 4
 
     def __str__(self):
         return "BotNet " + super().__str__()
@@ -59,22 +55,8 @@ def __init__(self, data):
         self.packet_id = self.get_byte()
         self.length = self.get_word()
 
-    def __len__(self):
-        return self.length
-
     def __str__(self):
         return "BotNet " + super().__str__()
 
     def get_name(self):
         return get_packet_name(self, globals(), "BOTNET_")
-
-    @classmethod
-    async def read_from(cls, stream):
-        from asyncio import IncompleteReadError
-        try:
-            packet = cls(await stream.readexactly(4))
-        except IncompleteReadError:
-            return None
-
-        await packet.fill(stream)
-        return packet