Refactor and clean up protocol implementations

Refactored multiple protocol files for improved readability and consistency, including formatting, string handling, and error management. Removed unused imports, standardized byte and string operations, and enhanced code clarity across AoE1, AoE2, CoD1, CoD4, CoD5, DirectPlay, and related modules. No functional changes were made to protocol logic.

Author: BattlefieldDuck

opengsq/binary_reader.py

@@ -16,7 +16,7 @@ def prepend_bytes(self, data):
         self.__data = data + self.__data
 
     def read(self) -> bytes:
-        return self.__data[self.stream_position:]
+        return self.__data[self.stream_position :]
 
     def read_byte(self) -> int:
         data = self.__data[self.stream_position]
@@ -25,39 +25,49 @@ def read_byte(self) -> int:
         return data
 
     def read_bytes(self, count: int) -> bytes:
-        data = self.__data[self.stream_position:self.stream_position + count]
+        data = self.__data[self.stream_position : self.stream_position + count]
         self.stream_position += count
 
         return data
 
     def read_short(self, unsigned=True) -> int:
-        format = 'H' if unsigned else 'h'
-        data = struct.unpack(f'<{format}', self.__data[self.stream_position:self.stream_position + 2])[0]
+        format = "H" if unsigned else "h"
+        data = struct.unpack(
+            f"<{format}", self.__data[self.stream_position : self.stream_position + 2]
+        )[0]
         self.stream_position += 2
 
         return data
 
     def read_long(self, unsigned=False) -> int:
-        format = 'L' if unsigned else 'l'
-        data = struct.unpack(f'<{format}', self.__data[self.stream_position:self.stream_position + 4])[0]
+        format = "L" if unsigned else "l"
+        data = struct.unpack(
+            f"<{format}", self.__data[self.stream_position : self.stream_position + 4]
+        )[0]
         self.stream_position += 4
 
         return data
 
     def read_long_long(self) -> int:
-        data = struct.unpack('<q', self.__data[self.stream_position:self.stream_position + 8])[0]
+        data = struct.unpack(
+            "<q", self.__data[self.stream_position : self.stream_position + 8]
+        )[0]
         self.stream_position += 8
 
         return data
 
     def read_float(self) -> float:
-        data = struct.unpack('<f', self.__data[self.stream_position:self.stream_position + 4])[0]
+        data = struct.unpack(
+            "<f", self.__data[self.stream_position : self.stream_position + 4]
+        )[0]
         self.stream_position += 4
 
         return data
 
-    def read_string(self, delimiters=[b'\x00'], encoding='utf-8', errors='ignore') -> str:
-        bytes_string = b''
+    def read_string(
+        self, delimiters=[b"\x00"], encoding="utf-8", errors="ignore"
+    ) -> str:
+        bytes_string = b""
 
         while self.remaining_bytes() > 0:
             stream_byte = bytes([self.read_byte()])
@@ -69,7 +79,9 @@ def read_string(self, delimiters=[b'\x00'], encoding='utf-8', errors='ignore') -
 
         return str(bytes_string, encoding=encoding, errors=errors)
 
-    def read_pascal_string(self, encoding='utf-8', errors='ignore'):
+    def read_pascal_string(self, encoding="utf-8", errors="ignore"):
         length = self.read_byte()
-        pascal_string = str(self.read_bytes(length - 1), encoding=encoding, errors=errors)
+        pascal_string = str(
+            self.read_bytes(length - 1), encoding=encoding, errors=errors
+        )
         return pascal_string

opengsq/protocol_socket.py

@@ -10,10 +10,12 @@ class SocketKind(Enum):
     SOCK_DGRAM = auto()
 
 
-class Socket():
+class Socket:
     @staticmethod
     async def gethostbyname(hostname: str):
-        return await asyncio.get_running_loop().run_in_executor(None, socket.gethostbyname, hostname)
+        return await asyncio.get_running_loop().run_in_executor(
+            None, socket.gethostbyname, hostname
+        )
 
     class Protocol(asyncio.Protocol):
         def __init__(self, timeout: float):
@@ -78,13 +80,17 @@ async def __connect(self, remote_addr):
                 lambda: self.__protocol,
                 host=remote_addr[0],
                 port=remote_addr[1],
-                local_addr=('0.0.0.0', self.__local_port) if self.__local_port else None
+                local_addr=("0.0.0.0", self.__local_port)
+                if self.__local_port
+                else None,
             )
         else:
             self.__transport, _ = await loop.create_datagram_endpoint(
                 lambda: self.__protocol,
                 remote_addr=remote_addr,
-                local_addr=('0.0.0.0', self.__local_port) if self.__local_port else None
+                local_addr=("0.0.0.0", self.__local_port)
+                if self.__local_port
+                else None,
             )
 
     def close(self):
@@ -156,8 +162,8 @@ async def communicate(protocol: ProtocolBase, data: bytes, source_port: int = No
             loop = asyncio.get_running_loop()
             transport, protocol_instance = await loop.create_datagram_endpoint(
                 lambda: Socket.Protocol(protocol._timeout),  # Use public Protocol class
-                local_addr=('0.0.0.0', source_port if source_port else 0),
-                allow_broadcast=protocol._allow_broadcast
+                local_addr=("0.0.0.0", source_port if source_port else 0),
+                allow_broadcast=protocol._allow_broadcast,
             )
 
             try:
@@ -183,13 +189,15 @@ def __init__(self):
         super().__init__(SocketKind.SOCK_STREAM)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
+
     async def test_socket_async():
         with Socket() as socket_async:
             socket_async.settimeout(5)
-            await socket_async.connect(('122.128.109.245', 27015))
+            await socket_async.connect(("122.128.109.245", 27015))
             socket_async.send(
-                b'\xFF\xFF\xFF\xFFTSource Engine Query\x00\xFF\xFF\xFF\xFF')
+                b"\xff\xff\xff\xffTSource Engine Query\x00\xff\xff\xff\xff"
+            )
             print(await socket_async.recv())
 
     loop = asyncio.get_event_loop()