"""Little-endian binary reading/writing helpers shared by the IDAT, BPM, EGT and GTC codecs. All Illumina binary formats are little-endian. Strings are .NET ``BinaryWriter`` strings: a 7-bit varint (LEB128) length prefix followed by UTF-8 bytes (plan ยง6; BeadArrayFiles ``read_string``, BSD-2). Decoding is tolerant (spec IN-033): invalid UTF-8 becomes U+FFFD, while a length running past the end of the buffer is a :class:`~simurg.core.errors.FormatError`, never an uncaught exception. """ from __future__ import annotations import gzip import io import mmap import struct from pathlib import Path from typing import Any import numpy as np import numpy.typing as npt from simurg.core.errors import FormatError _I8 = struct.Struct(" bytes | mmap.mmap: """Return the file content: memory-mapped when plain, decompressed when gzip.""" p = Path(path) with p.open("rb") as fh: head = fh.read(2) if head == GZIP_MAGIC: fh.seek(0) with gzip.GzipFile(fileobj=fh) as gz: return gz.read() fh.seek(0, io.SEEK_END) if fh.tell() == 0: return b"" return mmap.mmap(fh.fileno(), 0, access=mmap.ACCESS_READ) class BinaryReader: """Bounds-checked sequential reader over an immutable buffer. ``err_code`` is the catalogue code raised on malformed input (e.g. ``E_IDAT_FORMAT``). """ __slots__ = ("_buf", "_view", "err_code", "path", "pos", "size") def __init__(self, buf: bytes | mmap.mmap | memoryview, err_code: str, path: str | None = None) -> None: self._buf = buf self._view = memoryview(buf) self.size = len(self._view) self.pos = 0 self.err_code = err_code self.path = path # -- errors ------------------------------------------------------------------------------------ def fail(self, message: str) -> FormatError: where = f"{self.path}: " if self.path else "" return FormatError(self.err_code, f"{where}{message}", path=self.path) def _need(self, n: int, what: str) -> int: if n < 0 or self.pos + n > self.size: raise self.fail(f"unexpected end of file reading {what} at offset {self.pos} (need {n} bytes)") start = self.pos self.pos += n return start # -- positioning ------------------------------------------------------------------------------- def seek(self, offset: int, what: str = "section") -> None: if offset < 0 or offset > self.size: raise self.fail(f"{what} offset {offset} outside file (size {self.size})") self.pos = offset def remaining(self) -> int: return self.size - self.pos # -- scalars ----------------------------------------------------------------------------------- def _unpack(self, s: struct.Struct, what: str) -> Any: start = self._need(s.size, what) return s.unpack_from(self._view, start)[0] def i8(self, what: str = "int8") -> int: return int(self._unpack(_I8, what)) def u8(self, what: str = "uint8") -> int: return int(self._unpack(_U8, what)) def i16(self, what: str = "int16") -> int: return int(self._unpack(_I16, what)) def u16(self, what: str = "uint16") -> int: return int(self._unpack(_U16, what)) def i32(self, what: str = "int32") -> int: return int(self._unpack(_I32, what)) def u32(self, what: str = "uint32") -> int: return int(self._unpack(_U32, what)) def i64(self, what: str = "int64") -> int: return int(self._unpack(_I64, what)) def f32(self, what: str = "float32") -> float: return float(self._unpack(_F32, what)) def raw(self, n: int, what: str = "bytes") -> bytes: start = self._need(n, what) return bytes(self._view[start : start + n]) # -- strings ----------------------------------------------------------------------------------- def varint_length(self, what: str = "string length") -> int: """.NET 7-bit encoded int (at most 5 bytes).""" result = 0 shift = 0 for _ in range(5): b = self.u8(what) result |= (b & 0x7F) << shift if not b & 0x80: return result shift += 7 raise self.fail(f"invalid 7-bit length prefix for {what} at offset {self.pos}") def string(self, what: str = "string") -> str: n = self.varint_length(what) start = self._need(n, what) return bytes(self._view[start : start + n]).decode("utf-8", errors="replace") def skip_string(self, what: str = "string") -> None: n = self.varint_length(what) self._need(n, what) # -- arrays ------------------------------------------------------------------------------------ def array(self, dtype: npt.DTypeLike, count: int, what: str = "array") -> npt.NDArray[Any]: """Zero-copy little-endian array view (read-only).""" dt = np.dtype(dtype).newbyteorder("<") if count < 0: raise self.fail(f"negative element count {count} for {what}") start = self._need(count * dt.itemsize, what) return np.frombuffer(self._view, dtype=dt, count=count, offset=start) def counted_array(self, dtype: npt.DTypeLike, what: str = "array") -> npt.NDArray[Any]: """int32 element count followed by the elements.""" return self.array(dtype, self.i32(f"{what} count"), what) class BinaryWriter: """Little-endian writer into an in-memory buffer.""" def __init__(self) -> None: self._buf = io.BytesIO() def tell(self) -> int: return self._buf.tell() def getvalue(self) -> bytes: return self._buf.getvalue() def write(self, b: bytes) -> None: self._buf.write(b) def u8(self, v: int) -> None: self._buf.write(_U8.pack(v)) def i16(self, v: int) -> None: self._buf.write(_I16.pack(v)) def u16(self, v: int) -> None: self._buf.write(_U16.pack(v)) def i32(self, v: int) -> None: self._buf.write(_I32.pack(v)) def u32(self, v: int) -> None: self._buf.write(_U32.pack(v)) def f32(self, v: float) -> None: self._buf.write(_F32.pack(v)) def string(self, s: str | bytes) -> None: data = s.encode("utf-8") if isinstance(s, str) else s n = len(data) while True: b = n & 0x7F n >>= 7 if n: self._buf.write(bytes([b | 0x80])) else: self._buf.write(bytes([b])) break self._buf.write(data) def array(self, arr: npt.ArrayLike, dtype: npt.DTypeLike) -> None: self._buf.write(np.ascontiguousarray(arr, dtype=np.dtype(dtype).newbyteorder("<")).tobytes()) def counted_array(self, arr: npt.ArrayLike, dtype: npt.DTypeLike) -> None: a = np.asarray(arr) self.i32(len(a)) self.array(a, dtype)