• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import io
2import textwrap
3from typing import Dict
4import vulkan_printer
5
6
7class CommandPrinter:
8    """This class is responsible for printing the commands found in the minidump file to the terminal."""
9
10    def __init__(self, opcode: int, original_size: int, data: bytes, stream_idx: int, cmd_idx: int):
11        self.opcode = opcode
12        self.original_size = original_size
13        self.data = io.BytesIO(data)
14        self.stream_idx = stream_idx
15        self.cmd_idx = cmd_idx
16
17    def print_cmd(self):
18        """
19        Tries to decode and pretty print the command to the terminal.
20        Falls back to printing hex data if the command doesn't have a printer.
21        """
22
23        # Print out the command name
24        print("\n{}.{} - {}: ({} bytes)".format(self.stream_idx, self.cmd_idx, self.cmd_name(), self.original_size - 8))
25
26        pretty_printer = getattr(vulkan_printer, self.cmd_name(), None)
27        if not pretty_printer:
28            self.print_raw()
29            return
30
31        try:
32            pretty_printer(self, indent=4)
33            # Check that we processed all the bytes, otherwise there's probably a bug in the pretty printing logic
34            if self.data.tell() != len(self.data.getbuffer()):
35                raise BufferError(
36                    "Not all data was decoded. Decoded {} bytes but expected {}".format(
37                        self.data.tell(), len(self.data.getbuffer())))
38        except Exception as ex:
39            print("Error while processing {}: {}".format(self.cmd_name(), repr(ex)))
40            print("Command raw data:")
41            self.print_raw()
42            raise ex
43
44    def cmd_name(self) -> str:
45        """Returns the command name (e.g.: "OP_vkBeginCommandBuffer", or the opcode as a string if unknown"""
46        return vulkan_printer.opcodes.get(self.opcode, str(self.opcode))
47
48    def print_raw(self):
49        """Prints the command data as a hex bytes, as a fallback if we don't know how to decode it"""
50        truncated = self.original_size > len(self.data.getbuffer()) + 8
51        indent = 8
52        hex = ' '.join(["{:02x}".format(x) for x in self.data.getbuffer()])
53        if truncated:
54            hex += " [...]"
55        lines = textwrap.wrap(hex, width=16 * 3 + indent, initial_indent=' ' * indent, subsequent_indent=' ' * indent)
56        for l in lines:
57            print(l)
58
59    def read_int(self, num_bytes: int, signed: bool = False) -> int:
60        assert num_bytes == 4 or num_bytes == 8
61        buf = self.data.read(num_bytes)
62        if len(buf) != num_bytes:
63            raise EOFError("Unexpectly reached the end of the buffer")
64        return int.from_bytes(buf, byteorder='little', signed=signed)
65
66    def write(self, msg: str, indent: int):
67        """Prints a string at a given indentation level"""
68        assert type(msg) == str
69        assert type(indent) == int and indent >= 0
70        print("  " * indent + msg, end='')
71
72    def write_int(self,
73                  field_name: str,
74                  num_bytes: int,
75                  indent: int,
76                  signed: bool = False,
77                  value: int = None):
78        """Reads the next 32 or 64 bytes integer from the data stream and prints it"""
79        if value is None:
80            value = self.read_int(num_bytes, signed)
81        self.write("{}: {}\n".format(field_name, value), indent)
82
83    def write_enum(self, field_name: str, enum: Dict[int, str], indent: int, value: int = None):
84        """Reads the next 32-byte int from the data stream and prints it as an enum"""
85        if value is None:
86            value = self.read_int(4)
87        self.write("{}: {} ({})\n".format(field_name, enum.get(value, ""), value), indent)
88
89    def write_stype_and_pnext(self, expected_stype: str, indent: int):
90        """Reads and prints the sType and pNext fields found in many Vulkan structs, while also sanity checking them"""
91        stype = self.read_int(4)
92        self.write_enum("sType", vulkan_printer.VkStructureType, indent, value=stype)
93        if vulkan_printer.VkStructureType.get(stype) != expected_stype:
94            raise ValueError("Wrong sType while decoding data. Expected: " + expected_stype)
95
96        pnext_size = self.read_int(4)
97        self.write_int("pNextSize", 4, indent, value=pnext_size)
98        if pnext_size != 0:
99            raise NotImplementedError("Decoding structs with pNextSize > 0 not supported")
100
101    def write_struct(self, field_name: str, struct_fn, indent: int):
102        """Reads and prints a struct, calling `struct_fn` to pretty-print it"""
103        self.write("{}:\n".format(field_name), indent)
104        struct_fn(self, indent + 1)
105
106    def write_repeated(self,
107                       count_name: str,
108                       field_name: str,
109                       struct_fn,
110                       indent: int,
111                       pointer_name: str = None):
112        """
113        Reads and prints repeated structs, with a 32-byte count field followed by the struct data.
114        If pointer_name is not None, reads an additional 64-bit pointer within the repeated block
115        before reading repeated data.
116        """
117        count = self.read_int(4)
118        if pointer_name is not None:
119            self.write_int(pointer_name, 8, indent)
120        assert count < 1000, "count too large: {}".format(count)  # Sanity check that we haven't read garbage data
121        self.write_int(count_name, 4, indent, value=count)
122        for i in range(0, count):
123            self.write_struct("{} #{}".format(field_name, i), struct_fn, indent)
124