bta-proxy/bta_proxy/debug.py

from collections.abc import Iterable

from bta_proxy.packets.base import Packet
from bta_proxy.packets import *
from .datainputstream import DataInputStream
from typing import Generator, TextIO, TypeVar

T = TypeVar('T')

def chunks(gen: Iterable[T], size: int) -> Generator[list[T], None, None]:
    bucket: list[T] = []
    for item in gen:
        bucket.append(item)
        if len(bucket) >= size:
            yield bucket
            bucket.clear()
    if bucket:
        yield bucket

def debug_client(buffer: bytes, tmpfile: TextIO):
    stream = DataInputStream(buffer)

    while not stream.empty():
        try:
            packet = Packet.parse_packet(stream)
            match packet.packet_id:
                case _:
                    print('[C]', packet)
        except ValueError:
            # print(f'[C:rest] {stream.end()}')
            buf = stream.end()
            print(f"[C] {buf[0]=} {len(buf)=}, {buf=}", file=tmpfile)


def debug_server(buffer: bytes, tmpfile: TextIO):
    stream = DataInputStream(buffer)

    while not stream.empty():
        try:
            packet = Packet.parse_packet(stream)
            match packet.packet_id:
                case Packet50PreChunk.packet_id:
                    continue
                case Packet38EntityStatus.packet_id:
                    continue
                case _:
                    print('[S]', packet)
        except ValueError:
            # print(f'[S:rest] {stream.end()}')
            buf = stream.end()
            print(f"[S] {buf[0]=} {len(buf)=}, {buf=}", file=tmpfile)