#!/usr/bin/env python
__version__ = "0.1.1"
__copyright__ = """
objutils - Object file library for Python.
(C) 2010-2025 by Christoph Schueler <cpu12.gems@googlemail.com>
All Rights Reserved
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License along
with this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""
import math
import os
import re
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from functools import partial
from operator import itemgetter
from typing import Any, Dict, Iterable, Iterator, List, Mapping, MutableMapping, Optional, Sequence, Tuple, Union
from objutils.image import Image
from objutils.logger import Logger
from objutils.section import Section, join_sections
from objutils.utils import create_string_buffer, slicer
"""
MemoryBlocks
------------
Segment-Types: generic,code,const,bss,... (s. ELF!?)
"""
"""
TODO: Create a base class for Reader/Writer common functionality.
FactoryMethods ???
Parameter ['exceptionsOnErrors'].
"""
SIXTEEN_BITS = 0
TWENTY_BITS = 1
TWENTYFOUR_BITS = 2
THIRTYTWO_BITS = 3
START = 0
LENGTH = 1
TYPE = 2
ADDRESS = 3
DATA = 4
UNPARSED = 5
CHECKSUM = 6
ADDR_CHECKSUM = 7
MAP_GROUP_TO_REGEX = {
LENGTH: r"(?P<length>[0-9a-zA-Z]{%d})",
TYPE: r"(?P<type>\d{%d})",
ADDRESS: r"(?P<address>[0-9a-zA-Z]{%d})",
DATA: r"(?P<chunk>[0-9a-zA-Z]+)",
UNPARSED: r"(?P<chunk>.*)",
CHECKSUM: r"(?P<checksum>[0-9a-zA-Z]{%d})",
ADDR_CHECKSUM: r"(?P<addrChecksum>[0-9a-zA-Z]{%d})",
}
MAP_CHAR_TO_GROUP = {
"L": LENGTH,
"T": TYPE,
"A": ADDRESS,
"D": DATA,
"U": UNPARSED,
"C": CHECKSUM,
"B": ADDR_CHECKSUM,
}
TYPE_FROM_RECORD = 0
atoi = partial(int, base=16)
[docs]
@dataclass
class Statistics:
record_types: Counter[int] = field(default_factory=Counter)
data_bytes: Counter[int] = field(default_factory=Counter)
[docs]
class Invalidrecord_typeError(Exception):
pass
[docs]
class InvalidRecordLengthError(Exception):
pass
[docs]
class InvalidRecordChecksumError(Exception):
pass
[docs]
class AddressRangeToLargeError(Exception):
pass
[docs]
class Container:
def __init__(self) -> None:
self.processing_instructions: List[Any] = []
[docs]
def add_processing_instruction(self, pi: Any) -> None:
self.processing_instructions.append(pi)
[docs]
class BaseType:
logger: Logger
valid: bool
[docs]
def error(self, msg: str) -> None:
self.logger.error(msg)
self.valid = False
[docs]
def warn(self, msg: str) -> None:
self.logger.warn(msg)
[docs]
def info(self, msg: str) -> None:
self.logger.info(msg)
[docs]
def debug(self, msg: str) -> None:
self.logger.debug(msg)
[docs]
class Reader(BaseType):
ALIGMENT = 0 # 2**n
DATA_SEPARATOR: Optional[str] = None
VALID_CHARS = re.compile(r"^[a-fA-F0-9 :/;,%\n\r!?S]*$") # General case, fits most formats.
def __init__(self) -> None:
self.logger = Logger("Reader")
self.stats = Statistics()
# formats is a list of (format_type, compiled_pattern)
if isinstance(self.FORMAT_SPEC, str):
self.formats: List[Tuple[int, re.Pattern[str]]] = [(0, FormatParser(self.FORMAT_SPEC, self.DATA_SEPARATOR).parse())]
elif isinstance(self.FORMAT_SPEC, (list, tuple)):
self.formats = []
for format_type, format in self.FORMAT_SPEC:
self.formats.append((format_type, FormatParser(format, self.DATA_SEPARATOR).parse()))
else:
self.formats = []
[docs]
def load(self, fp: Union[str, Any], **kws: Any) -> Image:
if isinstance(fp, str):
fp = open(fp, "rb")
data = self.read(fp) # .decode()
if hasattr(fp, "close"):
fp.close()
return data
[docs]
def loads(self, image: Union[str, bytes, bytearray], **kws: Any) -> Image:
if isinstance(image, str):
return self.load(create_string_buffer(bytes(image, "ascii")))
else:
return self.load(create_string_buffer(image))
[docs]
def read(self, fp: Any) -> Image:
sections = []
matched = False
self.valid = True
meta_data = defaultdict(list)
for line_number, line in enumerate(fp.readlines(), 1):
for format_type, format in self.formats:
if isinstance(line, bytes):
match = format.match(line.decode())
else:
match = format.match(line)
if match:
matched = True
container = Container()
container.line_number = line_number
dict_ = match.groupdict()
if dict_ != {}:
self.stats.record_types[format_type] += 1
# Handle scalar values.
for key, value in dict_.items():
if key not in ("chunk", "junk"):
setattr(container, key, atoi(value))
elif key == "junk":
setattr(container, key, value)
if "chunk" in dict_:
if self.parseData(container, format_type):
chunk = bytearray.fromhex(dict_["chunk"])
else:
# don't convert/parse stuff like symbols.
chunk = dict_["chunk"]
dict_.pop("chunk")
container.chunk = chunk # setattr(container, "chunk", chunk)
self.stats.data_bytes[len(chunk)] += 1
self.check_line(container, format_type)
# this is to handle esoteric stuff like Intel seg:offs addressing and symbols.
self.special_processing(container, format_type)
if self.is_data_line(container, format_type):
sections.append(Section(container.address, container.chunk))
else:
chunk = container.chunk if hasattr(container, "chunk") else None
address = container.address if hasattr(container, "address") else None
meta_data[format_type].append(MetaRecord(format_type, address, chunk))
break
if not matched:
self.warn(f"Ignoring garbage line #{line_number:d}")
if sections:
return Image(sections=join_sections(sections), meta=meta_data)
else:
self.error("File seems to be invalid.")
return Image([], valid=False)
[docs]
def _address_space(self, value: int) -> int:
if value < 2**16:
return SIXTEEN_BITS
elif value < 2**20:
return TWENTY_BITS
elif value < 2**24:
return TWENTYFOUR_BITS
elif value < 2**32:
return THIRTYTWO_BITS
else:
raise ValueError("Unsupported Addressspace size.")
[docs]
def maybe_binary_file(self, fp: Any) -> bool:
fp.seek(0, os.SEEK_SET)
header = fp.read(128)
fp.seek(0, os.SEEK_SET)
try:
hdr = header.decode(errors="ignore")
except Exception:
hdr = ""
result = not bool(self.VALID_CHARS.match(hdr))
return result
[docs]
def probe(self, fp: Any) -> bool:
"""Determine if object is likely valid for this codec.
Strategy (robust, production-grade):
- Quickly reject binary-looking content.
- Sample up to 50 non-empty lines (or 8KB), decoding safely.
- Count how many lines match this codec's patterns.
- Accept when we see sufficient evidence (>= 2 matches), or when
there's exactly one non-empty sampled line and it matches (tiny files).
- Always restore the file pointer to the beginning before returning.
"""
# Always start from beginning
try:
fp.seek(0, os.SEEK_SET)
except Exception:
pass
if self.maybe_binary_file(fp):
try:
fp.seek(0, os.SEEK_SET)
except Exception:
pass
return False
matched_count = 0
examined = 0
max_lines = 50
max_bytes = 8192
consumed = 0
# Iterate lines safely, up to limits
for line_number, raw in enumerate(fp.readlines(), 1):
if isinstance(raw, bytes):
try:
line = raw.decode(errors="ignore")
except Exception:
line = ""
else:
line = raw
consumed += len(raw) if isinstance(raw, (bytes, bytearray)) else len(line)
if consumed > max_bytes:
break
s = line.strip()
if not s:
continue # skip empty/blank lines
examined += 1
# If the line has clearly invalid characters for ASCII-hex records, reject fast.
if not self.VALID_CHARS.match(s):
try:
fp.seek(0, os.SEEK_SET)
except Exception:
pass
return False
# Try all known formats for this codec
this_line_matched = False
for _format_type, fmt in self.formats:
if fmt.match(s):
this_line_matched = True
break
if this_line_matched:
matched_count += 1
if matched_count >= 3:
break # strong signal; no need to continue
if examined >= max_lines:
break
# Reset file pointer before returning
try:
fp.seek(0, os.SEEK_SET)
except Exception:
pass
if matched_count >= 2:
return True
# Allow tiny one-line files (headers or footers) when it's the only sampled line
if matched_count == 1 and examined == 1:
return True
return False
[docs]
def probes(self, image: Union[str, bytes, bytearray]) -> bool:
if isinstance(image, str):
return self.probe(create_string_buffer(bytes(image, "ascii")))
else:
return self.probe(create_string_buffer(image))
[docs]
def check_line(self, line: Any, format_type: int) -> None:
raise NotImplementedError()
[docs]
def is_data_line(self, line: Any, format_type: int) -> bool:
raise NotImplementedError()
[docs]
def classifyLine(self, line: Any) -> int:
raise NotImplementedError()
[docs]
def special_processing(self, line: Any, format_type: int) -> None:
pass
[docs]
def parseData(self, line: Any, format_type: int) -> bool:
return True
[docs]
class Writer(BaseType):
def __init__(self) -> None:
self.logger = Logger("Writer")
[docs]
def dump(self, fp: Union[str, Any], image: Image, row_length: int = 16, **kws: Any) -> None: # TODO: rename to bytesPerRow!
if isinstance(fp, str):
fp = open(fp, "wb")
fp.write(self.dumps(image, row_length))
if hasattr(fp, "close"):
fp.close()
[docs]
def dumps(self, image: Image, row_length: int = 16, **kws: Any) -> bytes:
result: List[str] = []
self.row_length = row_length
if hasattr(image, "sections") and not image.sections:
return b""
if self.calculate_address_bits(image) > self.MAX_ADDRESS_BITS:
raise AddressRangeToLargeError("could not encode image.")
self.set_parameters(**kws)
self.pre_processing(image)
header = self.compose_header(image.meta if hasattr(image, "meta") else {})
if header:
result.append(header)
for section in image:
address = section.start_address
rows = slicer(section.data, row_length, lambda x: [int(y) for y in x])
for row in rows:
length = len(row)
result.append(self.compose_row(address, length, row))
address += row_length
footer = self.compose_footer(image.meta if hasattr(image, "meta") else {})
if footer:
result.append(footer)
return self.post_processing(bytes("\n".join(result), "ascii"))
[docs]
def calculate_address_bits(self, image: Image) -> int:
if hasattr(image, "sections"):
last_segment = sorted(image.sections, key=lambda s: s.start_address)[-1]
else:
last_segment = image
highest_address = last_segment.start_address + last_segment.length
return int(math.ceil(math.log(highest_address + 1) / math.log(2)))
[docs]
def post_processing(self, data: bytes) -> bytes:
if not data.endswith(b"\n"):
data += b"\n"
return data
[docs]
def pre_processing(self, image: Image) -> None:
pass
[docs]
def set_parameters(self, **kws: Any) -> None:
params: Dict[str, Any] = {}
for k, v in kws.items():
try:
params[k] = getattr(self, k)
except AttributeError as e:
raise AttributeError(f"Invalid keyword argument {k!r}.") from e
else:
setattr(self, k, v)
[docs]
def compose_row(self, address: int, length: int, row: Sequence[int]) -> str:
raise NotImplementedError()
[docs]
def word_to_bytes(self, word: int) -> Tuple[int, int]:
word = int(word)
hi = (word & 0xFF00) >> 8
lo = word & 0x00FF
return hi, lo
[docs]
@staticmethod
def hex_bytes(row: Sequence[int], spaced: bool = False) -> str:
spacer = " " if spaced else ""
return spacer.join([f"{x:02X}" for x in row])
[docs]
class ASCIIHexReader(Reader):
FORMAT_SPEC = None
def __init__(self, address_pattern: str, data_pattern: str, etx_pattern: str, separators: str = ", "):
self.separators = separators
self.DATA_PATTERN: re.Pattern[str] = re.compile(data_pattern.format(separators), re.DOTALL | re.MULTILINE)
self.ADDRESS_PATTERN: re.Pattern[str] = re.compile(address_pattern, re.DOTALL | re.MULTILINE)
self.ETX_PATTERN: re.Pattern[str] = re.compile(etx_pattern, re.DOTALL | re.MULTILINE)
self.SPLITTER = re.compile(f"[{separators}]")
self.patterns: Sequence[Tuple[re.Pattern[str], Any]] = (
(self.ADDRESS_PATTERN, self.getAddress),
(self.DATA_PATTERN, self.parse_line),
(self.ETX_PATTERN, self.nop),
)
self.formats: List[Tuple[int, re.Pattern[str]]] = [
(0, self.ADDRESS_PATTERN),
(1, self.DATA_PATTERN),
(2, self.ETX_PATTERN),
]
super().__init__()
[docs]
def getAddress(self, line: str, match: re.Match[str]) -> bool:
self.address = int(match.group("address"), 16)
self.previous_address = self.address
return True
[docs]
def nop(self, line: str, match: re.Match[str]) -> bool:
return False
[docs]
def parse_line(self, line: str, match: re.Match[str]) -> bool:
section = Section(
self.address,
bytearray([int(ch, 16) for ch in filter(lambda x: x, self.SPLITTER.split(line))]),
)
self.sections.append(section)
self.address += len(section)
return True
[docs]
def read(self, fp: Any) -> Image:
lines = fp.read().decode()
self.sections: List[Section] = []
self.address = 0
breakRequest = False
for line in lines.splitlines():
for pattern, action in self.patterns:
match = pattern.match(line)
if match:
if not action(line, match):
breakRequest = True
break
if breakRequest:
break
return Image(join_sections(self.sections))
[docs]
class ASCIIHexWriter(Writer):
MAX_ADDRESS_BITS = 16
previous_address: Optional[int] = None
def __init__(self, address_designator: str):
self.separator = " "
self.address_designator = address_designator
super().__init__()
[docs]
def compose_row(self, address: int, length: int, row: Sequence[int]) -> str:
prepend_address = True if address != self.previous_address else False
self.previous_address = address + length
if prepend_address:
line = "{}\n{}".format(
f"{self.address_designator}{address:04X}",
f"{self.separator}".join([f"{x:02X}" for x in row]),
)
else:
line = " ".join([f"{x:02X}" for x in row])
self.row_callout(address, length, row)
return line
[docs]
def row_callout(self, address: int, length: int, row: Sequence[int]) -> None:
pass