-
Notifications
You must be signed in to change notification settings - Fork 2
BMG #39
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
BMG #39
Changes from all commits
1288b8d
84d85b5
90d995c
b0b6426
21954fb
6d1a560
23f0e91
4061340
e6a6bae
d3a53d4
bb139ff
104f2ea
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| *.iso | ||
| .idea | ||
| *.arc | ||
| **/__pycache__ | ||
|
|
||
| /src_dir/ | ||
| /assets | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,15 +15,26 @@ def from_bytes(cls, data: bytes) -> "BinaryReader": | |
| stream = BytesIO(data) | ||
| return cls(stream) | ||
|
|
||
| def seek(self, offset: int) -> None: | ||
| self.stream.seek(offset) | ||
| def seek(self, offset: int) -> int: | ||
| return self.stream.seek(offset) | ||
|
|
||
| def size(self) -> int: | ||
| current_position = self.tell() | ||
| size = self.stream.seek(0, 2) | ||
| self.seek(current_position) | ||
| return size | ||
|
|
||
| def tell(self) -> int: | ||
| return self.stream.tell() | ||
|
|
||
| def skip(self, count: int) -> None: | ||
| self.stream.read(count) | ||
|
|
||
| def back(self, count: int) -> int: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. You are using it one time and it's for backing and reread node size. You could read node_size and getting the type from the node_byte. And it clamp when at 0 without any errors Remove it from the PR |
||
| if count > self.stream.tell(): | ||
| return self.stream.seek(0) | ||
| return self.stream.seek(-count, 1) | ||
|
|
||
| def _read_number(self, size: int, unpack_fmt: str) -> int: | ||
| data = self.stream.read(size) | ||
| if len(data) != size: | ||
|
|
@@ -94,4 +105,4 @@ def string_until_null(self, encoding: str | None = None) -> str: | |
| break | ||
| chars += byte | ||
|
|
||
| return chars.decode(encoding) | ||
| return chars.decode(encoding) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| from io import BytesIO | ||
| from typing import BinaryIO | ||
|
|
||
| from wiithon.binary.reader import BinaryReader | ||
| from wiithon.binary.writer import BinaryWriter | ||
| from wiithon.formats.bmg_sections.bmg_section import BMGSection | ||
| from wiithon.formats.bmg_sections.inf1 import INF1Section | ||
| from wiithon.formats.bmg_sections.dat1 import DAT1Section | ||
| from wiithon.formats.bmg_sections.flw1 import FLW1Section | ||
| from wiithon.formats.bmg_sections.fli1 import FLI1Section | ||
|
|
||
| DATA_MAGIC = "MESG" | ||
| FILE_MAGIC = "bmg1" | ||
|
|
||
| class BMG: | ||
| """ | ||
| BMG (Binary Message Data) file handler for parsing and exporting binary message data. | ||
| The BMG class manages the structure of BMG files which contain multiple sections | ||
| (INF1, DAT1, FLW1, FLI1) that store message information and data. | ||
| Attributes: | ||
| section_count (int): Number of sections in the BMG file. | ||
| sections (list[bmg_section]): List of parsed section objects. | ||
| flw1_section_offset (int): Offset to the FLW1 section in the file. | ||
| unknown (int): Unknown single byte value from file header. | ||
| Methods: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Docstring methods needs to be under method, not under class (but at least, you have docstring like my code....) Also, section: bmg_section is the module, not the class |
||
| __init__(raw_bytes: BytesIO) -> None: | ||
| Parses a BMG file from raw bytes. Validates magic numbers and reads | ||
| all sections from the file. | ||
| add_header(section: bmg_section) -> BytesIO: | ||
| Wraps a section with its BMG header (magic and size) and applies | ||
| 32-byte alignment padding. Returns the complete section data. | ||
| export_bmg() -> BytesIO: | ||
| Reconstructs the complete BMG file from the current sections list. | ||
| Rebuilds the header and all sections with proper formatting and padding. | ||
| Returns the complete BMG file as bytes. | ||
| """ | ||
| section_count: int | ||
| sections: list[BMGSection] | ||
|
|
||
| def __init__(self, raw_bytes: BinaryIO): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The init need to have a "empty" body just to initalize fields. All the reading stuff goes to a |
||
| reader = BinaryReader(raw_bytes) | ||
| data_magic = reader.string(0x4) | ||
| assert data_magic == DATA_MAGIC | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Remove assert, throw exception instead |
||
|
|
||
| file_magic = reader.string(0x4) | ||
| assert file_magic == FILE_MAGIC | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Assert -> exceptions |
||
|
|
||
| self.flw1_section_offset = reader.u32() | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This name is partially false. In Mario Kart Wii, it's the dataSize of the file. It's seems correct in SMG though |
||
| self.section_count = reader.u32() | ||
| self.unknown = reader.u8() | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It's not unknown, it's the encoding. seems to be UTF-16 in SMG for example |
||
| reader.seek(0x20) | ||
|
|
||
| self.sections = [] | ||
|
|
||
| for section in range(self.section_count): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. section will be shadowed in the match line 67-73 |
||
| section_magic = reader.string(0x4) | ||
| section_size = reader.u32() - 0x8 | ||
|
|
||
| # Take into account the removed padding at the end of the file | ||
| if section_size > reader.size() - reader.tell(): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Probably better to test some stuffs:
|
||
| section_size = reader.size() - reader.tell() | ||
|
|
||
| section_bytes = reader.raw(section_size) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If the size is below 0x8, it became a negative number and the reader will read everything I don't know if it's possible but MKWii is slightly different, so it's better to test if section_size > 0 |
||
| section_bytes = BytesIO(section_bytes) | ||
|
|
||
| match section_magic: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Add a default case here, MKWii uses MID1 for example For the default case, maybe adding a RawSection(BMGSection) with raw data can be useful |
||
| case "INF1": | ||
| section = INF1Section.import_section(section_bytes) | ||
| case "DAT1": | ||
| section = DAT1Section.import_section(section_bytes) | ||
| case "FLW1": | ||
| section = FLW1Section.import_section(section_bytes) | ||
| case "FLI1": | ||
| section = FLI1Section.import_section(section_bytes) | ||
|
|
||
| self.sections.append(section) | ||
|
|
||
| def add_header(self, section: BMGSection) -> BinaryIO: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Better to be private
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Just thought about that by re-reading my comment: moving this to BMGSection |
||
| total_bytes = BytesIO() | ||
| writer = BinaryWriter(total_bytes) | ||
|
|
||
| section_bytes = section.export_section() | ||
| section_size = section_bytes.seek(0, 2) + 0x8 | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
|
|
||
| padding = 0 | ||
| if section_size % 32: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. There is an align method in wiithon.binary.align |
||
| padding = 32 - section_size % 32 | ||
| section_size += padding | ||
|
|
||
| writer.string(section.magic, 0x4) | ||
| writer.u32(section_size) | ||
| writer.raw(section_bytes.read) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
btw, the read() will probably have some issues, better to use .getvalue() to not move the cursor |
||
| writer.pad(padding) # should be align(0x20) | ||
|
|
||
| return total_bytes | ||
|
|
||
| def get_section(self, section_magic: str) -> list[BMGSection]: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| out: list[BMGSection] = [] | ||
|
|
||
| for section in self.sections: | ||
| if section.magic == section_magic: | ||
| out.append(section) | ||
|
|
||
| return out | ||
|
|
||
| def export_bmg(self) -> BinaryIO: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| bmg_bytes = BytesIO() | ||
| writer = BinaryWriter(bmg_bytes) | ||
|
|
||
| writer.string(DATA_MAGIC) | ||
| writer.string(FILE_MAGIC) | ||
| writer.u32(0) # Write the flw1_section_offset later | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. MKWii doesn't have this field. It's the file length (so the if line 118 is useless for them) |
||
| writer.u32(len(self.sections)) | ||
| writer.u8(self.unknown) | ||
| writer.seek(0x20) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Use |
||
|
|
||
| for section in self.sections: | ||
| if section.magic == "FLW1": | ||
| position = writer.tell() | ||
| writer.seek(0x8) | ||
| writer.u32(position) | ||
| writer.seek(position) | ||
|
|
||
| section_bytes = self.add_header(section) | ||
| writer.raw(section_bytes.read()) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The cursor of the stream is in EOF, use |
||
|
|
||
| return bmg_bytes | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| from abc import ABC, abstractmethod | ||
| from typing import BinaryIO | ||
|
|
||
| class BMGSection(ABC): | ||
| """ | ||
| Base class for BMG file sections. | ||
| Provides the interface for importing and exporting binary section data. | ||
| Subclasses must override import_section() and export_section() methods. | ||
| Attributes: | ||
| magic (str): The magic identifier for this section type. | ||
| """ | ||
| magic: str | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Use a ClassVar here |
||
|
|
||
| def __init__(self, magic: str): | ||
| self.magic = magic | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. All subclasses have magic in their files, passing it to the constructor is useless |
||
|
|
||
| @classmethod | ||
| @abstractmethod | ||
| def import_section(cls, raw_data: BinaryIO) -> "BMGSection": | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Read/write like everywhere |
||
| """ | ||
| Import a section from raw bytes. | ||
| This method must be overridden in subclasses to provide proper implementation. | ||
| Raises NotImplementedError: If not properly overridden in a subclass. | ||
| """ | ||
| raise NotImplementedError("Import section is not implemented") | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. No need, it's an abstract method, it's a dead code |
||
|
|
||
| @abstractmethod | ||
| def export_section(self) -> BinaryIO: | ||
| """ | ||
| Export a section from raw bytes. | ||
| This method must be overridden in subclasses to provide proper implementation. | ||
| Raises NotImplementedError: If not properly overridden in a subclass. | ||
| """ | ||
| raise NotImplementedError("Export section is not implemented") | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| from io import BytesIO | ||
| from enum import IntEnum | ||
| from typing import BinaryIO, NamedTuple | ||
|
|
||
| from wiithon.binary.reader import BinaryReader | ||
| from wiithon.binary.writer import BinaryWriter | ||
| from wiithon.formats.bmg_sections.bmg_section import BMGSection | ||
|
|
||
| DAT1_MAGIC: str = "DAT1" | ||
| TAG_IDENTIFIER = b"\x00\x1A" | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Well, it's true for SMG and MKWii because the format is utf-16, but not true for all encoding. Same for NULL_BYTE below |
||
| NULL_BYTE = b"\x00\x00" | ||
|
|
||
| class TagIdentifier(IntEnum): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It's more TagGroup than identifier. |
||
| """ | ||
| Identifies when a tag/action will be used when the console displays this in game. | ||
| This can range from delaying more text from appearing, playing a sound effect, coloring text, etc. | ||
| """ | ||
| delay = 0x01 | ||
| sound_effect = 0x02 | ||
| load_image = 0x03 | ||
| unknown1 = 0x04 | ||
| unknown2 = 0x05 | ||
| unknown3 = 0x06 | ||
| unknown4 = 0x07 | ||
| unknown5 = 0x08 | ||
| unknown6 = 0x09 | ||
| colour_text = 0xFF | ||
|
|
||
| class Tag: | ||
| """ | ||
| Container object for one or more tags that will be used within a Message object. | ||
| """ | ||
| def __init__(self, | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. read/write |
||
| offset: int, | ||
| size: int, | ||
| identifier: TagIdentifier, | ||
| data: bytes = None): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. bytes | None |
||
|
|
||
| if not isinstance(identifier, int) and not isinstance(identifier, TagIdentifier): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. See my comment in inf1.py line 46 |
||
| raise Exception("Bad Input") | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Better exception (and message) |
||
|
|
||
| self.offset: int = offset | ||
| self.size: int = size | ||
| self.identifier: TagIdentifier = TagIdentifier(identifier) | ||
| self.data = data | ||
|
|
||
| @classmethod | ||
| def import_tag(cls, raw_bytes: BinaryIO, offset: int) -> "Tag": | ||
| reader = BinaryReader(raw_bytes) | ||
| size = reader.u8() | ||
| identifier = reader.u8() | ||
| data = reader.raw(size - 4) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Test is size - 4 > 0 |
||
|
|
||
| return cls(offset, size, identifier, data) | ||
|
|
||
| def export_tag(self) -> BinaryIO: | ||
| assert isinstance(self.data, bytes) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| tag_bytes: BytesIO = BytesIO() | ||
| writer = BinaryWriter(tag_bytes) | ||
|
|
||
| writer.raw(TAG_IDENTIFIER) | ||
| writer.u8(self.size) | ||
| writer.u8(self.identifier) | ||
| writer.raw(self.data) | ||
|
|
||
| return tag_bytes | ||
|
|
||
| class Message(NamedTuple): | ||
| """ | ||
| The collection of a single or multi-lined message that will be displayed in an event, sign, message bubble, etc. | ||
| This also contains the list of actions/tags that will occur with the given message. | ||
| """ | ||
| string: str | ||
| tags: list[Tag] | ||
|
|
||
| class DAT1Section(BMGSection): | ||
| """ | ||
| Represents a section of DAT1 message data containing multiple messages with their associated tags. | ||
| This class handles the serialization and deserialization of message sections encoded in a binary format | ||
| that combines UTF-8/Shift-JIS encoded text with embedded tag markers. Messages are delimited by null | ||
| characters and can contain formatting or metadata tags at various offsets within the string. | ||
| Attributes: | ||
| messages (list[Message]): A list of Message objects contained in this section. | ||
| """ | ||
| def __init__(self, messages: list[Message] = None): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. | None |
||
| super().__init__(DAT1_MAGIC) | ||
|
|
||
| if messages == None: | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. messages is None |
||
| messages = [] | ||
| self.messages: list[Message] = messages | ||
|
|
||
| def add_message(self, message: Message): | ||
| self.messages.append(message) | ||
|
|
||
| @classmethod | ||
| def import_section(cls, raw_bytes: BinaryIO): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Need the return type |
||
| reader = BinaryReader(raw_bytes) | ||
| section = cls() | ||
|
|
||
| string = '' | ||
| tags: list[Tag] = [] | ||
|
|
||
| while reader.tell() < reader.size(): | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. reader.size() before. Every size costs 2 seek. |
||
| char_bytes = reader.raw(2) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Using the encoding here |
||
| if char_bytes == TAG_IDENTIFIER: # Found a tag | ||
| offset = len(string) | ||
| tag = Tag.import_tag(raw_bytes, offset) | ||
| tags.append(tag) | ||
| else: | ||
| int_value = int.from_bytes(char_bytes) | ||
| string += chr(int_value) | ||
|
|
||
| if char_bytes == NULL_BYTE: # Reading a null character | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. With that after the else statement before, the null char will be added |
||
| message = Message(string, tags) | ||
| section.add_message(message) | ||
|
|
||
| string = '' | ||
| tags = [] | ||
|
|
||
| return section | ||
|
|
||
| def export_section(self) -> BinaryIO: | ||
| """ | ||
| Export message section by serializing messages with their tags and characters into binary data. | ||
| Iterates through each message's characters and associated tags, writing tag data before each character | ||
| and any closing tags at the end of the string, encoding characters in Shift-JIS format. | ||
| """ | ||
| data = BytesIO() | ||
| writer = BinaryWriter(data) | ||
|
|
||
| for message in self.messages: | ||
| string = message.string | ||
| tags = message.tags | ||
|
|
||
| offset = -1 | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. You are using this and the offset + 1 in case the string is empty and the loop doesn't loop However, the string is never empty (because of the bug above) |
||
| for offset, char in enumerate(string): | ||
| current_tags = [tag for tag in tags if tag.offset == offset] | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. n x m comparisons here. |
||
| for tag in current_tags: | ||
| tag_data = tag.export_tag() | ||
| writer.raw(tag_data.read()) | ||
|
|
||
| writer.string(char, 2) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. will pad to the right if one char. Every non ascii character will raise an error. I don't think it's intended |
||
|
|
||
| if not string: | ||
| writer.raw(NULL_BYTE) | ||
|
|
||
| # Since message.string does not contains the tags themselves, we must also check to see if there are tags at the end of the string | ||
| closing_tags = [tag for tag in tags if tag.offset == offset + 1] | ||
| for tag in closing_tags: | ||
| tag_data = tag.export_tag() | ||
| writer.raw(tag_data.read) | ||
|
Owner
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. .getvalue() |
||
|
|
||
| return data | ||
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You never use the seek return value, remove it from the PR