Start working on connection less messages

scripts/generate_messages.py

@@ -4,7 +4,12 @@ import os
 import json
 
 from typing import \
-    TypedDict, Literal, Optional, Dict, Union, TextIO
+    TypedDict, Literal, Optional, Dict, Union, TextIO, Annotated
+
+KINDCONLESS = Literal[ \
+            'packed_addresses', \
+            'be_uint16', \
+            'uint8']
 
 KIND = Literal[ \
             'int32', \
@@ -44,6 +49,9 @@ class ArrayMemberTypeJson(TypedDict):
     # strings
     disallow_cc: bool
 
+class NetConnlessMemberTypeJson(TypedDict):
+    kind: KINDCONLESS
+
 class NetMessageMemberTypeJson(TypedDict):
     kind: KIND
     inner: InnerNetMessageMemberTypeJson
@@ -71,11 +79,72 @@ class NetMessageJson(TypedDict):
     members: list[NetMessageMemberJson]
     attributes: list[Literal['msg_encoding']]
 
+class NetConnlessMemberJson(TypedDict):
+    name: list[str]
+    type: NetConnlessMemberTypeJson
+
+class NetConnlessJson(TypedDict):
+    id: Annotated[list[int], 8]
+    name: list[str]
+    members: list[NetConnlessMemberJson]
+
 class SpecJson(TypedDict):
     constants: list[ConstantJson]
     game_enumerations: list[GameEnumJson]
     game_messages: list[NetMessageJson]
     system_messages: list[NetMessageJson]
+    connless_messages: list[NetConnlessJson]
+
+def gen_match_file_connless7(
+        messages: list[NetMessageJson]
+):
+    match_code: str = """# generated by scripts/generate_messages.py
+from typing import Optional
+
+import twnet_parser.msg7
+from twnet_parser.connless_message import ConnlessMessage
+
+"""
+
+    msg: NetConnlessJson
+    for msg in messages:
+        name_snake = name_to_snake(msg['name'])
+        match_code += f"import twnet_parser.messages7.connless" \
+                f".{name_snake}" \
+                " as \\\n" \
+                f"       connless_{name_snake}\n"
+
+    match_code += f"""
+def match_connless7(msg_id: list[int], data: bytes) -> ConnlessMessage:
+    msg: Optional[ConnlessMessage] = None
+"""
+
+    if_ = 'if'
+    for msg in messages:
+        name_snake = name_to_snake(msg['name'])
+        name_camel = name_to_camel(msg['name'])
+        match_code += \
+f"""
+    {if_} msg_id == twnet_parser.msg7.{name_snake.upper()}:
+        msg = connless_{name_snake}.Msg{name_camel}()"""
+        if_ = 'elif'
+
+    match_code += '\n\n    if msg is None:\n'
+    match_code += '        '
+    match_code += 'raise ValueError('
+    match_code += 'f"Error: unknown conless ' \
+            ' message id={msg_id} data={data[0]}")\n'
+    match_code += '\n'
+    match_code += '    msg.unpack(data)\n'
+    match_code += '    return msg\n'
+
+    dirname = os.path.dirname(__file__)
+    file_path= os.path.join(
+            dirname,
+            f'../twnet_parser/msg_matcher/connless7.py')
+    with open(file_path, 'w') as out_file:
+        print(f"Generating {file_path} ...")
+        out_file.write(match_code)
 
 def gen_match_file7(
         msg_type: Literal['system', 'game'],
@@ -147,6 +216,22 @@ def name_to_snake(name_list: list[str]) -> str:
     name = '_'.join(name_list)
     return fix_name_conflict(name)
 
+def gen_unpack_members_connless7(msg: NetConnlessJson) -> str:
+    res: str = ''
+    for member in msg['members']:
+        unpacker = 'int()'
+        if member['type']['kind'] == 'be_uint16':
+            unpacker = 'be_uint16()'
+        elif member['type']['kind'] == 'uint8':
+            unpacker = 'uint8()'
+        elif member['type']['kind'] == 'packed_addresses': # TODO: packed_addresses
+            unpacker = 'int()'
+        else:
+            raise ValueError(f"Error: unknown type {member['type']}")
+        name = name_to_snake(member["name"])
+        res += f'        self.{name} = unpacker.get_{unpacker}\n'
+    return res
+
 def gen_unpack_members(msg: NetMessageJson) -> str:
     res: str = ''
     for member in msg['members']:
@@ -227,6 +312,27 @@ def gen_unpack_members(msg: NetMessageJson) -> str:
         res += f'        self.{name} = unpacker.get_{unpacker}\n'
     return res
 
+def get_dependencies_connless7(msg: NetConnlessJson) -> str:
+    packer_deps: list[str] = []
+    typing_deps: list[str] = ['Literal']
+    for member in msg['members']:
+        if member['type']['kind'] == 'packed_addresses': # TODO: packed_addresses
+            packer_deps.append('pack_int')
+        elif member['type']['kind'] == 'uint8':
+            packer_deps.append('pack_uint8')
+        elif member['type']['kind'] == 'be_uint16':
+            packer_deps.append('pack_be_uint16')
+        else:
+            raise ValueError(f"Error: unknown type {member['type']}")
+    res: str = ''
+    if len(packer_deps) > 0:
+        res += 'from twnet_parser.packer import ' + \
+            ', '.join(sorted(set(packer_deps))) + '\n'
+    if len(typing_deps) > 0:
+        res += 'from typing import ' + \
+            ', '.join(sorted(set(typing_deps))) + '\n'
+    return res
+
 def get_dependencies(msg: NetMessageJson) -> str:
     packer_deps: list[str] = []
     typing_deps: list[str] = ['Literal']
@@ -362,6 +468,31 @@ def gen_pack_return(msg: NetMessageJson) -> str:
     return f"        return {pack_field(members[0])} + \\\n" + \
             ' + \\\n'.join(mem_strs)
 
+def pack_field_connless7(member: NetConnlessMemberJson) -> str:
+    name: str = name_to_snake(member["name"])
+    field: str = f'self.{name}'
+    packer = 'int'
+    if member['type']['kind'] == 'packed_addresses': # TODO: packed
+        packer = 'int'
+    elif member['type']['kind'] == 'be_uint16':
+        packer = 'be_uint16'
+    elif member['type']['kind'] == 'uint8':
+        packer = 'uint8'
+    else:
+        raise ValueError(f"Error: unknown type {member['type']}")
+    return f'pack_{packer}({field})'
+
+def gen_pack_return_connless7(msg: NetConnlessJson) -> str:
+    members: list[NetConnlessMemberJson] = msg['members']
+    if len(members) == 0:
+        return "        return b''"
+    if len(members) == 1:
+        return f'        return {pack_field_connless7(members[0])}'
+    mem_strs: list[str] = [
+            f'            {pack_field_connless7(member)}' for member in members[1:]]
+    return f"        return {pack_field_connless7(members[0])} + \\\n" + \
+            ' + \\\n'.join(mem_strs)
+
 def get_default(field_path: str) -> Optional[str]:
     """
     field_path has the following format:
@@ -508,6 +639,48 @@ class CodeGenerator():
         out_file.write(',\n'.join(args) + '\n')
         out_file.write('    ) -> None:\n')
 
+    def generate_msg_connless(
+            self,
+            msg: NetConnlessJson
+    ) -> None:
+        name_snake = name_to_snake(msg['name'])
+        name_camel = name_to_camel(msg['name'])
+        dirname = os.path.dirname(__file__)
+        file_path= os.path.join(
+                dirname,
+                f'../twnet_parser/messages7/connless/',
+                f'{name_snake}.py')
+        with open(file_path, 'w') as out_file:
+            print(f"Generating {file_path} ...")
+            out_file.write('# generated by scripts/generate_messages.py\n')
+            out_file.write('\n')
+            out_file.write('from typing import Literal\n\n')
+            out_file.write('from twnet_parser.pretty_print import PrettyPrint\n')
+            if len(msg['members']) > 0:
+                out_file.write('from twnet_parser.packer import Unpacker\n')
+            out_file.write(get_dependencies_connless7(msg))
+            out_file.write('\n')
+            out_file.write(f'class Msg{name_camel}(PrettyPrint):\n')
+            out_file.write( \
+                '    def __init__(\n' \
+                '            self,\n' \
+                '    ) -> None:\n')
+            out_file.write(f"        self.message_type: Literal['connless'] = 'connless'\n")
+            out_file.write(f"        self.message_name: str = 'connless.{name_snake}'\n")
+            out_file.write(f"        self.message_id: list[int] = {msg['id']}\n")
+            out_file.write('\n')
+            out_file.write('    # first byte of data\n')
+            out_file.write('    # has to be the first byte of the message payload\n')
+            out_file.write('    # NOT the chunk header and NOT the message id\n')
+            out_file.write('    def unpack(self, data: bytes) -> bool:\n')
+            if len(msg['members']) > 0:
+                out_file.write('        unpacker = Unpacker(data)\n')
+            out_file.write(gen_unpack_members_connless7(msg))
+            out_file.write('        return True\n')
+            out_file.write('\n')
+            out_file.write('    def pack(self) -> bytes:\n')
+            out_file.write(gen_pack_return_connless7(msg))
+
     def generate_msg(
             self,
             msg: NetMessageJson,
@@ -669,13 +842,17 @@ class CodeGenerator():
             self.game_enums = spec_data['game_enumerations']
             game_messages: list[NetMessageJson] = spec_data['game_messages']
             system_messages: list[NetMessageJson] = spec_data['system_messages']
+            connless_messages: list[NetConnlessJson] = spec_data['connless_messages']
             self.gen_enum_file7()
             gen_match_file7('game', game_messages)
             gen_match_file7('system', system_messages)
+            gen_match_file_connless7(connless_messages)
             for msg in game_messages:
                 self.generate_msg(msg, 'game')
             for msg in system_messages:
                 self.generate_msg(msg, 'system')
+            for msg in connless_messages:
+                self.generate_msg_connless(msg)
 
 def main() -> None:
     dirname = os.path.dirname(__file__)

tests/unpacker_state_test.py

@@ -1,7 +1,45 @@
-from twnet_parser.packer import Unpacker, pack_int, pack_str
+from twnet_parser.packer import \
+        Unpacker, pack_int, pack_str, pack_uint8, pack_be_uint16
 from twnet_parser.packer import \
         NO_SANITIZE, SANITIZE, SANITIZE_CC, SKIP_START_WHITESPACES
 
+def test_pack_uint8():
+    assert pack_uint8(1) == b'\x01'
+    assert pack_uint8(2) == b'\x02'
+    assert pack_uint8(3) == b'\x03'
+
+def test_pack_be_uint16():
+    assert pack_be_uint16(1) == b'\x00\x01'
+    assert pack_be_uint16(2) == b'\x00\x02'
+    assert pack_be_uint16(3) == b'\x00\x03'
+    assert pack_be_uint16(256) == b'\x01\x00'
+
+def test_unpack_uint8():
+    u = Unpacker(b'\x01')
+    assert u.get_uint8() == 1
+    u = Unpacker(b'\x02')
+    assert u.get_uint8() == 2
+
+def test_unpack_multiple_uint8():
+    u = Unpacker(b'\x01\x02')
+    assert u.get_uint8() == 1
+    assert u.get_uint8() == 2
+
+def test_unpack_be_uint16():
+    u = Unpacker(b'\x00\x01')
+    assert u.get_be_uint16() == 1
+    u = Unpacker(b'\x00\x02')
+    assert u.get_be_uint16() == 2
+    u = Unpacker(b'\x00\x33')
+    assert u.get_be_uint16() == 51
+    u = Unpacker(b'\x01\x00')
+    assert u.get_be_uint16() == 256
+
+def test_unpack_multiple_be_uint16():
+    u = Unpacker(b'\x00\x01\x00\x02')
+    assert u.get_be_uint16() == 1
+    assert u.get_be_uint16() == 2
+
 def test_unpack_ints_and_strings() -> None:
     u = Unpacker(b'\x01\x02\x03\x01foo\x00bar\x00')
     assert u.get_int() == 1

twnet_parser/connless_message.py

@@ -0,0 +1,10 @@
+from typing import Protocol, Literal, Annotated
+
+class ConnlessMessage(Protocol):
+    message_type: Literal['connless']
+    message_name: str
+    message_id: Annotated[list[int], 8]
+    def unpack(self, data: bytes) -> bool:
+        ...
+    def pack(self) -> bytes:
+        ...

twnet_parser/ctrl_message.py

@@ -1,6 +1,7 @@
-from typing import Protocol
+from typing import Protocol, Literal
 
 class CtrlMessage(Protocol):
+    message_type: Literal['control']
     message_name: str
     message_id: int
     def unpack(self, data: bytes, we_are_a_client: bool = True) -> bool:

twnet_parser/messages7/connless/count.py

@@ -0,0 +1,27 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from twnet_parser.packer import Unpacker
+from twnet_parser.packer import pack_be_uint16
+from typing import Literal
+
+class MsgCount(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.count'
+        self.message_id: list[int] = [255, 255, 255, 255, 115, 105, 122, 50]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        unpacker = Unpacker(data)
+        self.count = unpacker.get_be_uint16()
+        return True
+
+    def pack(self) -> bytes:
+        return pack_be_uint16(self.count)

twnet_parser/messages7/connless/forward_check.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgForwardCheck(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.forward_check'
+        self.message_id: list[int] = [255, 255, 255, 255, 102, 119, 63, 63]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/messages7/connless/forward_error.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgForwardError(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.forward_error'
+        self.message_id: list[int] = [255, 255, 255, 255, 102, 119, 101, 114]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/messages7/connless/forward_ok.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgForwardOk(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.forward_ok'
+        self.message_id: list[int] = [255, 255, 255, 255, 102, 119, 111, 107]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/messages7/connless/forward_response.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgForwardResponse(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.forward_response'
+        self.message_id: list[int] = [255, 255, 255, 255, 102, 119, 33, 33]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/messages7/connless/heartbeat.py

@@ -0,0 +1,27 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from twnet_parser.packer import Unpacker
+from twnet_parser.packer import pack_be_uint16
+from typing import Literal
+
+class MsgHeartbeat(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.heartbeat'
+        self.message_id: list[int] = [255, 255, 255, 255, 98, 101, 97, 50]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        unpacker = Unpacker(data)
+        self.alt_port = unpacker.get_be_uint16()
+        return True
+
+    def pack(self) -> bytes:
+        return pack_be_uint16(self.alt_port)

twnet_parser/messages7/connless/list.py

@@ -0,0 +1,27 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from twnet_parser.packer import Unpacker
+from twnet_parser.packer import pack_int
+from typing import Literal
+
+class MsgList(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.list'
+        self.message_id: list[int] = [255, 255, 255, 255, 108, 105, 115, 50]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        unpacker = Unpacker(data)
+        self.servers = unpacker.get_int()
+        return True
+
+    def pack(self) -> bytes:
+        return pack_int(self.servers)

twnet_parser/messages7/connless/request_count.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgRequestCount(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.request_count'
+        self.message_id: list[int] = [255, 255, 255, 255, 99, 111, 117, 50]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/messages7/connless/request_info.py

@@ -0,0 +1,27 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from twnet_parser.packer import Unpacker
+from twnet_parser.packer import pack_uint8
+from typing import Literal
+
+class MsgRequestInfo(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.request_info'
+        self.message_id: list[int] = [255, 255, 255, 255, 103, 105, 101, 51]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        unpacker = Unpacker(data)
+        self.token = unpacker.get_uint8()
+        return True
+
+    def pack(self) -> bytes:
+        return pack_uint8(self.token)

twnet_parser/messages7/connless/request_list.py

@@ -0,0 +1,23 @@
+# generated by scripts/generate_messages.py
+
+from typing import Literal
+
+from twnet_parser.pretty_print import PrettyPrint
+from typing import Literal
+
+class MsgRequestList(PrettyPrint):
+    def __init__(
+            self,
+    ) -> None:
+        self.message_type: Literal['connless'] = 'connless'
+        self.message_name: str = 'connless.request_list'
+        self.message_id: list[int] = [255, 255, 255, 255, 114, 101, 113, 50]
+
+    # first byte of data
+    # has to be the first byte of the message payload
+    # NOT the chunk header and NOT the message id
+    def unpack(self, data: bytes) -> bool:
+        return True
+
+    def pack(self) -> bytes:
+        return b''

twnet_parser/msg7.py

@@ -1,3 +1,16 @@
+# connless
+
+REQUEST_LIST = [255, 255, 255, 255, 114, 101, 113, 50]
+LIST = [255, 255, 255, 255, 108, 105, 115, 50]
+REQUEST_COUNT = [255, 255, 255, 255, 99, 111, 117, 50]
+COUNT = [255, 255, 255, 255, 115, 105, 122, 50]
+REQUEST_INFO = [255, 255, 255, 255, 103, 105, 101, 51]
+HEARTBEAT = [255, 255, 255, 255, 98, 101, 97, 50]
+FORWARD_CHECK = [255, 255, 255, 255, 102, 119, 63, 63]
+FORWARD_RESPONSE = [255, 255, 255, 255, 102, 119, 33, 33]
+FORWARD_OK = [255, 255, 255, 255, 102, 119, 111, 107]
+FORWARD_ERROR = [255, 255, 255, 255, 102, 119, 101, 114]
+
 # control
 
 CTRL_KEEPALIVE = 0

twnet_parser/msg_matcher/connless7.py

@@ -0,0 +1,56 @@
+# generated by scripts/generate_messages.py
+from typing import Optional
+
+import twnet_parser.msg7
+from twnet_parser.connless_message import ConnlessMessage
+
+import twnet_parser.messages7.connless.request_list as \
+       connless_request_list
+import twnet_parser.messages7.connless.list as \
+       connless_list
+import twnet_parser.messages7.connless.request_count as \
+       connless_request_count
+import twnet_parser.messages7.connless.count as \
+       connless_count
+import twnet_parser.messages7.connless.request_info as \
+       connless_request_info
+import twnet_parser.messages7.connless.heartbeat as \
+       connless_heartbeat
+import twnet_parser.messages7.connless.forward_check as \
+       connless_forward_check
+import twnet_parser.messages7.connless.forward_response as \
+       connless_forward_response
+import twnet_parser.messages7.connless.forward_ok as \
+       connless_forward_ok
+import twnet_parser.messages7.connless.forward_error as \
+       connless_forward_error
+
+def match_connless7(msg_id: list[int], data: bytes) -> ConnlessMessage:
+    msg: Optional[ConnlessMessage] = None
+
+    if msg_id == twnet_parser.msg7.REQUEST_LIST:
+        msg = connless_request_list.MsgRequestList()
+    elif msg_id == twnet_parser.msg7.LIST:
+        msg = connless_list.MsgList()
+    elif msg_id == twnet_parser.msg7.REQUEST_COUNT:
+        msg = connless_request_count.MsgRequestCount()
+    elif msg_id == twnet_parser.msg7.COUNT:
+        msg = connless_count.MsgCount()
+    elif msg_id == twnet_parser.msg7.REQUEST_INFO:
+        msg = connless_request_info.MsgRequestInfo()
+    elif msg_id == twnet_parser.msg7.HEARTBEAT:
+        msg = connless_heartbeat.MsgHeartbeat()
+    elif msg_id == twnet_parser.msg7.FORWARD_CHECK:
+        msg = connless_forward_check.MsgForwardCheck()
+    elif msg_id == twnet_parser.msg7.FORWARD_RESPONSE:
+        msg = connless_forward_response.MsgForwardResponse()
+    elif msg_id == twnet_parser.msg7.FORWARD_OK:
+        msg = connless_forward_ok.MsgForwardOk()
+    elif msg_id == twnet_parser.msg7.FORWARD_ERROR:
+        msg = connless_forward_error.MsgForwardError()
+
+    if msg is None:
+        raise ValueError(f"Error: unknown conless  message id={msg_id} data={data[0]}")
+
+    msg.unpack(data)
+    return msg

twnet_parser/net_message.py

@@ -1,8 +1,9 @@
-from typing import Protocol
+from typing import Protocol, Literal
 
 from twnet_parser.chunk_header import ChunkHeader
 
 class NetMessage(Protocol):
+    message_type: Literal['system', 'game']
     message_name: str
     system_message: bool
     message_id: int

twnet_parser/packer.py

@@ -40,6 +40,18 @@ class Unpacker():
         self.idx = end
         return data
 
+    def get_uint8(self) -> int:
+        res = self.byte()
+        self.idx += 1
+        return res
+
+    def get_be_uint16(self) -> int:
+        left = self.byte()
+        self.idx += 1
+        right = self.byte()
+        self.idx += 1
+        return left << 8 | right
+
     def get_int(self) -> int:
         sign = (self.byte() >> 6) & 1
         res = self.byte() & 0x3F
@@ -121,6 +133,14 @@ def pack_int(num: int) -> bytes:
         num >>= 7 # discard 7 bits
     return bytes(res)
 
+def pack_be_uint16(num: int) -> bytes:
+    high = (num >> 8) & 0xff
+    low = num & 0xff
+    return bytes([high, low])
+
+def pack_uint8(num: int) -> bytes:
+    return bytes([num])
+
 def pack_str(data: str) -> bytes:
     return data.encode('utf-8') + b'\x00'