diff --git a/src/spmkit/core/io/__init__.py b/src/spmkit/core/io/__init__.py index fc17d92..8a83f81 100644 --- a/src/spmkit/core/io/__init__.py +++ b/src/spmkit/core/io/__init__.py @@ -7,6 +7,7 @@ from spmkit.core.io.forceload import load_force, supported_force_extensions from spmkit.core.io.gwy import load_gwy, save_gwy +from spmkit.core.io.igor_ibw import load_igor_ibw from spmkit.core.io.jpk import load_jpk_force from spmkit.core.io.loadany import inspect_any, load_any from spmkit.core.io.nhf import load_nhf @@ -21,6 +22,7 @@ "load_nid_force", "load_nhf", "load_gwy", + "load_igor_ibw", "save_gwy", "load_jpk_force", "load_force", diff --git a/src/spmkit/core/io/igor_ibw.py b/src/spmkit/core/io/igor_ibw.py new file mode 100644 index 0000000..b5af22d --- /dev/null +++ b/src/spmkit/core/io/igor_ibw.py @@ -0,0 +1,298 @@ +"""Limited native reader for observed Igor Binary Wave v5 AFM image files. + +This reader intentionally supports only the structural family exercised by the +native-IBW pilot: little-endian v5 headers, scalar FP32 samples, and 2D image +channels stored in the third dimension. It does not claim general ``.ibw`` +support and leaves unsupported variants available to optional readers. +""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import numpy as np + +from spmkit.core.models import SPMChannel, SPMData +from spmkit.core.plugins.contracts import DatasetInfo + +_BIN_HEADER_SIZE: Final = 64 +_WAVE_HEADER_SIZE: Final = 320 +_HEADER_SIZE: Final = _BIN_HEADER_SIZE + _WAVE_HEADER_SIZE +_IGOR_V5: Final = 5 +_IGOR_SINGLE: Final = 0x02 +_LABEL_SIZE: Final = 32 +_UNIT_TO_METERS: Final = { + "m": 1.0, + "mm": 1e-3, + "um": 1e-6, + "µm": 1e-6, + "nm": 1e-9, + "pm": 1e-12, +} +_TITLE_UNITS: Final = ( + ("Height", "m"), + ("ZSensor", "m"), + ("Deflection", "m"), + ("Amplitude", "m"), + ("Phase", "deg"), + ("Current", "A"), + ("Frequency", "Hz"), + ("Capacitance", "F"), + ("Potential", "V"), +) + + +class IbwFormatError(ValueError): + """Malformed or unsupported Igor Binary Wave data.""" + + +@dataclass(frozen=True) +class _Header: + dimensions: tuple[int, int, int, int] + scales: tuple[float, float, float, float] + data_unit: str + dimension_units: tuple[str, str, str, str] + data_end: int + extra_start: int + formula_size: int + note_size: int + data_e_units_size: int + dimension_e_units_sizes: tuple[int, int, int, int] + dimension_label_sizes: tuple[int, int, int, int] + + +def _read_c_string(value: bytes) -> str: + return value.split(b"\0", 1)[0].decode("latin-1", "replace").strip() + + +def _span(blob: bytes, start: int, size: int, label: str) -> bytes: + end = start + size + if start < 0 or size < 0 or end > len(blob): + raise IbwFormatError(f"IBW v5 truncated: {label} exceeds file size") + return blob[start:end] + + +def _checksum_ok(header: bytes) -> bool: + if len(header) != _HEADER_SIZE: + return False + words = struct.unpack(f"<{len(header) // 2}H", header) + return sum(words) & 0xFFFF == 0 + + +def _parse_header(blob: bytes, *, validate_payload: bool = True) -> _Header: + if len(blob) < _HEADER_SIZE: + raise IbwFormatError("IBW v5 truncated before the binary headers") + if blob[0] == 0: + raise IbwFormatError("IBW big-endian files are not supported by this native reader") + version = struct.unpack_from(" tuple[list[str], dict[str, str]]: + position = header.extra_start + position += header.formula_size + note_bytes = _span(blob, position, header.note_size, "note") + position += header.note_size + header.data_e_units_size + sum(header.dimension_e_units_sizes) + for size in header.dimension_label_sizes[:2]: + _span(blob, position, size, "lower-dimension labels") + position += size + channel_size = header.dimension_label_sizes[2] + channel_blob = _span(blob, position, channel_size, "channel labels") + if channel_size % _LABEL_SIZE: + raise IbwFormatError("IBW channel-label block is not aligned to the v5 label size") + labels = [ + _read_c_string(channel_blob[i : i + _LABEL_SIZE]) + for i in range(0, channel_size, _LABEL_SIZE) + ] + _span(blob, position + channel_size, sum(header.dimension_label_sizes[3:]), "trailing labels") + _, _, channels, _ = header.dimensions + if len(labels) < channels + 1: + raise IbwFormatError("IBW channel-label block is shorter than the declared channel count") + parsed_note: dict[str, str] = {} + for raw in note_bytes.decode("latin-1", "replace").replace("\r", "\n").split("\n"): + if ":" in raw: + key, value = raw.split(":", 1) + parsed_note[key.strip()] = value.strip() + return labels[1 : channels + 1], parsed_note + + +def _canonical_name(label: str) -> str: + name = label or "Unknown" + marker = name.find("Mod") + if marker != -1 and name[marker + 3 :].isdigit(): + name = name[:marker] + for suffix in ("Retrace", "Trace"): + if name.endswith(suffix): + return name[: -len(suffix)] or "Unknown" + return name + + +def _direction(label: str) -> str: + return "backward" if label.endswith("Retrace") else "forward" + + +def _channel_unit(label: str, header_unit: str, note: dict[str, str]) -> tuple[str, float]: + name = _canonical_name(label) + note_unit = note.get(f"{name}Unit", "") + unit = note_unit or header_unit + for prefix, known_unit in _TITLE_UNITS: + if name.startswith(prefix): + unit = known_unit + break + scale = _UNIT_TO_METERS.get(unit, 1.0) + return unit or "", scale + + +def _lateral_range(scale: float, unit: str, resolution: int) -> float: + factor = _UNIT_TO_METERS.get(unit) + if factor is None: + raise IbwFormatError(f"IBW lateral unit {unit!r} is not supported") + return scale * resolution * factor + + +def _inspect_limited_header(source: Path) -> tuple[_Header, list[str]]: + with open(source, "rb") as file: # noqa: PTH123 - header and metadata only + header_blob = file.read(_HEADER_SIZE) + header = _parse_header(header_blob, validate_payload=False) + file.seek(0, 2) + if file.tell() < header.data_end: + raise IbwFormatError("IBW v5 truncated: image payload exceeds file size") + file.seek(header.extra_start) + extra_size = ( + header.formula_size + + header.note_size + + header.data_e_units_size + + sum(header.dimension_e_units_sizes) + + sum(header.dimension_label_sizes) + ) + extras = file.read(extra_size) + labels, _note = _label_sections( + header_blob + b"\0" * (header.extra_start - _HEADER_SIZE) + extras, header + ) + return header, labels + + +def looks_like_limited_igor_ibw(path: str | Path) -> bool: + try: + _inspect_limited_header(Path(path)) + return True + except (IbwFormatError, OSError): + return False + + +def inspect_igor_ibw(path: str | Path) -> DatasetInfo: + source = Path(path) + header, labels = _inspect_limited_header(source) + return DatasetInfo( + path=source, + format="igor-ibw-v5-native-limited", + kinds=("image",), + channels=tuple(_canonical_name(label) for label in labels), + metadata={ + "ibw_version": _IGOR_V5, + "endianness": "little", + "sample_type": "fp32", + "declared_shape": header.dimensions, + }, + ) + + +def load_igor_ibw(path: str | Path) -> SPMData: + source = Path(path) + blob = source.read_bytes() + header = _parse_header(blob) + labels, note = _label_sections(blob, header) + xres, yres, channels, _ = header.dimensions + samples = np.frombuffer(blob, dtype=" Any: return load_bruker_spm +class IgorIbwReader(_ImageReader): + """Limited native Igor Binary Wave v5 image reader. + + A header predicate deliberately lets unsupported ``.ibw`` variants reach the + optional ``afmformats`` reader when that extra is installed. + """ + + extensions: tuple[str, ...] = (".ibw",) + format = "igor-ibw-v5-native-limited" + + def matches_path(self, path: str | Path) -> bool: + from spmkit.core.io.igor_ibw import looks_like_limited_igor_ibw + + return looks_like_limited_igor_ibw(path) + + def inspect(self, path: str | Path) -> Any: + from spmkit.core.io.igor_ibw import inspect_igor_ibw + + return inspect_igor_ibw(path) + + def _loader(self) -> Any: + from spmkit.core.io.igor_ibw import load_igor_ibw + + return load_igor_ibw + + class JpkForceReader: """JPK/Bruker ``.jpk-force`` — curva de fuerza (envuelta en un volumen 1×1).""" @@ -132,5 +158,6 @@ def load(self, path: str | Path, kind: Kind | None = None) -> Any: NhfReader(), GwyReader(), BrukerSpmReader(), + IgorIbwReader(), JpkForceReader(), ) diff --git a/src/spmkit/core/plugins/registry.py b/src/spmkit/core/plugins/registry.py index 083b231..ee67073 100644 --- a/src/spmkit/core/plugins/registry.py +++ b/src/spmkit/core/plugins/registry.py @@ -29,10 +29,16 @@ def readers() -> tuple[Reader, ...]: def reader_for(path: str | Path) -> Reader | None: - """El lector que maneja la extensión de ``path`` (o ``None``).""" + """El lector que maneja ``path`` por extensión y predicado opcional.""" _ensure_discovered() ext = Path(path).suffix.lower() - return next((r for r in _READERS if ext in r.extensions), None) + for reader in _READERS: + if ext not in reader.extensions: + continue + matches_path = getattr(reader, "matches_path", None) + if matches_path is None or matches_path(path): + return reader + return None def supported_extensions() -> tuple[str, ...]: diff --git a/tests/core/test_igor_ibw.py b/tests/core/test_igor_ibw.py new file mode 100644 index 0000000..c82ab36 --- /dev/null +++ b/tests/core/test_igor_ibw.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import struct +import sys +from pathlib import Path + +import numpy as np +import pytest + +from spmkit.core.io import inspect_any, load_any +from spmkit.core.io.igor_ibw import IbwFormatError, load_igor_ibw, looks_like_limited_igor_ibw +from spmkit.core.io.readers import IgorIbwReader +from spmkit.core.plugins import registry + + +def _checksum(header: bytearray) -> None: + struct.pack_into(" bytes: + return b"".join(value.encode("latin-1")[:31].ljust(32, b"\0") for value in values) + + +def _write_ibw( + path: Path, + data: np.ndarray, + *, + labels: list[str] | None = None, + dimensions: tuple[int, int, int, int] | None = None, + sample_type: int = 0x02, + scale: tuple[float, float] = (2e-9, 3e-9), + dimension_unit: str = "m", + data_unit: str = "m", + wfm_size: int | None = None, +) -> None: + channels, yres, xres = data.shape + dimensions = dimensions or (xres, yres, channels, 0) + declared_points = dimensions[0] * dimensions[1] * dimensions[2] + payload = data.astype(" None: + raw = np.array( + [ + [[1e-9, 2e-9, 3e-9], [4e-9, 5e-9, 6e-9]], + [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]], + ], + dtype=np.float32, + ) + path = tmp_path / "image.ibw" + _write_ibw(path, raw) + + loaded = load_igor_ibw(path) + + assert loaded.names == ["Height", "Phase"] + height, phase = loaded.channels + assert height.shape == (2, 3) + assert height.x_range == pytest.approx(6e-9) + assert height.y_range == pytest.approx(6e-9) + assert height.unit == "m" and phase.unit == "deg" + assert height.direction == "forward" and phase.direction == "backward" + assert np.array_equal(height.data, np.flipud(raw[0]).astype(np.float64)) + assert np.array_equal(phase.data, np.flipud(raw[1]).astype(np.float64)) + assert loaded.metadata["format"] == "igor-ibw-v5-native-limited" + + +def test_inspect_and_public_dispatch_use_native_reader(tmp_path: Path) -> None: + path = tmp_path / "native.ibw" + _write_ibw(path, np.zeros((2, 2, 2), dtype=np.float32)) + + assert looks_like_limited_igor_ibw(path) + info = inspect_any(path) + loaded, kind = load_any(path) + + assert info.format == "igor-ibw-v5-native-limited" + assert info.channels == ("Height", "Phase") + assert kind == "image" and loaded.names == ["Height", "Phase"] + + +@pytest.mark.parametrize( + ("shape", "labels"), + [ + ( + (4, 256, 256), + ["", "HeightRetrace", "AmplitudeRetrace", "PhaseRetrace", "ZSensorRetrace"], + ), + ((3, 1024, 1024), ["", "HeightRetrace", "DeflectionRetrace", "ZSensorRetrace"]), + ], +) +def test_loads_each_frozen_structural_shape( + tmp_path: Path, shape: tuple[int, int, int], labels: list[str] +) -> None: + path = tmp_path / "frozen-shape.ibw" + _write_ibw(path, np.zeros(shape, dtype=np.float32), labels=labels) + + loaded = load_igor_ibw(path) + + assert len(loaded.channels) == shape[0] + assert all(channel.shape == shape[1:] for channel in loaded.channels) + assert all(channel.direction == "backward" for channel in loaded.channels) + + +def test_unsupported_native_header_can_reach_a_later_reader( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + class FallbackReader: + extensions = (".ibw",) + + def inspect(self, path: str | Path) -> object: + return path + + def load(self, path: str | Path, kind: object = None) -> object: + return path + + path = tmp_path / "unsupported.ibw" + path.write_bytes(b"\x04\0" + b"\0" * 382) + fallback = FallbackReader() + monkeypatch.setattr(registry, "_READERS", [IgorIbwReader(), fallback]) + monkeypatch.setattr(registry, "_discovered", True) + + assert registry.reader_for(path) is fallback + + +def test_non_ibw_content_is_not_detected(tmp_path: Path) -> None: + path = tmp_path / "not-an-ibw.ibw" + path.write_bytes(b"not an igor binary wave") + + assert not looks_like_limited_igor_ibw(path) + + +@pytest.mark.parametrize( + ("mutator", "message", "rechecksum"), + [ + (lambda value: value.__setitem__(slice(0, 2), b"\0\x05"), "big-endian", True), + (lambda value: value.__setitem__(2, value[2] ^ 1), "checksum", False), + (lambda value: struct.pack_into(" None: # type: ignore[no-untyped-def] + path = tmp_path / "bad.ibw" + _write_ibw(path, np.zeros((1, 2, 2), dtype=np.float32), labels=["", "HeightTrace"]) + raw = bytearray(path.read_bytes()) + mutator(raw) + if rechecksum: + _checksum(raw) + path.write_bytes(raw) + + with pytest.raises(IbwFormatError, match=message): + load_igor_ibw(path) + + +def test_rejects_truncated_header_payload_and_channel_labels(tmp_path: Path) -> None: + header = tmp_path / "header.ibw" + header.write_bytes(b"\x05\0") + with pytest.raises(IbwFormatError, match="truncated"): + load_igor_ibw(header) + + payload = tmp_path / "payload.ibw" + _write_ibw(payload, np.zeros((1, 2, 2), dtype=np.float32), labels=["", "HeightTrace"]) + payload.write_bytes(payload.read_bytes()[:399]) + with pytest.raises(IbwFormatError, match="payload"): + load_igor_ibw(payload) + + labels = tmp_path / "labels.ibw" + _write_ibw(labels, np.zeros((2, 2, 2), dtype=np.float32), labels=["", "HeightTrace"]) + with pytest.raises(IbwFormatError, match="shorter"): + load_igor_ibw(labels) + + +def test_native_reader_does_not_import_gui_or_optional_parsers(tmp_path: Path) -> None: + path = tmp_path / "nogui.ibw" + _write_ibw(path, np.zeros((1, 2, 2), dtype=np.float32), labels=["", "HeightTrace"]) + before = set(sys.modules) + + load_igor_ibw(path) + + imported = set(sys.modules) - before + assert not any(name.startswith(("PyQt", "pyqtgraph", "afmformats")) for name in imported)