LEGO_Universe/utils
1
0
Fork 0
mirror of https://github.com/lcdr/utils.git synced 2024-08-30 17:32:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactored BitStream into read-only and write-only versions.

Browse Source
This commit is contained in:
lcdr 2017-12-25 13:43:33 +01:00
parent 20db28b0a2
commit 54eccf854f
5 changed files with 26 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
captureviewer.pyw
View File

@ -18,7 +18,7 @@ import amf3
import viewer import viewer
import ldf import ldf
from structparser import StructParser from structparser import StructParser
from pyraknet.bitstream import BitStream, c_bit, c_bool, c_float, c_int, c_int64, c_ubyte, c_uint, c_uint64, c_ushort from pyraknet.bitstream import c_bit, c_bool, c_float, c_int, c_int64, c_ubyte, c_uint, c_uint64, c_ushort, ReadStream
component_name = OrderedDict() component_name = OrderedDict()
component_name[108] = "Component 108", component_name[108] = "Component 108",
@ -202,28 +202,28 @@ class CaptureViewer(viewer.Viewer):
print("Parsing creations") print("Parsing creations")
creations = [i for i in files if "[24]" in i] creations = [i for i in files if "[24]" in i]
for packet_name in creations: for packet_name in creations:
packet = BitStream(capture.read(packet_name)) packet = ReadStream(capture.read(packet_name))
self.parse_creation(packet_name, packet) self.parse_creation(packet_name, packet)
if self.parse_serializations.get(): if self.parse_serializations.get():
print("Parsing serializations") print("Parsing serializations")
serializations = [i for i in files if "[27]" in i] serializations = [i for i in files if "[27]" in i]
for packet_name in serializations: for packet_name in serializations:
packet = BitStream(capture.read(packet_name)[1:]) packet = ReadStream(capture.read(packet_name)[1:])
self.parse_serialization_packet(packet_name, packet) self.parse_serialization_packet(packet_name, packet)
if self.parse_game_messages.get(): if self.parse_game_messages.get():
print("Parsing game messages") print("Parsing game messages")
game_messages = [i for i in files if "[53-05-00-0c]" in i or "[53-04-00-05]" in i] game_messages = [i for i in files if "[53-05-00-0c]" in i or "[53-04-00-05]" in i]
for packet_name in game_messages: for packet_name in game_messages:
packet = BitStream(capture.read(packet_name)[8:]) packet = ReadStream(capture.read(packet_name)[8:])
self.parse_game_message(packet_name, packet) self.parse_game_message(packet_name, packet)
if self.parse_normal_packets.get(): if self.parse_normal_packets.get():
print("Parsing normal packets") print("Parsing normal packets")
packets = [i for i in files if "[24]" not in i and "[27]" not in i and "[53-05-00-0c]" not in i and "[53-04-00-05]" not in i] packets = [i for i in files if "[24]" not in i and "[27]" not in i and "[53-05-00-0c]" not in i and "[53-04-00-05]" not in i]
for packet_name in packets: for packet_name in packets:
packet = BitStream(capture.read(packet_name)) packet = ReadStream(capture.read(packet_name))
self.parse_normal_packet(packet_name, packet) self.parse_normal_packet(packet_name, packet)
def object_id_handler(self, stream): def object_id_handler(self, stream):
@ -254,7 +254,7 @@ class CaptureViewer(viewer.Viewer):
assert len(uncompressed) == uncompressed_size assert len(uncompressed) == uncompressed_size
else: else:
uncompressed = stream.read(bytes, length=size) uncompressed = stream.read(bytes, length=size)
return ldf.from_ldf(BitStream(uncompressed)) return ldf.from_ldf(ReadStream(uncompressed))
def parse_creation(self, packet_name, packet, retry_with_components=[]): def parse_creation(self, packet_name, packet, retry_with_components=[]):
packet.skip_read(1) packet.skip_read(1)
@ -294,7 +294,7 @@ class CaptureViewer(viewer.Viewer):
else: else:
lot_name, parsers, error = self.lot_data[lot] lot_name, parsers, error = self.lot_data[lot]
id_ = packet.read(str, length_type=c_ubyte) + " " + lot_name id_ = packet.read(str, length_type=c_ubyte) + " " + lot_name
packet._read_offset = 0 packet.read_offset = 0
parser_output = ParserOutput() parser_output = ParserOutput()
with parser_output: with parser_output:
parser_output.append(self.creation_header_parser.parse(packet)) parser_output.append(self.creation_header_parser.parse(packet))
@ -319,7 +319,7 @@ class CaptureViewer(viewer.Viewer):
if retry_with_components: if retry_with_components:
print("retrying with", retry_with_components, packet_name) print("retrying with", retry_with_components, packet_name)
del self.lot_data[lot] del self.lot_data[lot]
packet._read_offset = 0 packet.read_offset = 0
self.parse_creation(packet_name, packet, retry_with_components) self.parse_creation(packet_name, packet, retry_with_components)
return return
@ -333,7 +333,7 @@ class CaptureViewer(viewer.Viewer):
parser_output.text += "\n"+name+"\n\n" parser_output.text += "\n"+name+"\n\n"
parser_output.append(parser.parse(packet, {"creation":is_creation})) parser_output.append(parser.parse(packet, {"creation":is_creation}))
if not packet.all_read(): if not packet.all_read():
raise IndexError("Not completely read, %i bytes unread" % (len(packet) - math.ceil(packet._read_offset / 8))) raise IndexError("Not completely read, %i bytes unread" % (len(packet) - math.ceil(packet.read_offset / 8)))
def parse_serialization_packet(self, packet_name, packet): def parse_serialization_packet(self, packet_name, packet):
network_id = packet.read(c_ushort) network_id = packet.read(c_ushort)

4
ldf.py
View File

@ -1,8 +1,8 @@
from pyraknet.bitstream import BitStream, c_bool, c_float, c_int, c_int64, c_ubyte, c_uint from pyraknet.bitstream import c_bool, c_float, c_int, c_int64, c_ubyte, c_uint, ReadStream
def from_ldf(ldf): def from_ldf(ldf):
ldf_dict = {} ldf_dict = {}
if isinstance(ldf, BitStream): if isinstance(ldf, ReadStream):
for _ in range(ldf.read(c_uint)): for _ in range(ldf.read(c_uint)):
encoded_key = ldf.read(bytes, length=ldf.read(c_ubyte)) encoded_key = ldf.read(bytes, length=ldf.read(c_ubyte))
key = encoded_key.decode("utf-16-le") key = encoded_key.decode("utf-16-le")

18
luzviewer.pyw
View File

@ -9,7 +9,7 @@ import tkinter.messagebox as messagebox
from tkinter import END, Menu from tkinter import END, Menu
import viewer import viewer
from pyraknet.bitstream import BitStream, c_bool, c_float, c_int, c_int64, c_ubyte, c_uint, c_uint64, c_ushort from pyraknet.bitstream import c_bool, c_float, c_int, c_int64, c_ubyte, c_uint, c_uint64, c_ushort, ReadStream
class PathType(enum.IntEnum): class PathType(enum.IntEnum):
Movement = 0 Movement = 0
@ -53,7 +53,7 @@ class LUZViewer(viewer.Viewer):
self.tree.set_children("") self.tree.set_children("")
print("Loading", luz_path) print("Loading", luz_path)
with open(luz_path, "rb") as file: with open(luz_path, "rb") as file:
stream = BitStream(file.read()) stream = ReadStream(file.read())
version = stream.read(c_uint) version = stream.read(c_uint)
assert version in (36, 38, 39, 40, 41), version assert version in (36, 38, 39, 40, 41), version
@ -85,7 +85,7 @@ class LUZViewer(viewer.Viewer):
with open(lvl_path, "rb") as lvl: with open(lvl_path, "rb") as lvl:
print("Loading lvl", filename) print("Loading lvl", filename)
try: try:
self.parse_lvl(BitStream(lvl.read()), scene) self.parse_lvl(ReadStream(lvl.read()), scene)
except Exception: except Exception:
import traceback import traceback
traceback.print_exc() traceback.print_exc()
@ -115,7 +115,7 @@ class LUZViewer(viewer.Viewer):
self.tree.insert(scene_transition, END, text="Transition Point", values=(transition_point_scene_id, transition_point_position)) self.tree.insert(scene_transition, END, text="Transition Point", values=(transition_point_scene_id, transition_point_position))
remaining_length = stream.read(c_uint) remaining_length = stream.read(c_uint)
assert len(stream) - stream._read_offset//8 == remaining_length assert len(stream) - stream.read_offset//8 == remaining_length
assert stream.read(c_uint) == 1 assert stream.read(c_uint) == 1
### paths ### paths
@ -231,16 +231,16 @@ class LUZViewer(viewer.Viewer):
# newer lvl file structure # newer lvl file structure
# chunk based # chunk based
while not stream.all_read(): while not stream.all_read():
assert stream._read_offset//8 % 16 == 0 # seems everything is aligned like this? assert stream.read_offset//8 % 16 == 0 # seems everything is aligned like this?
start_pos = stream._read_offset//8 start_pos = stream.read_offset//8
assert stream.read(bytes, length=4) == b"CHNK" assert stream.read(bytes, length=4) == b"CHNK"
chunk_type = stream.read(c_uint) chunk_type = stream.read(c_uint)
assert stream.read(c_ushort) == 1 assert stream.read(c_ushort) == 1
assert stream.read(c_ushort) in (1, 2) assert stream.read(c_ushort) in (1, 2)
chunk_length = stream.read(c_uint) chunk_length = stream.read(c_uint)
data_pos = stream.read(c_uint) data_pos = stream.read(c_uint)
stream._read_offset = data_pos * 8 stream.read_offset = data_pos * 8
assert stream._read_offset//8 % 16 == 0 assert stream.read_offset//8 % 16 == 0
if chunk_type == 1000: if chunk_type == 1000:
pass pass
elif chunk_type == 2000: elif chunk_type == 2000:
@ -249,7 +249,7 @@ class LUZViewer(viewer.Viewer):
self.lvl_parse_chunk_type_2001(stream, scene) self.lvl_parse_chunk_type_2001(stream, scene)
elif chunk_type == 2002: elif chunk_type == 2002:
pass pass
stream._read_offset = (start_pos + chunk_length) * 8 # go to the next CHNK stream.read_offset = (start_pos + chunk_length) * 8 # go to the next CHNK
else: else:
self.parse_old_lvl_header(stream) self.parse_old_lvl_header(stream)
self.lvl_parse_chunk_type_2001(stream, scene) self.lvl_parse_chunk_type_2001(stream, scene)

6
pkviewer.pyw
View File

@ -10,7 +10,7 @@ from tkinter.ttk import Entry, Scrollbar, Treeview
import decompress_sd0 import decompress_sd0
import viewer import viewer
from pyraknet.bitstream import BitStream, c_bool, c_int, c_ubyte, c_uint from pyraknet.bitstream import c_bool, c_int, c_ubyte, c_uint, ReadStream
class PKViewer(viewer.Viewer): class PKViewer(viewer.Viewer):
def __init__(self): def __init__(self):
@ -79,7 +79,7 @@ class PKViewer(viewer.Viewer):
def load_pki(self, path): def load_pki(self, path):
# unused, alternate way to get the list of pks # unused, alternate way to get the list of pks
with open(path, "rb") as file: with open(path, "rb") as file:
stream = BitStream(file.read()) stream = ReadStream(file.read())
assert stream.read(c_uint) == 3 assert stream.read(c_uint) == 3
pack_files = [] pack_files = []
@ -101,7 +101,7 @@ class PKViewer(viewer.Viewer):
if unknown != 0: if unknown != 0:
print(unknown, path) print(unknown, path)
file.seek(number_of_records_address) file.seek(number_of_records_address)
data = BitStream(file.read()[:-8]) data = ReadStream(file.read()[:-8])
number_of_records = data.read(c_uint) number_of_records = data.read(c_uint)
for _ in range(number_of_records): for _ in range(number_of_records):

6
structparser.py
View File

@ -5,7 +5,7 @@ import argparse
import re import re
from collections import namedtuple from collections import namedtuple
from pyraknet.bitstream import BitStream, c_bit, c_float, c_double, c_int8, c_uint8, c_int16, c_uint16, c_int32, c_uint32, c_int64, c_uint64 from pyraknet.bitstream import c_bit, c_float, c_double, c_int8, c_uint8, c_int16, c_uint16, c_int32, c_uint32, c_int64, c_uint64, ReadStream
VAR_CHARS = r"[^ \t\[\]]+" VAR_CHARS = r"[^ \t\[\]]+"
@ -88,10 +88,10 @@ class StructParser:
if variables is None: if variables is None:
variables = {} variables = {}
self._variables = variables self._variables = variables
if isinstance(data, BitStream): if isinstance(data, ReadStream):
stream = data stream = data
else: else:
stream = BitStream(data) stream = ReadStream(data)
yield from self._parse_struct_occurrences(stream, self.defs) yield from self._parse_struct_occurrences(stream, self.defs)
def _to_tree(self, def_iter, stack_level=0, start_def=None): def _to_tree(self, def_iter, stack_level=0, start_def=None):