feat: implement to_json() for TwPacket

This commit is contained in:
ChillerDragon 2024-06-16 13:07:08 +08:00
parent 355699d33e
commit c3ff27f3b9
4 changed files with 84 additions and 2 deletions

View file

@ -25,6 +25,29 @@ print(packet.header) # => <class: 'Header'>: {'flags': <class: 'PacketFlags7, 's
print(packet.header.flags) # => <class: 'PacketFlags7'>: {'control': True, 'resend': False, 'compression': False, 'connless': False} print(packet.header.flags) # => <class: 'PacketFlags7'>: {'control': True, 'resend': False, 'compression': False, 'connless': False}
for msg in packet.messages: for msg in packet.messages:
print(msg.message_name) # => close print(msg.message_name) # => close
print(packet.to_json())
# {
# "version": "0.7",
# "payload_raw": "04",
# "payload_decompressed": "04",
# "header": {
# "flags": [
# "control"
# ],
# "ack": 10,
# "token": "cf2ede1d",
# "num_chunks": 0
# },
# "messages": [
# {
# "message_type": "control",
# "message_name": "close",
# "message_id": 4,
# "reason": null
# }
# ]
# }
``` ```
More examples can be found in the [examples/](./examples/) folder: More examples can be found in the [examples/](./examples/) folder:

44
tests/json_test.py Normal file
View file

@ -0,0 +1,44 @@
from typing import cast
import textwrap
from twnet_parser.messages6.game.cl_change_info import MsgClChangeInfo
from twnet_parser.packet import TwPacket, NetMessage
def test_change_info_packet() -> None:
packet = TwPacket()
packet.messages.append(cast(NetMessage, MsgClChangeInfo()))
packet = packet.to_json()
expected = textwrap.dedent("""\
{
"version": "0.7",
"payload_raw": "",
"payload_decompressed": "",
"header": {
"flags": [],
"ack": 0,
"token": "ffffffff",
"num_chunks": null
},
"messages": [
{
"message_type": "game",
"message_name": "cl_change_info",
"system_message": false,
"message_id": 21,
"header": {
"version": "0.6",
"flags": [],
"size": null,
"seq": -1
},
"name": "default",
"clan": "default",
"country": 0,
"skin": "default",
"use_custom_color": false,
"color_body": 0,
"color_feet": 0
}
]
}""")
assert packet == expected

View file

@ -1,6 +1,7 @@
#!/usr/bin/env python #!/usr/bin/env python
from typing import Union, cast, Optional, Literal from typing import Union, cast, Optional, Literal, Iterator, Any
import json
from twnet_parser.packer import Unpacker, pack_int from twnet_parser.packer import Unpacker, pack_int
from twnet_parser.pretty_print import PrettyPrint from twnet_parser.pretty_print import PrettyPrint
@ -16,6 +17,7 @@ from twnet_parser.msg_matcher.connless6 import match_connless6
from twnet_parser.constants import NET_MAX_SEQUENCE, NET_PACKETVERSION from twnet_parser.constants import NET_MAX_SEQUENCE, NET_PACKETVERSION
import twnet_parser.huffman import twnet_parser.huffman
import twnet_parser.serialize
# TODO: what is a nice pythonic way of storing those? # TODO: what is a nice pythonic way of storing those?
# also does some version:: namespace thing make sense? # also does some version:: namespace thing make sense?
@ -257,7 +259,7 @@ class TwPacket(PrettyPrint):
raise ValueError(f"Error: invalid packet version '{self.version}'") raise ValueError(f"Error: invalid packet version '{self.version}'")
self.messages: list[Union[CtrlMessage, NetMessage, ConnlessMessage]] = [] self.messages: list[Union[CtrlMessage, NetMessage, ConnlessMessage]] = []
def __iter__(self): def __iter__(self) -> Iterator[tuple[str, Any]]:
yield 'version', self.version yield 'version', self.version
yield 'payload_raw', self.payload_raw yield 'payload_raw', self.payload_raw
yield 'payload_decompressed', self.payload_decompressed yield 'payload_decompressed', self.payload_decompressed
@ -265,6 +267,14 @@ class TwPacket(PrettyPrint):
yield 'messages', [dict(msg) for msg in self.messages] yield 'messages', [dict(msg) for msg in self.messages]
def to_json(self) -> str:
return json.dumps(
dict(self),
indent=2,
sort_keys=False,
default=twnet_parser.serialize.bytes_to_hex
)
@property @property
def version(self) -> Literal['0.6', '0.7']: def version(self) -> Literal['0.6', '0.7']:
return self._version return self._version

View file

@ -0,0 +1,5 @@
def bytes_to_hex(obj):
if isinstance(obj, bytes):
return obj.hex()
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")