Update section handling

Author: christoph2

objutils/__init__.py

@@ -13,7 +13,7 @@
 The first parameter is always the codec name.
 """
 
-__version__ = "0.9.0"
+__version__ = "0.9.1"
 
 __all__ = [
     "Image",

objutils/ieee695.py

@@ -524,12 +524,12 @@ def onE2(self):
         elif discr == ASG:
             delim = self.readByte(self.fpos)
             if delim != 0xBE:
-                pass  # todo: FormatError!!!
+                raise hexfile.FormatError(f"Expected 0xBE, got 0x{delim:02X} at 0x{self.fpos:X}")
             executionStartingAddr = self.readNumber(self.fpos)
             self.logger.debug(f"STARTING-ADDRESS: 0x{executionStartingAddr:04X}")
             delim = self.readByte(self.fpos)
             if delim != 0xBF:
-                pass  # todo: FormatError!!!
+                raise hexfile.FormatError(f"Expected 0xBF, got 0x{delim:02X} at 0x{self.fpos:X}")
             self.info.executionStartingAddr = executionStartingAddr
         else:
             raise NotImplementedError(hex(discr))
@@ -587,7 +587,7 @@ def onF1(self):
             elif attrDef == 19:
                 pass
             else:
-                pass  # todo: FormatError!!!
+                raise hexfile.FormatError(f"Invalid symbol attribute at 0x{self.fpos:X}")
             ##            n4 =self.readNumber(self.fpos) # If n2 is non-zero, number of elements in the symbol type specified in n2
             if symbolTypeIndex != 0:
                 numberOfElements = self.readNumber(self.fpos)
@@ -873,7 +873,7 @@ def onTY(self):
         """"""
         typeIndex = self.readNumber(self.fpos)  # noqa: F841
         if self.readByte(self.fpos) != 0xCE:
-            pass  # todo: raise FormatError!!
+            raise hexfile.FormatError(f"Expected 0xCE at 0x{self.fpos:X}")
         localNameIndex = self.readNumber(self.fpos)  # noqa: F841
         values = []
         while True:

objutils/image.py

@@ -421,6 +421,8 @@ def __init__(
             sections = list(sections)
         else:
             raise TypeError(f"Argument section is of wrong type {sections!r}")
+        for section in sections:
+            section._parent_image = self
         self._sections = sorted(sections, key=attrgetter("start_address"))
         self._join = join
         if join:
@@ -832,6 +834,17 @@ def join_sections(self) -> None:
         into single sections, reducing fragmentation and improving efficiency.
         """
         self._sections = join_sections(self._sections)
+        for section in self._sections:
+            section._parent_image = self
+
+    def _validate_address_change(self, section: Section, new_address: int) -> None:
+        """Validate that changing a section's address doesn't cause overlaps."""
+        # Temporary remove section to check against others
+        others = [s for s in self._sections if s is not section]
+        for s in others:
+            # Check if new range [new_address, new_address + len(section)) overlaps with s
+            if not (new_address + len(section) <= s.start_address or new_address >= s.start_address + len(s)):
+                raise InvalidAddressError(f"New address 0x{new_address:08x} causes overlap with section at 0x{s.start_address:08x}")
 
     def split(self, at: Optional[int] = None, equal_parts: Optional[int] = None, remap: Optional[bool] = None) -> None:
         """Split image into multiple parts.

objutils/scripts/oj_hex_info.py

@@ -27,31 +27,17 @@
 import sys
 from os import path
 
-from objutils import load
+from objutils import load, probe
 
 
 def main():
     parser = argparse.ArgumentParser(description="Displays informations about HEX files.")
     parser.add_argument(
-        "file_type",
-        help="file type",
-        choices=[
-            "ash",
-            "cosmac",
-            "emon52",
-            "etek",
-            "fpc",
-            "ihex",
-            "mostec",
-            "rca",
-            "shf",
-            "sig",
-            "srec",
-            "tek",
-            "titxt",
-        ],
+        "file_or_type",
+        nargs="?",
+        help="file type or HEX file (if type is omitted)",
     )
-    parser.add_argument("hex_file", help="HEX file")
+    parser.add_argument("hex_file", nargs="?", help="HEX file")
     parser.add_argument(
         "-d",
         "--dump",
@@ -79,15 +65,32 @@ def main():
 
     args = parser.parse_args()
 
-    if not path.exists(args.hex_file):
-        print(f"File '{args.hex_file}' does not exist.")
+    if args.hex_file is None:
+        if args.file_or_type is None:
+            parser.error("too few arguments")
+        # Only one positional argument given: it's the hex_file, type is to be probed.
+        hex_file = args.file_or_type
+        file_type = None
+    else:
+        # Two positional arguments given: type and file.
+        file_type = args.file_or_type
+        hex_file = args.hex_file
+
+    if not path.exists(hex_file):
+        print(f"File '{hex_file}' does not exist.")
+        sys.exit(1)
+
+    if file_type is None:
+        with open(hex_file, "rb") as f:
+            file_type = probe(f)
+
+    if file_type is None:
+        print(f"Could not determine file type for '{hex_file}'.")
         sys.exit(1)
 
-    img = load(args.file_type.lower(), args.hex_file)
+    img = load(file_type.lower(), hex_file, join=args.join_section)
     if args.print_filename:
-        print(f"\nFile: {args.hex_file}")
-    if args.join_section:
-        img.join_sections()
+        print(f"\nFile: {hex_file}")
     print("\nSections")
     print("--------\n")
     print("Num   Address    Length")

objutils/section.py

@@ -184,7 +184,7 @@
 from array import array
 from collections import namedtuple
 from copy import copy
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from functools import reduce
 from operator import attrgetter, mul
 from typing import Any, TextIO, Union
@@ -435,15 +435,32 @@ class Section:
         are equal if they have the same start address, data, and name.
     """
 
-    start_address: int = field(hash=True, compare=True, default=0)
-    data: bytearray = field(default_factory=bytearray, compare=True, hash=True)
-    name: str = field(default="", compare=True, hash=True)
-
-    def __post_init__(self):
+    def __init__(self, start_address: int = 0, data: Any = None, name: str = ""):
+        self._start_address = start_address
+        self.data = _data_converter(data if data is not None else bytearray())
+        self.name = name
+        self._parent_image = None
         self.repr = reprlib.Repr()
         self.repr.maxstring = 64
         self.repr.maxother = 64
-        self.data = _data_converter(self.data)
+
+    @property
+    def start_address(self) -> int:
+        return self._start_address
+
+    @start_address.setter
+    def start_address(self, value: int) -> None:
+        if not isinstance(value, int):
+            raise TypeError("start_address must be of type int")
+        if value < 0:
+            raise ValueError("start_address must be >= 0")
+        if hasattr(self, "_parent_image") and self._parent_image:
+            self._parent_image._validate_address_change(self, value)
+        self._start_address = value
+
+    def __post_init__(self):
+        # We handle initialization in __init__ now.
+        pass
 
     def __iter__(self):
         yield self
@@ -791,7 +808,7 @@ def __repr__(self) -> str:
         return "Section(address = 0X{:08X}, length = {:d}, data = {})".format(
             self.start_address,
             self.length,
-            self.repr.repr(memoryview(self.data).tobytes()),
+            self.repr.repr(bytes(self.data)),
         )
 
     def __len__(self) -> int:
@@ -808,6 +825,63 @@ def __contains__(self, addr) -> bool:
     def address(self) -> int:  # Alias
         return self.start_address
 
+    @address.setter
+    def address(self, value: int) -> None:
+        self.start_address = value
+
+
+class LazySection(Section):
+    """Memory-mapped Section for large binary files.
+
+    LazySection uses `mmap` to map a file into memory instead of loading
+    everything into RAM. This is more efficient for very large files.
+    """
+
+    def __init__(self, start_address: int, filename: str, offset: int = 0, length: int = -1, name: str = ""):
+        import mmap
+        import os
+
+        self._start_address = start_address
+        self.name = name
+        self.filename = filename
+        self._file = open(filename, "rb")
+        if length == -1:
+            length = os.path.getsize(filename) - offset
+
+        self.data = mmap.mmap(self._file.fileno(), length, offset=offset, access=mmap.ACCESS_READ)
+        self.__post_init__()
+
+    def __post_init__(self):
+        super().__post_init__()
+
+    def write(self, addr: int, data: bytes, **kws) -> None:
+        raise NotImplementedError("LazySection is read-only")
+
+    def write_numeric(self, addr: int, value: Union[int, float], dtype: str, **kws) -> None:
+        raise NotImplementedError("LazySection is read-only")
+
+    def write_numeric_array(self, addr: int, data: Union[list[int], list[float]], dtype: str, **kws) -> None:
+        raise NotImplementedError("LazySection is read-only")
+
+    def write_string(self, addr: int, value: str, encoding: str = "latin1", **kws):
+        raise NotImplementedError("LazySection is read-only")
+
+    def write_ndarray(self, addr: int, array: np.ndarray, order: str = None, **kws) -> None:
+        raise NotImplementedError("LazySection is read-only")
+
+    def __del__(self):
+        """Close memory-mapped file when section is destroyed."""
+        try:
+            if hasattr(self, "data") and self.data:
+                self.data.close()
+        except (AttributeError, ValueError):
+            pass
+        try:
+            if hasattr(self, "_file") and self._file:
+                self._file.close()
+        except (AttributeError, ValueError):
+            pass
+
 
 def join_sections(sections: list[Section]) -> list[Section]:
     """Join consecutive sections into contiguous blocks.

objutils/tests/test_hexfile.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 import unittest
 
-from objutils import hexfile, loads
+from objutils import hexfile, loads, dumps, Section
 from objutils.image import Image
 
 
@@ -28,6 +28,34 @@ def testRaisesError(self):
         #        self.assertRaises(hexfile.InvalidRecordChecksumError, loads, "srec", b'S110000048656C6C6F20776F726C642166')
         pass
 
+    def test_reader_join_flag_preserves_sections(self):
+        """Ensure Reader join flag keeps original section boundaries."""
+        sec_a = Section(0xFFE0, b"\x01\x02\x03\x04\x05\x06\x07\x08")
+        sec_b = Section(0xFFE8, b"\x11\x12\x13\x14\x15\x16\x17\x18")
+        img = Image([sec_a, sec_b], join=False)
+
+        ihex_bytes = dumps("ihex", img)
+        roundtrip = loads("ihex", ihex_bytes, join=False)
+
+        self.assertEqual(len(roundtrip.sections), 2)
+        self.assertEqual(roundtrip.sections[0].start_address, 0xFFE0)
+        self.assertEqual(roundtrip.sections[0].data, sec_a.data)
+        self.assertEqual(roundtrip.sections[1].start_address, 0xFFE8)
+        self.assertEqual(roundtrip.sections[1].data, sec_b.data)
+
+    def test_reader_defaults_to_no_join(self):
+        """Default reader should preserve sections without merging."""
+        sec_a = Section(0xFFE0, b"\x01" * 8)
+        sec_b = Section(0xFFE8, b"\x02" * 8)
+        img = Image([sec_a, sec_b], join=False)
+
+        ihex_bytes = dumps("ihex", img)
+        roundtrip = loads("ihex", ihex_bytes)
+
+        self.assertEqual(len(roundtrip.sections), 2)
+        self.assertEqual(roundtrip.sections[0].length, 8)
+        self.assertEqual(roundtrip.sections[1].length, 8)
+
 
 def main():
     unittest.main()

objutils/version.py

@@ -1,3 +1,3 @@
 """objutils version module"""
 
-__version__ = "0.9.0"
+__version__ = "0.9.1"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "objutils"
-version = "0.9.0"
+version = "0.9.1"
 description = "Objectfile library for Python"
 authors = ["Christoph Schueler <cpu12.gems@googlemail.com>"]
 license = "GPLv2"
@@ -121,7 +121,7 @@ skip = ["*_i686", "*-musllinux*"]
 build-frontend = "build"
 
 [tool.bumpver]
-current_version = "0.9.0"
+current_version = "0.9.1"
 version_pattern = "MAJOR.MINOR.PATCH"
 commit_message = "bump version {old_version} -> {new_version}"
 commit = true