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.""" 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..a2479c5d --- /dev/null +++ b/cocas/library/stdint.py @@ -0,0 +1,72 @@ +__all__ = ("sized", "validate") + +import dataclasses +import functools +from collections.abc import Mapping +from typing import TypedDict, TypeGuard, TypeVar, 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 = "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, # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] + ) -> None: + 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 diff --git a/cocas/ple/__init__.py b/cocas/ple/__init__.py new file mode 100644 index 00000000..ce7c92a5 --- /dev/null +++ b/cocas/ple/__init__.py @@ -0,0 +1,17 @@ +"""Plain Executable file format utilities.""" + +__all__ = ( + # public modules + "constants", + "exceptions", + "types", + + # quality of life re-exports + "PleSegmentType", + "PleSegmentEntry", + "PleSegmentFlag", + "PleImage", +) + +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..4e8f32bd --- /dev/null +++ b/cocas/ple/_io.py @@ -0,0 +1,139 @@ +__all__ = ( + "round_up_div", + "Reader", + "Writer", +) + +from typing import TYPE_CHECKING, 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_div(value: int, multiple: int) -> int: + """Like `//`, but rounds to ceil, not to floor.""" + c, r = divmod(value, multiple) + return c + bool(r) + + +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) + + +class Reader: + 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: + 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: + """Read an unsigned byte. + + Raises: + ExhaustedStreamError: + If there are no enough bytes. + """ + return int.from_bytes(self.bytes(1), byteorder="little", 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="little", signed=False) + + def skip(self, alignment: int, force: bool = False) -> 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 or force: + _ = self.bytes(r or alignment) + + +class Writer: + def __init__(self, fp: "SupportsWrite[bytes]") -> None: + self._fp: "SupportsWrite[bytes]" = fp + self._emitted: int = 0 + + @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: + 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: + """Write unsigned byte `value` to the stream. + + Raises: + FailedWriteError: + If write hasn't succeeded. + """ + 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. + + Raises: + FailedWriteError: + If write hasn't succeeded. + """ + 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`.""" + 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 new file mode 100644 index 00000000..79a90e94 --- /dev/null +++ b/cocas/ple/constants.py @@ -0,0 +1,13 @@ +"""Plain Executalbe constants.""" + +__all__ = ( + "MAGIC_BYTES", + "SECTOR_SIZE", + "PARAGRAPH_SIZE", +) + +from typing import Final + +MAGIC_BYTES: Final[bytes] = b"\x7fPLE" +SECTOR_SIZE: Final[int] = 512 +PARAGRAPH_SIZE: Final[int] = 16 diff --git a/cocas/ple/exceptions.py b/cocas/ple/exceptions.py new file mode 100644 index 00000000..5e96b3f2 --- /dev/null +++ b/cocas/ple/exceptions.py @@ -0,0 +1,43 @@ +"""Plain Executable exceptions.""" + +__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/types.py b/cocas/ple/types.py new file mode 100644 index 00000000..76b9f478 --- /dev/null +++ b/cocas/ple/types.py @@ -0,0 +1,170 @@ +__all__ = ( + "PleImage", + "PleSegmentEntry", + "PleSegmentType", + "PleSegmentFlag", +) + +import bisect +from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import IntEnum, IntFlag, auto +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 + +if TYPE_CHECKING: + from _typeshed import SupportsRead, SupportsWrite + +PleImageT = TypeVar("PleImageT", bound="PleImage") +SENTINEL = object() + + +class PleSegmentType(IntEnum): + NONE = 0x0 + LOAD = 0x1 + + +class PleSegmentFlag(IntFlag): + DATA = auto() + WRITE_PROTECTED = auto() + + +@stdint.validate +@dataclass +class PleSegmentEntry: + 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_div(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 PleImage: + 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: PleImageT, segment: PleSegmentEntry) -> PleImageT: + 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 + + 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_div(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_div(s.physical_size * PARAGRAPH_SIZE, 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_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, force=True) + 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() + 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]) + + 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 * SECTOR_SIZE:] + 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_div(size_delta, SECTOR_SIZE) * 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)) diff --git a/tests/tests_ple/ple.bin b/tests/tests_ple/ple.bin new file mode 100644 index 00000000..bbc7bc0a Binary files /dev/null and b/tests/tests_ple/ple.bin differ 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()