From e38023d4d81fc74ed64aeb4f4115f0e7e1fe67af Mon Sep 17 00:00:00 2001 From: kapkekes Date: Fri, 2 May 2025 01:05:16 +0700 Subject: [PATCH 01/12] [cocas] Add sized integers shenanigans --- cocas/library/__init__.py | 3 ++ cocas/library/stdint.py | 70 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+) create mode 100644 cocas/library/__init__.py create mode 100644 cocas/library/stdint.py diff --git a/cocas/library/__init__.py b/cocas/library/__init__.py new file mode 100644 index 00000000..24af3e9b --- /dev/null +++ b/cocas/library/__init__.py @@ -0,0 +1,3 @@ +__all__ = ("stdint",) + +from cocas.library import stdint diff --git a/cocas/library/stdint.py b/cocas/library/stdint.py new file mode 100644 index 00000000..5c763935 --- /dev/null +++ b/cocas/library/stdint.py @@ -0,0 +1,70 @@ +__all__ = ("sized", "validate") + +import dataclasses +import functools +from collections.abc import Mapping +from typing import TypeGuard, TypeVar, TypedDict, cast + +DataclassT = TypeVar("DataclassT") + + +class StdintMetadata(TypedDict): + range: range + + +def is_stdint_metadata(m: object) -> TypeGuard[StdintMetadata]: + if not isinstance(m, Mapping): + return False + + r = m.get("range") # pyright: ignore[reportUnknownMemberType] + return r is not None and isinstance(r, range) + + +def sized(n: int, unsigned: bool = True) -> StdintMetadata: + """Set a bit size constraint for an `int` field. + + Remember to use a `stdint.validate` on your dataclass! + """ + if n <= 0: + message = f"bit size of int cannot be less than 1, while {n} was provided" + raise ValueError(message) + + if unsigned: + return StdintMetadata(range=range(0, 1 << n)) + + if n == 1: + message = f"1-bit signed int does not have any meaning" + raise ValueError(message) + + bound = 1 << (n - 1) + return StdintMetadata(range=range(-bound, bound)) + + +def validate(cls: type[DataclassT]) -> type[DataclassT]: + """Modify the provided `dataclasses.dataclass` to validate marked `int` fields.""" + + if not dataclasses.is_dataclass(cls): + message = f"provided class should be a dataclass, while {cls} is not" + raise TypeError(message) + + collected: dict[str, StdintMetadata] = {} + for field in dataclasses.fields(cls): + if is_stdint_metadata(field.metadata): + collected[field.name] = field.metadata + + if not collected: + return cls + + init = cls.__init__ + @functools.wraps(init) + def init_with_validate(self, *args, **kwargs) -> None: # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] + init(self, *args, **kwargs) # pyright: ignore[reportUnknownArgumentType] + + for name, metadata in collected.items(): + value = cast(int, getattr(self, name)) # pyright: ignore[reportUnknownArgumentType] + if value not in metadata["range"]: + message = f"field {name} should be in {metadata['range']}, while {value} is not" + raise ValueError(message) + + cls.__init__ = init_with_validate + return cls From afa3e0e79e276a912fcad03df87b9259762193e1 Mon Sep 17 00:00:00 2001 From: kapkekes Date: Fri, 2 May 2025 01:08:26 +0700 Subject: [PATCH 02/12] [cocas] Add basic PLE kit --- cocas/ple/__init__.py | 13 ++++++++ cocas/ple/constants.py | 17 ++++++++++ cocas/ple/dump.py | 72 ++++++++++++++++++++++++++++++++++++++++++ cocas/ple/load.py | 17 ++++++++++ cocas/ple/types.py | 63 ++++++++++++++++++++++++++++++++++++ 5 files changed, 182 insertions(+) create mode 100644 cocas/ple/__init__.py create mode 100644 cocas/ple/constants.py create mode 100644 cocas/ple/dump.py create mode 100644 cocas/ple/load.py create mode 100644 cocas/ple/types.py diff --git a/cocas/ple/__init__.py b/cocas/ple/__init__.py new file mode 100644 index 00000000..b4a05aca --- /dev/null +++ b/cocas/ple/__init__.py @@ -0,0 +1,13 @@ +__all__ = ( + "constants", + "PleSegmentType", + "PleSegmentEntry", + "PleSegmentFlag", + "PlainExecutable", + "dump", + "dumps", +) + +from cocas.ple import constants +from cocas.ple.types import PleSegmentType, PleSegmentEntry, PleSegmentFlag, PlainExecutable +from cocas.ple.dump import dump, dumps diff --git a/cocas/ple/constants.py b/cocas/ple/constants.py new file mode 100644 index 00000000..de3096af --- /dev/null +++ b/cocas/ple/constants.py @@ -0,0 +1,17 @@ +__all__ = ( + "MAGIC_BYTES", + "SECTOR_SIZE", + "PARAGRAPH_SIZE", + "SEGMENT_HEADER_SIZE", +) + +from typing import Final + +MAGIC_BYTES: Final[bytes] = b"\x7fPLE" +"""todo: add docstring""" +SECTOR_SIZE: Final[int] = 512 +"""todo: add docstring""" +PARAGRAPH_SIZE: Final[int] = 16 +"""todo: add docstring""" +SEGMENT_HEADER_SIZE: Final[int] = 1 + 1 + 2 + 2 + 2 + 2 +"""todo: add docstring""" diff --git a/cocas/ple/dump.py b/cocas/ple/dump.py new file mode 100644 index 00000000..7e0530af --- /dev/null +++ b/cocas/ple/dump.py @@ -0,0 +1,72 @@ +__all__ = ("dump", "dumps") + +from collections.abc import Callable +from functools import partial +from io import BytesIO +from typing import TYPE_CHECKING + +from cocas.ple.constants import MAGIC_BYTES, PARAGRAPH_SIZE, SECTOR_SIZE, SEGMENT_HEADER_SIZE +from cocas.ple.types import PlainExecutable + +if TYPE_CHECKING: + from _typeshed import SupportsWrite + +to_bytes = partial(int.to_bytes, byteorder="big", signed=False) +u8 = partial(to_bytes, length=1) +u16 = partial(to_bytes, length=2) + + +def fill(n: int, size: int) -> int: + """todo: add docstring""" + return 0 if n == size else size - (n % size) + + +def takes(n: int, size: int) -> int: + fully_occupied, rem_bytes = divmod(n, size) + return fully_occupied + bool(rem_bytes) + + +def writer(fp: "SupportsWrite[bytes]") -> Callable[[*tuple[bytes, ...]], int]: + """Generate an auto-aligning write function for provided `fp`.""" + def write_bytes(*args: bytes, size: int = PARAGRAPH_SIZE) -> int: + n = 0 + for arg in args: + n += len(arg) + _ = fp.write(arg) + f = fill(n, size) + if f: + _ = fp.write(b"\x00" * f) + return n + f + return write_bytes + + +def dump(o: PlainExecutable, fp: "SupportsWrite[bytes]") -> int: + """todo: add docstring""" + write = writer(fp) + + n = write(MAGIC_BYTES, u8(o.version), u8(len(o.segments)), u16(o.entrypoint)) + headers_offset = n + (SEGMENT_HEADER_SIZE + fill(SEGMENT_HEADER_SIZE, PARAGRAPH_SIZE)) * len(o.segments) + current_sector = takes(headers_offset, SECTOR_SIZE) + + for seg in o.segments: + paragraphs = takes(len(seg.content), SEGMENT_HEADER_SIZE) + n += write( + u8(seg.type), + u8(seg.flags), + u16(current_sector), + u16(paragraphs), + u16(paragraphs), + u16(seg.virtual_address), + ) + + for seg in o.segments: + n += write(seg.content) + + return n + + +def dumps(o: PlainExecutable) -> bytes: + """todo: add docstring""" + buffer = BytesIO() + dump(o, buffer) + return buffer.getvalue() diff --git a/cocas/ple/load.py b/cocas/ple/load.py new file mode 100644 index 00000000..1f9ff1e1 --- /dev/null +++ b/cocas/ple/load.py @@ -0,0 +1,17 @@ +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from _typeshed import SupportsRead + + +def read(f: SupportsRead[bytes], length: int) -> tuple[int, bytes]: + try: + b = f.read(length) + except EOFError: + return (0, b"") + + return len(b), b + + +# WIP diff --git a/cocas/ple/types.py b/cocas/ple/types.py new file mode 100644 index 00000000..bef61d8b --- /dev/null +++ b/cocas/ple/types.py @@ -0,0 +1,63 @@ +__all__ = ( + "PlainExecutable", + "PleSegmentEntry", + "PleSegmentType", + "PleSegmentFlag", +) + +from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import IntEnum, IntFlag, auto +from typing import SupportsIndex, TypeVar + +from cocas.library import stdint + +PleT = TypeVar("PleT", bound="PlainExecutable") + + +class PleSegmentType(IntEnum): + """todo: add docstring""" + NONE = auto() + LOAD = auto() + + +class PleSegmentFlag(IntFlag): + """todo: add docstring""" + DATA = auto() + WRITE_PROTECTED = auto() + + +@stdint.validate +@dataclass +class PleSegmentEntry: + """todo: add docstring""" + content: bytes + virtual_address: int = field(metadata=stdint.sized(16)) + type: PleSegmentType = field(default=PleSegmentType.NONE) + flags: PleSegmentFlag = field(default=PleSegmentFlag(0)) + + +@stdint.validate +@dataclass +class PlainExecutable: + """todo: add docstring""" + entrypoint: int = field(metadata=stdint.sized(16)) + version: int = field(default=1, metadata=stdint.sized(8)) + + _segments: list[PleSegmentEntry] = field(default_factory=list, init=False) + + @property + def segments(self) -> Sequence[PleSegmentEntry]: + return self._segments + + def add(self: PleT, segment: PleSegmentEntry) -> PleT: + self._segments.append(segment) + return self + + def pop(self, index: SupportsIndex = -1) -> PleSegmentEntry | None: + try: + segment = self._segments.pop(index) + except IndexError: + segment = None + + return segment From 5dd0fd3454c49af50f588adaa0d2b3550ec4effe Mon Sep 17 00:00:00 2001 From: kapkekes Date: Fri, 2 May 2025 01:09:52 +0700 Subject: [PATCH 03/12] [cocas] Run linters --- cocas/library/stdint.py | 8 +++++--- cocas/ple/__init__.py | 2 +- cocas/ple/load.py | 1 - 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/cocas/library/stdint.py b/cocas/library/stdint.py index 5c763935..a2479c5d 100644 --- a/cocas/library/stdint.py +++ b/cocas/library/stdint.py @@ -3,7 +3,7 @@ import dataclasses import functools from collections.abc import Mapping -from typing import TypeGuard, TypeVar, TypedDict, cast +from typing import TypedDict, TypeGuard, TypeVar, cast DataclassT = TypeVar("DataclassT") @@ -33,7 +33,7 @@ def sized(n: int, unsigned: bool = True) -> StdintMetadata: return StdintMetadata(range=range(0, 1 << n)) if n == 1: - message = f"1-bit signed int does not have any meaning" + message = "1-bit signed int does not have any meaning" raise ValueError(message) bound = 1 << (n - 1) @@ -57,7 +57,9 @@ def validate(cls: type[DataclassT]) -> type[DataclassT]: init = cls.__init__ @functools.wraps(init) - def init_with_validate(self, *args, **kwargs) -> None: # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] + def init_with_validate( + self, *args, **kwargs, # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] + ) -> None: init(self, *args, **kwargs) # pyright: ignore[reportUnknownArgumentType] for name, metadata in collected.items(): diff --git a/cocas/ple/__init__.py b/cocas/ple/__init__.py index b4a05aca..946288c1 100644 --- a/cocas/ple/__init__.py +++ b/cocas/ple/__init__.py @@ -9,5 +9,5 @@ ) from cocas.ple import constants -from cocas.ple.types import PleSegmentType, PleSegmentEntry, PleSegmentFlag, PlainExecutable from cocas.ple.dump import dump, dumps +from cocas.ple.types import PlainExecutable, PleSegmentEntry, PleSegmentFlag, PleSegmentType diff --git a/cocas/ple/load.py b/cocas/ple/load.py index 1f9ff1e1..9726cc0c 100644 --- a/cocas/ple/load.py +++ b/cocas/ple/load.py @@ -1,6 +1,5 @@ from typing import TYPE_CHECKING - if TYPE_CHECKING: from _typeshed import SupportsRead From 0623f0e164257a86f3369055cf34470f5e795daf Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 00:52:29 +0700 Subject: [PATCH 04/12] [cocas] Add root library exception --- cocas/exceptions.py | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 cocas/exceptions.py diff --git a/cocas/exceptions.py b/cocas/exceptions.py new file mode 100644 index 00000000..b73e8e96 --- /dev/null +++ b/cocas/exceptions.py @@ -0,0 +1,5 @@ +__all__ = ("CocasException",) + + +class CocasException(Exception): + """Base exception class for `cocas` command utility.""" From 072411588f6a919a25a10f82428201d498ced1bd Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 00:52:57 +0700 Subject: [PATCH 05/12] [cocas] Add PLE utilities --- cocas/ple/__init__.py | 14 +++-- cocas/ple/_io.py | 93 ++++++++++++++++++++++++++++++ cocas/ple/constants.py | 3 - cocas/ple/dump.py | 72 ----------------------- cocas/ple/exceptions.py | 41 ++++++++++++++ cocas/ple/load.py | 16 ------ cocas/ple/types.py | 123 ++++++++++++++++++++++++++++++++++++++-- 7 files changed, 260 insertions(+), 102 deletions(-) create mode 100644 cocas/ple/_io.py delete mode 100644 cocas/ple/dump.py create mode 100644 cocas/ple/exceptions.py delete mode 100644 cocas/ple/load.py diff --git a/cocas/ple/__init__.py b/cocas/ple/__init__.py index 946288c1..5a1a0bc8 100644 --- a/cocas/ple/__init__.py +++ b/cocas/ple/__init__.py @@ -1,13 +1,15 @@ __all__ = ( + # public modules "constants", + "exceptions", + "types", + + # quality of life re-exports "PleSegmentType", "PleSegmentEntry", "PleSegmentFlag", - "PlainExecutable", - "dump", - "dumps", + "PleImage", ) -from cocas.ple import constants -from cocas.ple.dump import dump, dumps -from cocas.ple.types import PlainExecutable, PleSegmentEntry, PleSegmentFlag, PleSegmentType +from cocas.ple import constants, exceptions, types +from cocas.ple.types import PleImage, PleSegmentEntry, PleSegmentFlag, PleSegmentType diff --git a/cocas/ple/_io.py b/cocas/ple/_io.py new file mode 100644 index 00000000..3c5a7ed9 --- /dev/null +++ b/cocas/ple/_io.py @@ -0,0 +1,93 @@ +__all__ = ( + "round_up", + "Reader", + "Writer", +) + +from typing import TYPE_CHECKING, Generic, TypeVar + +from cocas.ple.exceptions import ExhaustedStreamError, FailedWriteError + +if TYPE_CHECKING: + from _typeshed import SupportsRead, SupportsWrite + +T = TypeVar("T") +WriterT = TypeVar("WriterT", bound="Writer") + + +def round_up(value: int, multiple: int) -> int: + c, r = divmod(value, multiple) + return c + bool(r) + + +def reverse_rem(value: int, divider: int) -> int: + r = value % divider + return 0 if r == 0 else (divider - r) + + +class Handle(Generic[T]): + def __init__(self, fp: T) -> None: + self._fp: T = fp + + +class Reader(Handle["SupportsRead[bytes]"]): + def __init__(self, fp: "SupportsRead[bytes]") -> None: + super().__init__(fp) + self._read: int = 0 + + def bytes(self, length: int) -> bytes: + try: + b = self._fp.read(length) + except Exception as exc: + message = "failed to read byte chunk from stream" + raise ExhaustedStreamError(message, None, length) from exc + + self._read += len(b) + if len(b) != length: + message = f"chunk from stream is too short (expected to read {length} bytes, while got only {len(b)} bytes)" + raise ExhaustedStreamError(message, b, length) + + return b + + def u8(self) -> int: + return int.from_bytes(self.bytes(1), byteorder="big", signed=False) + + def u16(self) -> int: + return int.from_bytes(self.bytes(2), byteorder="big", signed=False) + + def skip(self, alignment: int) -> None: + r = reverse_rem(self._read, alignment) + if r != 0: + _ = self.bytes(r) + + +class Writer(Handle["SupportsWrite[bytes]"]): + def __init__(self, fp: "SupportsWrite[bytes]") -> None: + super().__init__(fp) + self._emitted: int = 0 + + @property + def emitted(self) -> int: + return self._emitted + + def bytes(self: WriterT, value: bytes) -> WriterT: + try: + _ = self._fp.write(value) + except Exception as exc: + message = "failed to write payload to stream" + raise FailedWriteError(message, value) from exc + + self._emitted += len(value) + return self + + def u8(self: WriterT, value: int) -> WriterT: + return self.bytes(value.to_bytes(length=1, byteorder="big", signed=False)) + + def u16(self: WriterT, value: int) -> WriterT: + return self.bytes(value.to_bytes(length=2, byteorder="big", signed=False)) + + def align(self: WriterT, alignment: int) -> WriterT: + r = reverse_rem(self._emitted, alignment) + if r != 0: + return self.bytes(bytes(r)) + return self diff --git a/cocas/ple/constants.py b/cocas/ple/constants.py index de3096af..c843da63 100644 --- a/cocas/ple/constants.py +++ b/cocas/ple/constants.py @@ -2,7 +2,6 @@ "MAGIC_BYTES", "SECTOR_SIZE", "PARAGRAPH_SIZE", - "SEGMENT_HEADER_SIZE", ) from typing import Final @@ -13,5 +12,3 @@ """todo: add docstring""" PARAGRAPH_SIZE: Final[int] = 16 """todo: add docstring""" -SEGMENT_HEADER_SIZE: Final[int] = 1 + 1 + 2 + 2 + 2 + 2 -"""todo: add docstring""" diff --git a/cocas/ple/dump.py b/cocas/ple/dump.py deleted file mode 100644 index 7e0530af..00000000 --- a/cocas/ple/dump.py +++ /dev/null @@ -1,72 +0,0 @@ -__all__ = ("dump", "dumps") - -from collections.abc import Callable -from functools import partial -from io import BytesIO -from typing import TYPE_CHECKING - -from cocas.ple.constants import MAGIC_BYTES, PARAGRAPH_SIZE, SECTOR_SIZE, SEGMENT_HEADER_SIZE -from cocas.ple.types import PlainExecutable - -if TYPE_CHECKING: - from _typeshed import SupportsWrite - -to_bytes = partial(int.to_bytes, byteorder="big", signed=False) -u8 = partial(to_bytes, length=1) -u16 = partial(to_bytes, length=2) - - -def fill(n: int, size: int) -> int: - """todo: add docstring""" - return 0 if n == size else size - (n % size) - - -def takes(n: int, size: int) -> int: - fully_occupied, rem_bytes = divmod(n, size) - return fully_occupied + bool(rem_bytes) - - -def writer(fp: "SupportsWrite[bytes]") -> Callable[[*tuple[bytes, ...]], int]: - """Generate an auto-aligning write function for provided `fp`.""" - def write_bytes(*args: bytes, size: int = PARAGRAPH_SIZE) -> int: - n = 0 - for arg in args: - n += len(arg) - _ = fp.write(arg) - f = fill(n, size) - if f: - _ = fp.write(b"\x00" * f) - return n + f - return write_bytes - - -def dump(o: PlainExecutable, fp: "SupportsWrite[bytes]") -> int: - """todo: add docstring""" - write = writer(fp) - - n = write(MAGIC_BYTES, u8(o.version), u8(len(o.segments)), u16(o.entrypoint)) - headers_offset = n + (SEGMENT_HEADER_SIZE + fill(SEGMENT_HEADER_SIZE, PARAGRAPH_SIZE)) * len(o.segments) - current_sector = takes(headers_offset, SECTOR_SIZE) - - for seg in o.segments: - paragraphs = takes(len(seg.content), SEGMENT_HEADER_SIZE) - n += write( - u8(seg.type), - u8(seg.flags), - u16(current_sector), - u16(paragraphs), - u16(paragraphs), - u16(seg.virtual_address), - ) - - for seg in o.segments: - n += write(seg.content) - - return n - - -def dumps(o: PlainExecutable) -> bytes: - """todo: add docstring""" - buffer = BytesIO() - dump(o, buffer) - return buffer.getvalue() diff --git a/cocas/ple/exceptions.py b/cocas/ple/exceptions.py new file mode 100644 index 00000000..1483b145 --- /dev/null +++ b/cocas/ple/exceptions.py @@ -0,0 +1,41 @@ +__all__ = ( + "PleException", + "PleIoError", + "ExhaustedStreamError", + "FailedWriteError", + "CorruptedImageError", +) + +from cocas.exceptions import CocasException + + +class PleException(CocasException): + """Base exception class for PLE utilities.""" + + +class PleIoError(PleException): + """I/O PLE errors.""" + + +class ExhaustedStreamError(PleIoError): + """Failed to read a requested number of bytes from stream.""" + def __init__(self, message: str, chunk: bytes | None, length: int) -> None: + super().__init__(message) + self.chunk: bytes | None = chunk + """Successfully read bytes, if any.""" + self.length: int = length + """Length of requested chunk.""" + + +class FailedWriteError(PleIoError): + """Failed to write to provided stream.""" + def __init__(self, message: str, payload: bytes) -> None: + super().__init__(message) + self.payload: bytes = payload + """Payload, which was passed to `SupportsWrite[bytes].write`.""" + + +class CorruptedImageError(PleException): + """Provided image source is corrupted.""" + def __init__(self, message: str) -> None: + super().__init__(message) diff --git a/cocas/ple/load.py b/cocas/ple/load.py deleted file mode 100644 index 9726cc0c..00000000 --- a/cocas/ple/load.py +++ /dev/null @@ -1,16 +0,0 @@ -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from _typeshed import SupportsRead - - -def read(f: SupportsRead[bytes], length: int) -> tuple[int, bytes]: - try: - b = f.read(length) - except EOFError: - return (0, b"") - - return len(b), b - - -# WIP diff --git a/cocas/ple/types.py b/cocas/ple/types.py index bef61d8b..50906c86 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -1,28 +1,39 @@ __all__ = ( - "PlainExecutable", + "PleImage", "PleSegmentEntry", "PleSegmentType", "PleSegmentFlag", ) +import bisect from collections.abc import Sequence from dataclasses import dataclass, field from enum import IntEnum, IntFlag, auto -from typing import SupportsIndex, TypeVar +from io import BytesIO +from typing import TYPE_CHECKING, SupportsIndex, TypeVar, cast +import cocas.ple._io as _io from cocas.library import stdint +from cocas.ple.constants import MAGIC_BYTES, PARAGRAPH_SIZE, SECTOR_SIZE +from cocas.ple.exceptions import CorruptedImageError -PleT = TypeVar("PleT", bound="PlainExecutable") +if TYPE_CHECKING: + from _typeshed import SupportsRead, SupportsWrite + +PleImageT = TypeVar("PleImageT", bound="PleImage") +SENTINEL = object() class PleSegmentType(IntEnum): """todo: add docstring""" + NONE = auto() LOAD = auto() class PleSegmentFlag(IntFlag): """todo: add docstring""" + DATA = auto() WRITE_PROTECTED = auto() @@ -31,16 +42,27 @@ class PleSegmentFlag(IntFlag): @dataclass class PleSegmentEntry: """todo: add docstring""" + content: bytes virtual_address: int = field(metadata=stdint.sized(16)) + memory_size: int = field(default=cast(int, SENTINEL), metadata=stdint.sized(16)) type: PleSegmentType = field(default=PleSegmentType.NONE) flags: PleSegmentFlag = field(default=PleSegmentFlag(0)) + @property + def physical_size(self) -> int: + return _io.round_up(len(self.content), PARAGRAPH_SIZE) + + def __post_init__(self) -> None: + if self.memory_size is SENTINEL: + self.memory_size = self.physical_size + @stdint.validate @dataclass -class PlainExecutable: +class PleImage: """todo: add docstring""" + entrypoint: int = field(metadata=stdint.sized(16)) version: int = field(default=1, metadata=stdint.sized(8)) @@ -50,7 +72,7 @@ class PlainExecutable: def segments(self) -> Sequence[PleSegmentEntry]: return self._segments - def add(self: PleT, segment: PleSegmentEntry) -> PleT: + def add(self: PleImageT, segment: PleSegmentEntry) -> PleImageT: self._segments.append(segment) return self @@ -61,3 +83,94 @@ def pop(self, index: SupportsIndex = -1) -> PleSegmentEntry | None: segment = None return segment + + def dump(self, fp: "SupportsWrite[bytes]") -> int: + writer = _io.Writer(fp) + + segments_n = len(self.segments) + _ = writer.bytes(MAGIC_BYTES).u8(self.version).u8(segments_n).u16(self.entrypoint).align(PARAGRAPH_SIZE) + + headers_bytes = writer.emitted + PARAGRAPH_SIZE * segments_n + sector_index = _io.round_up(headers_bytes, SECTOR_SIZE) + + for s in self.segments: + _ = ( + writer.u8(s.type) + .u8(s.flags) + .u16(sector_index) + .u16(s.physical_size) + .u16(s.memory_size) + .u16(s.virtual_address) + .align(PARAGRAPH_SIZE) + ) + sector_index += _io.round_up(len(s.content), SECTOR_SIZE) + + for s in self.segments: + _ = writer.bytes(s.content).align(SECTOR_SIZE) + + return writer.emitted + + def dumps(self) -> bytes: + buffer = BytesIO() + _ = self.dump(buffer) + return buffer.getvalue() + + @classmethod + def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: + fp_reader = _io.Reader(fp) + + buffer = bytearray(fp_reader.bytes(SECTOR_SIZE)) + magic = buffer[0: len(MAGIC_BYTES)] + if magic != MAGIC_BYTES: # pyright: ignore[reportUnnecessaryComparison] + message = f"first {len(MAGIC_BYTES)} of PLE expected to be '{MAGIC_BYTES}', while '{magic}' has been read" + raise CorruptedImageError(message) + + index = len(MAGIC_BYTES) + version = buffer[index] + if version != 1: + message = f"only version 1 is supported, while provided image is version {version}" + raise NotImplementedError(message) + index += 1 + + segments_n = buffer[index] + headers_sectors = _io.round_up((1 + segments_n) * PARAGRAPH_SIZE, SECTOR_SIZE) + if headers_sectors > 1: + buffer += fp_reader.bytes((headers_sectors - 1) * SECTOR_SIZE) + + buffer_reader = _io.Reader(BytesIO(buffer)) + buffer_reader.skip(index) + entrypoint = buffer_reader.u16() + buffer_reader.skip(PARAGRAPH_SIZE) + + entries_metadata: list[tuple[int, int, PleSegmentEntry]] = [] + for _ in range(segments_n): + type = PleSegmentType(buffer_reader.u8()) + flags = PleSegmentFlag(buffer_reader.u8()) + physical_offset = buffer_reader.u16() + physical_size = buffer_reader.u16() + memory_size = buffer_reader.u16() + virtual_address = buffer_reader.u16() + entry = PleSegmentEntry(bytes(), virtual_address, memory_size, type, flags) + bisect.insort(entries_metadata, (physical_offset, physical_size, entry), key=lambda x: x[0]) + + current_offset = 0 + for physical_offset, physical_size, entry in entries_metadata: + offset_delta = physical_offset - current_offset + if offset_delta: + buffer = buffer[offset_delta:] + current_offset = physical_offset + + bytes_size = physical_size * PARAGRAPH_SIZE + size_delta = bytes_size - len(buffer) + if size_delta: + buffer += fp_reader.bytes(_io.round_up(size_delta, SECTOR_SIZE)) + + entry.content = bytes(buffer[0: bytes_size]) + + image = cls(entrypoint) + image._segments = [entry for _, _, entry in entries_metadata] + return image + + @classmethod + def loads(cls: type[PleImageT], s: bytes | bytearray) -> PleImageT: + return cls.load(BytesIO(s)) From bc9cab320b0ac503da5fb118703a958ffd27e7af Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 02:16:42 +0700 Subject: [PATCH 06/12] [cocas] Add missing alignment --- cocas/ple/types.py | 1 + 1 file changed, 1 insertion(+) diff --git a/cocas/ple/types.py b/cocas/ple/types.py index 50906c86..7c801ce0 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -150,6 +150,7 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: physical_size = buffer_reader.u16() memory_size = buffer_reader.u16() virtual_address = buffer_reader.u16() + buffer_reader.skip(PARAGRAPH_SIZE) entry = PleSegmentEntry(bytes(), virtual_address, memory_size, type, flags) bisect.insort(entries_metadata, (physical_offset, physical_size, entry), key=lambda x: x[0]) From 45426cf3ef12c7745bba10a4c7ae6e07efc1e900 Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 02:18:43 +0700 Subject: [PATCH 07/12] [cocas] Change obscure naming --- cocas/ple/_io.py | 5 +++-- cocas/ple/types.py | 10 +++++----- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/cocas/ple/_io.py b/cocas/ple/_io.py index 3c5a7ed9..8085c8eb 100644 --- a/cocas/ple/_io.py +++ b/cocas/ple/_io.py @@ -1,5 +1,5 @@ __all__ = ( - "round_up", + "round_up_div", "Reader", "Writer", ) @@ -15,7 +15,8 @@ WriterT = TypeVar("WriterT", bound="Writer") -def round_up(value: int, multiple: int) -> int: +def round_up_div(value: int, multiple: int) -> int: + """Like `//`, but rounds to ceil, not to floor.""" c, r = divmod(value, multiple) return c + bool(r) diff --git a/cocas/ple/types.py b/cocas/ple/types.py index 7c801ce0..4af3dcfe 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -51,7 +51,7 @@ class PleSegmentEntry: @property def physical_size(self) -> int: - return _io.round_up(len(self.content), PARAGRAPH_SIZE) + return _io.round_up_div(len(self.content), PARAGRAPH_SIZE) def __post_init__(self) -> None: if self.memory_size is SENTINEL: @@ -91,7 +91,7 @@ def dump(self, fp: "SupportsWrite[bytes]") -> int: _ = writer.bytes(MAGIC_BYTES).u8(self.version).u8(segments_n).u16(self.entrypoint).align(PARAGRAPH_SIZE) headers_bytes = writer.emitted + PARAGRAPH_SIZE * segments_n - sector_index = _io.round_up(headers_bytes, SECTOR_SIZE) + sector_index = _io.round_up_div(headers_bytes, SECTOR_SIZE) for s in self.segments: _ = ( @@ -103,7 +103,7 @@ def dump(self, fp: "SupportsWrite[bytes]") -> int: .u16(s.virtual_address) .align(PARAGRAPH_SIZE) ) - sector_index += _io.round_up(len(s.content), SECTOR_SIZE) + sector_index += _io.round_up_div(len(s.content), SECTOR_SIZE) for s in self.segments: _ = writer.bytes(s.content).align(SECTOR_SIZE) @@ -133,7 +133,7 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: index += 1 segments_n = buffer[index] - headers_sectors = _io.round_up((1 + segments_n) * PARAGRAPH_SIZE, SECTOR_SIZE) + headers_sectors = _io.round_up_div((1 + segments_n) * PARAGRAPH_SIZE, SECTOR_SIZE) if headers_sectors > 1: buffer += fp_reader.bytes((headers_sectors - 1) * SECTOR_SIZE) @@ -164,7 +164,7 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: bytes_size = physical_size * PARAGRAPH_SIZE size_delta = bytes_size - len(buffer) if size_delta: - buffer += fp_reader.bytes(_io.round_up(size_delta, SECTOR_SIZE)) + buffer += fp_reader.bytes(_io.round_up_div(size_delta, SECTOR_SIZE)) entry.content = bytes(buffer[0: bytes_size]) From 69a3eb3dd3d884904ec63ea432a8833d6867cf7f Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 02:19:20 +0700 Subject: [PATCH 08/12] [cocas] Add missing multiplication --- cocas/ple/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cocas/ple/types.py b/cocas/ple/types.py index 4af3dcfe..3c03c4fb 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -164,7 +164,7 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: bytes_size = physical_size * PARAGRAPH_SIZE size_delta = bytes_size - len(buffer) if size_delta: - buffer += fp_reader.bytes(_io.round_up_div(size_delta, SECTOR_SIZE)) + buffer += fp_reader.bytes(_io.round_up_div(size_delta, SECTOR_SIZE) * SECTOR_SIZE) entry.content = bytes(buffer[0: bytes_size]) From ca7a9236aa4984c739cfda7728c3c536c81659eb Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 02:23:00 +0700 Subject: [PATCH 09/12] [cocas] Remove `cocas.ple._io.Handle` --- cocas/ple/_io.py | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/cocas/ple/_io.py b/cocas/ple/_io.py index 8085c8eb..639a8982 100644 --- a/cocas/ple/_io.py +++ b/cocas/ple/_io.py @@ -26,14 +26,9 @@ def reverse_rem(value: int, divider: int) -> int: return 0 if r == 0 else (divider - r) -class Handle(Generic[T]): - def __init__(self, fp: T) -> None: - self._fp: T = fp - - -class Reader(Handle["SupportsRead[bytes]"]): +class Reader: def __init__(self, fp: "SupportsRead[bytes]") -> None: - super().__init__(fp) + self._fp: "SupportsRead[bytes]" = fp self._read: int = 0 def bytes(self, length: int) -> bytes: @@ -62,9 +57,9 @@ def skip(self, alignment: int) -> None: _ = self.bytes(r) -class Writer(Handle["SupportsWrite[bytes]"]): +class Writer: def __init__(self, fp: "SupportsWrite[bytes]") -> None: - super().__init__(fp) + self._fp: "SupportsWrite[bytes]" = fp self._emitted: int = 0 @property From 812b663f0045247ccef272d2c5b93911a2b7f341 Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 02:36:38 +0700 Subject: [PATCH 10/12] [cocas] Add docstrings --- cocas/ple/__init__.py | 2 ++ cocas/ple/_io.py | 52 ++++++++++++++++++++++++++++++++++++++++- cocas/ple/constants.py | 5 ++-- cocas/ple/exceptions.py | 2 ++ cocas/ple/types.py | 8 ------- 5 files changed, 57 insertions(+), 12 deletions(-) diff --git a/cocas/ple/__init__.py b/cocas/ple/__init__.py index 5a1a0bc8..ce7c92a5 100644 --- a/cocas/ple/__init__.py +++ b/cocas/ple/__init__.py @@ -1,3 +1,5 @@ +"""Plain Executable file format utilities.""" + __all__ = ( # public modules "constants", diff --git a/cocas/ple/_io.py b/cocas/ple/_io.py index 639a8982..49affb7e 100644 --- a/cocas/ple/_io.py +++ b/cocas/ple/_io.py @@ -4,7 +4,7 @@ "Writer", ) -from typing import TYPE_CHECKING, Generic, TypeVar +from typing import TYPE_CHECKING, TypeVar from cocas.ple.exceptions import ExhaustedStreamError, FailedWriteError @@ -22,6 +22,7 @@ def round_up_div(value: int, multiple: int) -> int: def reverse_rem(value: int, divider: int) -> int: + """Get such minimal `n`, that `(value + n) % divider == 0`.""" r = value % divider return 0 if r == 0 else (divider - r) @@ -31,7 +32,18 @@ def __init__(self, fp: "SupportsRead[bytes]") -> None: self._fp: "SupportsRead[bytes]" = fp self._read: int = 0 + @property + def read(self) -> int: + """Number of read bytes.""" + return self._read + def bytes(self, length: int) -> bytes: + """Read exactly `length` bytes from the stream. + + Raises: + ExhaustedStreamError: + If there are no enough bytes. + """ try: b = self._fp.read(length) except Exception as exc: @@ -46,12 +58,30 @@ def bytes(self, length: int) -> bytes: return b def u8(self) -> int: + """Read an unsigned byte. + + Raises: + ExhaustedStreamError: + If there are no enough bytes. + """ return int.from_bytes(self.bytes(1), byteorder="big", signed=False) def u16(self) -> int: + """Read an unsigned short. + + Raises: + ExhaustedStreamError: + If there are no enough bytes. + """ return int.from_bytes(self.bytes(2), byteorder="big", signed=False) def skip(self, alignment: int) -> None: + """Read and discard `n` bytes; `(Reader.read + n) % alignment == 0`. + + Raises: + ExhaustedStreamError: + If there are no enough bytes. + """ r = reverse_rem(self._read, alignment) if r != 0: _ = self.bytes(r) @@ -64,9 +94,16 @@ def __init__(self, fp: "SupportsWrite[bytes]") -> None: @property def emitted(self) -> int: + """Number of written bytes.""" return self._emitted def bytes(self: WriterT, value: bytes) -> WriterT: + """Write `value` bytes to the stream. + + Raises: + FailedWriteError: + If write hasn't succeeded. + """ try: _ = self._fp.write(value) except Exception as exc: @@ -77,12 +114,25 @@ def bytes(self: WriterT, value: bytes) -> WriterT: return self def u8(self: WriterT, value: int) -> WriterT: + """Write unsigned byte `value` to the stream. + + Raises: + FailedWriteError: + If write hasn't succeeded. + """ return self.bytes(value.to_bytes(length=1, byteorder="big", signed=False)) def u16(self: WriterT, value: int) -> WriterT: + """Write unsigned short `value` to the stream. + + Raises: + FailedWriteError: + If write hasn't succeeded. + """ return self.bytes(value.to_bytes(length=2, byteorder="big", signed=False)) def align(self: WriterT, alignment: int) -> WriterT: + """Write `n` null bytes; `(Writer.emitted + n) % alignment == 0`.""" r = reverse_rem(self._emitted, alignment) if r != 0: return self.bytes(bytes(r)) diff --git a/cocas/ple/constants.py b/cocas/ple/constants.py index c843da63..79a90e94 100644 --- a/cocas/ple/constants.py +++ b/cocas/ple/constants.py @@ -1,3 +1,5 @@ +"""Plain Executalbe constants.""" + __all__ = ( "MAGIC_BYTES", "SECTOR_SIZE", @@ -7,8 +9,5 @@ from typing import Final MAGIC_BYTES: Final[bytes] = b"\x7fPLE" -"""todo: add docstring""" SECTOR_SIZE: Final[int] = 512 -"""todo: add docstring""" PARAGRAPH_SIZE: Final[int] = 16 -"""todo: add docstring""" diff --git a/cocas/ple/exceptions.py b/cocas/ple/exceptions.py index 1483b145..5e96b3f2 100644 --- a/cocas/ple/exceptions.py +++ b/cocas/ple/exceptions.py @@ -1,3 +1,5 @@ +"""Plain Executable exceptions.""" + __all__ = ( "PleException", "PleIoError", diff --git a/cocas/ple/types.py b/cocas/ple/types.py index 3c03c4fb..e46f0a8b 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -25,15 +25,11 @@ class PleSegmentType(IntEnum): - """todo: add docstring""" - NONE = auto() LOAD = auto() class PleSegmentFlag(IntFlag): - """todo: add docstring""" - DATA = auto() WRITE_PROTECTED = auto() @@ -41,8 +37,6 @@ class PleSegmentFlag(IntFlag): @stdint.validate @dataclass class PleSegmentEntry: - """todo: add docstring""" - content: bytes virtual_address: int = field(metadata=stdint.sized(16)) memory_size: int = field(default=cast(int, SENTINEL), metadata=stdint.sized(16)) @@ -61,8 +55,6 @@ def __post_init__(self) -> None: @stdint.validate @dataclass class PleImage: - """todo: add docstring""" - entrypoint: int = field(metadata=stdint.sized(16)) version: int = field(default=1, metadata=stdint.sized(8)) From 9e854560c94fa8d7ea9f0959786251835d41a83b Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 03:15:40 +0700 Subject: [PATCH 11/12] [cocas] Fix bugs --- cocas/ple/_io.py | 14 +++++++------- cocas/ple/types.py | 11 ++++++----- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/cocas/ple/_io.py b/cocas/ple/_io.py index 49affb7e..4e8f32bd 100644 --- a/cocas/ple/_io.py +++ b/cocas/ple/_io.py @@ -64,7 +64,7 @@ def u8(self) -> int: ExhaustedStreamError: If there are no enough bytes. """ - return int.from_bytes(self.bytes(1), byteorder="big", signed=False) + return int.from_bytes(self.bytes(1), byteorder="little", signed=False) def u16(self) -> int: """Read an unsigned short. @@ -73,9 +73,9 @@ def u16(self) -> int: ExhaustedStreamError: If there are no enough bytes. """ - return int.from_bytes(self.bytes(2), byteorder="big", signed=False) + return int.from_bytes(self.bytes(2), byteorder="little", signed=False) - def skip(self, alignment: int) -> None: + def skip(self, alignment: int, force: bool = False) -> None: """Read and discard `n` bytes; `(Reader.read + n) % alignment == 0`. Raises: @@ -83,8 +83,8 @@ def skip(self, alignment: int) -> None: If there are no enough bytes. """ r = reverse_rem(self._read, alignment) - if r != 0: - _ = self.bytes(r) + if r != 0 or force: + _ = self.bytes(r or alignment) class Writer: @@ -120,7 +120,7 @@ def u8(self: WriterT, value: int) -> WriterT: FailedWriteError: If write hasn't succeeded. """ - return self.bytes(value.to_bytes(length=1, byteorder="big", signed=False)) + return self.bytes(value.to_bytes(length=1, byteorder="little", signed=False)) def u16(self: WriterT, value: int) -> WriterT: """Write unsigned short `value` to the stream. @@ -129,7 +129,7 @@ def u16(self: WriterT, value: int) -> WriterT: FailedWriteError: If write hasn't succeeded. """ - return self.bytes(value.to_bytes(length=2, byteorder="big", signed=False)) + return self.bytes(value.to_bytes(length=2, byteorder="little", signed=False)) def align(self: WriterT, alignment: int) -> WriterT: """Write `n` null bytes; `(Writer.emitted + n) % alignment == 0`.""" diff --git a/cocas/ple/types.py b/cocas/ple/types.py index e46f0a8b..76b9f478 100644 --- a/cocas/ple/types.py +++ b/cocas/ple/types.py @@ -25,8 +25,8 @@ class PleSegmentType(IntEnum): - NONE = auto() - LOAD = auto() + NONE = 0x0 + LOAD = 0x1 class PleSegmentFlag(IntFlag): @@ -95,7 +95,7 @@ def dump(self, fp: "SupportsWrite[bytes]") -> int: .u16(s.virtual_address) .align(PARAGRAPH_SIZE) ) - sector_index += _io.round_up_div(len(s.content), SECTOR_SIZE) + sector_index += _io.round_up_div(s.physical_size * PARAGRAPH_SIZE, SECTOR_SIZE) for s in self.segments: _ = writer.bytes(s.content).align(SECTOR_SIZE) @@ -128,9 +128,10 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: headers_sectors = _io.round_up_div((1 + segments_n) * PARAGRAPH_SIZE, SECTOR_SIZE) if headers_sectors > 1: buffer += fp_reader.bytes((headers_sectors - 1) * SECTOR_SIZE) + index += 1 buffer_reader = _io.Reader(BytesIO(buffer)) - buffer_reader.skip(index) + buffer_reader.skip(index, force=True) entrypoint = buffer_reader.u16() buffer_reader.skip(PARAGRAPH_SIZE) @@ -150,7 +151,7 @@ def load(cls: type[PleImageT], fp: "SupportsRead[bytes]") -> PleImageT: for physical_offset, physical_size, entry in entries_metadata: offset_delta = physical_offset - current_offset if offset_delta: - buffer = buffer[offset_delta:] + buffer = buffer[offset_delta * SECTOR_SIZE:] current_offset = physical_offset bytes_size = physical_size * PARAGRAPH_SIZE From 48af0e1558a895d0dc59dbd36f23c99aecf90b8d Mon Sep 17 00:00:00 2001 From: kapkekes Date: Mon, 5 May 2025 03:15:55 +0700 Subject: [PATCH 12/12] [cocas] Add basic test for PLE --- tests/tests_ple/ple.bin | Bin 0 -> 2048 bytes tests/tests_ple/test_sdk.py | 11 +++++++++++ 2 files changed, 11 insertions(+) create mode 100644 tests/tests_ple/ple.bin create mode 100644 tests/tests_ple/test_sdk.py diff --git a/tests/tests_ple/ple.bin b/tests/tests_ple/ple.bin new file mode 100644 index 0000000000000000000000000000000000000000..bbc7bc0a1c105959e056886add429bb70ca2fb36 GIT binary patch literal 2048 zcmb;q@Ns2iYG8l?Mg~R(1t5e0Mg}GZ2ZjI!1_7AJC_NeiqaiRF0;3@?8UmvsFw{c; E04GlZ$p8QV literal 0 HcmV?d00001 diff --git a/tests/tests_ple/test_sdk.py b/tests/tests_ple/test_sdk.py new file mode 100644 index 00000000..ecec5c20 --- /dev/null +++ b/tests/tests_ple/test_sdk.py @@ -0,0 +1,11 @@ +from pathlib import Path + +from cocas.ple.types import PleImage + + +def test_idempotency() -> None: + with open(Path(__file__).parent / "ple.bin", "rb") as file: + loaded = PleImage.load(file) + + with open(Path(__file__).parent / "ple.bin", "rb") as file: + assert PleImage.dumps(loaded) == file.read()