Skip to content
Open

BMG #39

Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
*.iso
.idea
*.arc
**/__pycache__

/src_dir/
/assets
Expand Down
17 changes: 14 additions & 3 deletions src/wiithon/binary/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@Demorck Demorck Jul 31, 2026

Copy link
Copy Markdown
Owner

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


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:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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:
Expand Down Expand Up @@ -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)
127 changes: 127 additions & 0 deletions src/wiithon/formats/bmg.py
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:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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 read method with @classmethod decorator

reader = BinaryReader(raw_bytes)
data_magic = reader.string(0x4)
assert data_magic == DATA_MAGIC

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Assert -> exceptions


self.flw1_section_offset = reader.u32()

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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()

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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():

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably better to test some stuffs:

  • Testing the section_size > remaining size
  • If it's not the last section or the remaining size is superior to 32, raise a corrupted file

section_size = reader.size() - reader.tell()

section_bytes = reader.raw(section_size)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Better to be private write_section_header or smth like that
For me, "add_header" it's to add an header to BMG, not bmgsection

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

len(section_bytes.getvalue()) instead of seeking


padding = 0
if section_size % 32:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

section_bytes.read()

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]:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

get_sections with an s, it's a list

out: list[BMGSection] = []

for section in self.sections:
if section.magic == section_magic:
out.append(section)

return out

def export_bmg(self) -> BinaryIO:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

write method, not export_bmg

bmg_bytes = BytesIO()
writer = BinaryWriter(bmg_bytes)

writer.string(DATA_MAGIC)
writer.string(FILE_MAGIC)
writer.u32(0) # Write the flw1_section_offset later

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use writer.pad, it's explicit


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())

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The cursor of the stream is in EOF, use .getvalue()


return bmg_bytes
Empty file.
34 changes: 34 additions & 0 deletions src/wiithon/formats/bmg_sections/bmg_section.py
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

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use a ClassVar here


def __init__(self, magic: str):
self.magic = magic

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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":

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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")

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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")
153 changes: 153 additions & 0 deletions src/wiithon/formats/bmg_sections/dat1.py
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"

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's more TagGroup than identifier.
identifier is everything (like 1A 06 02 0000 for Name of the current player's Mii for Mario Kart Wii)

"""
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,

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

read/write

offset: int,
size: int,
identifier: TagIdentifier,
data: bytes = None):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bytes | None


if not isinstance(identifier, int) and not isinstance(identifier, TagIdentifier):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See my comment in inf1.py line 46

raise Exception("Bad Input")

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. No assert
  2. the constructor accept false, you can't have that

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):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

| None

super().__init__(DAT1_MAGIC)

if messages == None:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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):

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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():

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reader.size() before. Every size costs 2 seek.
2 each loop on 240Ko (in SMG) so it can be a lot

char_bytes = reader.raw(2)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)
If you fix the bug above, you will write the null byte without the tag

for offset, char in enumerate(string):
current_tags = [tag for tag in tags if tag.offset == offset]

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

n x m comparisons here.
You also can create a dic by offset before the loop, it's more readable

for tag in current_tags:
tag_data = tag.export_tag()
writer.raw(tag_data.read())

writer.string(char, 2)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The 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)

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

.getvalue()


return data
Loading
Loading