Update teamspeak3.py

opengsq/protocols/teamspeak3.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import TcpClient
 
@@ -6,7 +8,8 @@ class TeamSpeak3(ProtocolBase):
     """
     This class represents the TeamSpeak 3 Protocol. It provides methods to interact with the TeamSpeak 3 API.
     """
-    full_name = 'TeamSpeak 3 Protocol'
+
+    full_name = "TeamSpeak 3 Protocol"
 
     def __init__(self, host: str, port: int, voice_port: int, timeout: float = 5):
         """
@@ -26,7 +29,7 @@ async def get_info(self) -> dict:
 
         :return: A dictionary containing the information of the game server.
         """
-        response = await self.__send_and_receive(b'serverinfo')
+        response = await self.__send_and_receive(b"serverinfo")
         return self.__parse_kvs(response)
 
     async def get_clients(self) -> list[dict]:
@@ -35,7 +38,7 @@ async def get_clients(self) -> list[dict]:
 
         :return: A list of clients on the game server.
         """
-        response = await self.__send_and_receive(b'clientlist')
+        response = await self.__send_and_receive(b"clientlist")
         return self.__parse_rows(response)
 
     async def get_channels(self) -> list[dict]:
@@ -44,7 +47,7 @@ async def get_channels(self) -> list[dict]:
 
         :return: A list of channels on the game server.
         """
-        response = await self.__send_and_receive(b'channellist -topic')
+        response = await self.__send_and_receive(b"channellist -topic")
         return self.__parse_rows(response)
 
     async def __send_and_receive(self, data: bytes):
@@ -63,13 +66,13 @@ async def __send_and_receive(self, data: bytes):
             await tcpClient.recv()
 
             # b'error id=0 msg=ok\n\r'
-            tcpClient.send(f'use port={self._voice_port}\n'.encode())
+            tcpClient.send(f"use port={self._voice_port}\n".encode())
             await tcpClient.recv()
 
-            tcpClient.send(data + b'\x0A')
-            response = b''
+            tcpClient.send(data + b"\x0A")
+            response = b""
 
-            while not response.endswith(b'error id=0 msg=ok\n\r'):
+            while not response.endswith(b"error id=0 msg=ok\n\r"):
                 response += await tcpClient.recv()
 
         # Remove last bytes b'\n\rerror id=0 msg=ok\n\r'
@@ -82,7 +85,7 @@ def __parse_rows(self, response: bytes):
         :param response: The response to parse rows from.
         :return: A list of dictionaries containing the parsed rows.
         """
-        return [self.__parse_kvs(row) for row in response.split(b'|')]
+        return [self.__parse_kvs(row) for row in response.split(b"|")]
 
     def __parse_kvs(self, response: bytes):
         """
@@ -93,26 +96,32 @@ def __parse_kvs(self, response: bytes):
         """
         kvs = {}
 
-        for kv in response.split(b' '):
-            items = kv.split(b'=', 1)
-            key = str(items[0], encoding='utf-8', errors='ignore')
-            val = str(items[1], encoding='utf-8', errors='ignore') if len(items) == 2 else ""
-            kvs[key] = val.replace('\\p', '|').replace('\\s', ' ').replace('\\/', '/')
+        for kv in response.split(b" "):
+            items = kv.split(b"=", 1)
+            key = str(items[0], encoding="utf-8", errors="ignore")
+            val = (
+                str(items[1], encoding="utf-8", errors="ignore")
+                if len(items) == 2
+                else ""
+            )
+            kvs[key] = val.replace("\\p", "|").replace("\\s", " ").replace("\\/", "/")
 
         return kvs
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
 
     async def main_async():
-        teamspeak3 = TeamSpeak3(host='145.239.200.2', port=10011, voice_port=9987, timeout=5.0)
+        teamspeak3 = TeamSpeak3(
+            host="145.239.200.2", port=10011, voice_port=9987, timeout=5.0
+        )
         info = await teamspeak3.get_info()
-        print(json.dumps(info, indent=None) + '\n')
+        print(json.dumps(info, indent=None) + "\n")
         clients = await teamspeak3.get_clients()
-        print(json.dumps(clients, indent=None) + '\n')
+        print(json.dumps(clients, indent=None) + "\n")
         channels = await teamspeak3.get_channels()
-        print(json.dumps(channels, indent=None) + '\n')
+        print(json.dumps(channels, indent=None) + "\n")
 
     asyncio.run(main_async())