diff --git a/docs/PIPELINE.md b/docs/PIPELINE.md index 73f3c12d..48e69cc5 100644 --- a/docs/PIPELINE.md +++ b/docs/PIPELINE.md @@ -138,6 +138,17 @@ $$I_{out} = \text{clip}\big(\text{lift} + \text{gain} \cdot (1 - \text{val}),\ 0 Both are fixed (no per-frame metering) so an evenly-exposed roll renders identically; manual white balance still rides as an additive per-channel shift in log space. For a flat intent the engine also bypasses the creative stages (Local Contrast, Retouch, Lab, Toning, Finish, and dodge/burn masks are skipped too); only Geometry → Normalization → this log map → Crop run. Export is full-resolution; the colour space follows the export selection (color-managed at encode like the print path), as 16-bit TIFF. CPU engine is forced (no GPU flat shader) for numerical exactness. +### Linear Output +**Code**: `negpy.services.export.linear_output` + +When the render intent is **Linear**, the entire darkroom pipeline is bypassed. The source file is decoded to its native linear buffer, lossless geometry (EXIF orientation + user rotation/flip) is applied, and the result is written as an untagged 16-bit TIFF with zlib compression. No normalization, exposure, colour management, flatfield, or sensor correction runs. + +* **Pakon RAW**: the uint16 scanner data is scaled by an expansion factor to use more of the 16-bit output range. F135 (14-bit sensor, confirmed) defaults to 4× (`PAKON_EXPANSION`); F335 (16-bit sensor, detected by file size) defaults to 1× (off). The 2k Square and Panoram specs are assumed 14-bit (same default as F135) but this has not been verified with real samples — override manually if needed. MakeTiff uses 2×; 4× places the typical F135 negative peak around 50–55 % of the range. The user can override via the Expansion combo. The applied expansion factor is recorded in the output TIFF's ImageDescription tag. +* **LinearRaw DNG**: 4-channel VueScan (RGB+IR) and 3-channel SilverFast HDRi files are read directly via tifffile, bypassing rawpy. The IR channel, when present, is written as a separate grayscale TIFF with an `_ir` suffix. Expansion defaults to off; 2× and 4× are available. +* **Camera RAW**: demosaiced by rawpy with `user_wb=[1,1,1,1]` (unity), `output_color=raw` (sensor-native), `gamma=(1,1)` (linear), `no_auto_bright=True`. The camera's as-shot white balance multipliers (green-normalized to three RGB values) are embedded in XMP as `RAW-WB: R G B` inside `dc:description`, matching the MakeTiff/ColorPerfect convention. The source filename is stored in `crs:RawFileName`. No expansion option (camera sensors use the full bit depth). + +Source device metadata (Make, Model, DateTime) is carried through to the output TIFF when available from the source file. + --- ## 4. Local Contrast (CLAHE) diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index c2970046..cce649ed 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -455,6 +455,11 @@ A scrollable list of every edit step (last 100 kept), newest on top; the current * **Flat**: a flat, neutral, low-contrast master that keeps maximum tonal/colour information for editing elsewhere (Lightroom, Darktable, Photoshop). Skips the print look, effects, toning, and vignette, and writes a wide-gamut 16-bit TIFF. Your in-app preview is unaffected. * **Preview Flat**: temporarily show the flat master on the canvas without changing your edit. * **Roll Baseline**: measure every visible frame and share one exposure baseline, so flat masters are consistent across a roll (recommended before a flat batch). +* **Linear**: bypass the entire darkroom pipeline and dump the scanner's or camera's decoded buffer as an untagged linear 16-bit TIFF. No normalization, exposure, colour management, flatfield, or sensor correction — just the raw data with lossless geometry (rotation/flip) applied. Supported sources: + * **Pakon RAW** — 4× expansion by default (14-bit sensor range scaled into 16-bit). F335 files (16-bit sensor) default to no expansion. + * **LinearRaw DNG** — SilverFast HDRi (3-channel) and VueScan (4-channel RGB+IR). IR is written as a separate grayscale TIFF with an `_ir` suffix. + * **Camera RAW** — demosaiced with unity white balance (1,1,1,1). The camera's as-shot WB is written into XMP (`RAW-WB: R G B`, MakeTiff-compatible) so it can be applied downstream. Source device and timestamp are preserved. + * **Expansion**: scales the linear data before writing. The combo box shows source-appropriate options: Pakon F135/F235 default to 4×, F335 and LinearRaw DNG default to off. Camera RAW files have no expansion option. Leave at the default unless you know why you need to change it. ### Export button diff --git a/negpy/desktop/controller.py b/negpy/desktop/controller.py index df19503f..004a0028 100644 --- a/negpy/desktop/controller.py +++ b/negpy/desktop/controller.py @@ -206,6 +206,7 @@ class AppController(QObject): monitor_profile_changed = pyqtSignal() compare_changed = pyqtSignal(bool) flat_output_changed = pyqtSignal(bool) + linear_output_changed = pyqtSignal(bool) flat_peek_changed = pyqtSignal(bool) zoom_requested = pyqtSignal(float) zoom_changed = pyqtSignal(float) @@ -3004,6 +3005,8 @@ def set_flat_output(self, enabled: bool) -> None: if self.state.flat_output == enabled: return self.state.flat_output = enabled + if enabled: + self.state.linear_output = False self.session.save_flat_output_prefs() # Flat masters default to full resolution; only honour Print/Pixels when the # user explicitly selects those modes in the export panel. @@ -3019,10 +3022,25 @@ def set_flat_output(self, enabled: bool) -> None: persist=True, ) self.flat_output_changed.emit(enabled) + if enabled: + self.linear_output_changed.emit(False) # If a peek is active and flat output was turned off, drop back to the edit. if not enabled and self.state.flat_peek: self.toggle_flat_peek(force=False) + def set_linear_output(self, enabled: bool) -> None: + """Toggle the linear output intent (raw loader dump, no pipeline).""" + if self.state.linear_output == enabled: + return + self.state.linear_output = enabled + if enabled: + self.state.flat_output = False + self.flat_output_changed.emit(False) + if self.state.flat_peek: + self.toggle_flat_peek(force=False) + self.session.save_flat_output_prefs() + self.linear_output_changed.emit(enabled) + def toggle_flat_peek(self, force: Optional[bool] = None) -> None: """Preview the flat master render in the canvas without changing the saved edit. @@ -3149,6 +3167,51 @@ def export_history_step(self, index: int) -> None: self.session.jump_to_step(index) self.request_export() + def request_linear_output_export(self, files: list[dict] | None = None) -> None: + """Export decoded linear buffers as untagged 16-bit TIFFs to the export folder.""" + from negpy.services.export.linear_output import export_linear_output, is_linear_output_supported + + export_path = self._ensure_valid_export_path() + if not export_path: + return + + if files is None: + file_path = self.state.current_file_path + if not file_path: + return + if not is_linear_output_supported(file_path): + self.set_status("Linear Output is not supported for this file type", 4000) + return + files = [{"path": file_path, "name": os.path.basename(file_path)}] + + supported = [f for f in files if is_linear_output_supported(f["path"])] + if not supported: + self.set_status("No files support Linear Output", 4000) + return + + if len(supported) > 1 and not self._confirm_bulk_export(f"Linear-export {len(supported)} frames?"): + return + + exported = 0 + geometry = self.state.config.geometry + expansion = self.state.linear_expansion + for f in supported: + stem = os.path.splitext(os.path.basename(f["path"]))[0] + out_path = os.path.join(export_path, f"{stem}_linear.tiff") + counter = 2 + while os.path.exists(out_path): + out_path = os.path.join(export_path, f"{stem}_linear_{counter}.tiff") + counter += 1 + try: + export_linear_output(f["path"], out_path, geometry=geometry, expansion=expansion) + exported += 1 + except Exception as e: + logger.warning("Linear output failed for %s: %s", f.get("name"), e) + self.set_status(f"Linear Output failed: {os.path.basename(f['path'])}: {e}", 4000) + + if exported: + self.set_status(f"Linear Output: exported {exported} file(s)", 4000) + def request_export(self) -> None: """Exports the current file using the settings currently shown in the Export panel.""" if self._batch_busy("export"): diff --git a/negpy/desktop/session.py b/negpy/desktop/session.py index f7ef1fd8..260b0c9b 100644 --- a/negpy/desktop/session.py +++ b/negpy/desktop/session.py @@ -162,6 +162,11 @@ class AppState: # Transient: preview is currently peeking the flat render (not persisted). flat_peek: bool = False + # Linear Output: export the loader's raw decoded buffer as an untagged 16-bit TIFF. + linear_output: bool = False + # Linear Output expansion factor override. None = source-type default (4× Pakon, off DNG). + linear_expansion: float | None = None + @property def local_hidden_masks(self) -> set: """The current file's hidden-mask indices (empty = all shown). Returns a fresh, @@ -463,6 +468,10 @@ def __init__(self, repo: StorageRepository): if saved_flat_output is not None: self.state.flat_output = bool(saved_flat_output) + saved_linear_output = self.repo.get_global_setting("linear_output") + if saved_linear_output is not None: + self.state.linear_output = bool(saved_linear_output) + self.state.export_presets = self.repo.load_export_presets() def set_gpu_enabled(self, enabled: bool) -> None: @@ -516,8 +525,9 @@ def save_export_presets(self) -> None: self.repo.save_export_presets(self.state.export_presets) def save_flat_output_prefs(self) -> None: - """Persists the flat ('for editing elsewhere') output preferences.""" + """Persists the flat / linear output preferences.""" self.repo.save_global_setting("flat_output", self.state.flat_output) + self.repo.save_global_setting("linear_output", self.state.linear_output) def _apply_sticky_settings(self, config: WorkspaceConfig, only_global: bool = False) -> WorkspaceConfig: """ diff --git a/negpy/desktop/view/keyboard_shortcuts.py b/negpy/desktop/view/keyboard_shortcuts.py index 0f8d5cb3..83f7213c 100644 --- a/negpy/desktop/view/keyboard_shortcuts.py +++ b/negpy/desktop/view/keyboard_shortcuts.py @@ -130,6 +130,7 @@ def _build_actions(self) -> dict[str, Callable[[], None]]: "zoom_100": self.window.canvas.zoom_to_original, "zoom_200": lambda: self.window.canvas.zoom_to_percent(200.0), "export": controller.request_export, + "export_linear_output": controller.request_linear_output_export, "copy": controller.session.copy_settings, "copy_with_bounds": controller.session.copy_settings_with_bounds, "paste": lambda: open_paste_dialog(self.window, controller), diff --git a/negpy/desktop/view/shortcut_registry.py b/negpy/desktop/view/shortcut_registry.py index f54b89a5..c40a401c 100644 --- a/negpy/desktop/view/shortcut_registry.py +++ b/negpy/desktop/view/shortcut_registry.py @@ -147,6 +147,7 @@ class ShortcutEntry: "zoom_100": ShortcutEntry("1", "Zoom 100%", "View"), "zoom_200": ShortcutEntry("2", "Zoom 200%", "View"), "export": ShortcutEntry("Ctrl+E", "Export", "Actions"), + "export_linear_output": ShortcutEntry("", "Export Linear Output", "Actions"), "copy": ShortcutEntry("Ctrl+C", "Copy settings", "Actions"), "copy_with_bounds": ShortcutEntry("Ctrl+Shift+C", "Copy settings (with bounds)", "Actions"), "paste": ShortcutEntry("Ctrl+V", "Paste settings", "Actions"), diff --git a/negpy/desktop/view/sidebar/export.py b/negpy/desktop/view/sidebar/export.py index 30450c61..7055606c 100644 --- a/negpy/desktop/view/sidebar/export.py +++ b/negpy/desktop/view/sidebar/export.py @@ -92,6 +92,7 @@ def _connect_signals(self) -> None: self.flat_peek_btn.toggled.connect(lambda checked: self.controller.toggle_flat_peek(force=checked)) self.flat_bake_btn.clicked.connect(self.controller.request_batch_normalization) self.controller.flat_output_changed.connect(self._on_flat_output_changed) + self.controller.linear_output_changed.connect(self._on_linear_output_changed) self.controller.flat_peek_changed.connect(self._on_flat_peek_changed) self.contact_sheet_btn.clicked.connect(self.controller.request_contact_sheet) @@ -145,10 +146,10 @@ def _add_presets_section(self) -> None: repo = self.controller.session.repo expanded = bool(repo.get_global_setting("section_expanded_export_presets", default=False)) - section = CollapsibleSection("Presets", expanded=expanded, icon=qta.icon("fa5s.layer-group", color="#aaa")) - section.set_content(content) - section.expanded_changed.connect(lambda checked: repo.save_global_setting("section_expanded_export_presets", checked)) - self.layout.addWidget(section) + self._presets_section = CollapsibleSection("Presets", expanded=expanded, icon=qta.icon("fa5s.layer-group", color="#aaa")) + self._presets_section.set_content(content) + self._presets_section.expanded_changed.connect(lambda checked: repo.save_global_setting("section_expanded_export_presets", checked)) + self.layout.addWidget(self._presets_section) # --- Contact sheet ------------------------------------------------------- @@ -492,7 +493,13 @@ def _add_flat_master_section(self) -> None: "print look (auto density/grade, cast removal, lab effects, toning, vignette) and " "writes a wide-gamut, high-bit-depth file. Your in-app preview is unaffected." ) - for btn in (self.intent_print_btn, self.intent_flat_btn): + self.intent_linear_btn = QPushButton("Linear") + self.intent_linear_btn.setToolTip( + "Export the raw decoded sensor data as an untagged 16-bit TIFF, before any " + "NegPy processing (no normalization, exposure, lab, toning, color management). " + "Supported for Pakon RAW and LinearRaw DNG (SilverFast/VueScan) files." + ) + for btn in (self.intent_print_btn, self.intent_flat_btn, self.intent_linear_btn): btn.setCheckable(True) btn.setStyleSheet(labeled_toggle_qss()) intent_row.addWidget(btn) @@ -500,7 +507,10 @@ def _add_flat_master_section(self) -> None: self.intent_btn_group.setExclusive(True) self.intent_btn_group.addButton(self.intent_print_btn, 0) self.intent_btn_group.addButton(self.intent_flat_btn, 1) - if self.state.flat_output: + self.intent_btn_group.addButton(self.intent_linear_btn, 2) + if self.state.linear_output: + self.intent_linear_btn.setChecked(True) + elif self.state.flat_output: self.intent_flat_btn.setChecked(True) else: self.intent_print_btn.setChecked(True) @@ -534,14 +544,49 @@ def _add_flat_master_section(self) -> None: self.flat_roll_warning = hint_label("For consistent masters across a roll, lock one baseline for every frame.", kind="warning") box.addWidget(self.flat_roll_warning) + self.linear_hint_label = hint_label( + "Exports the loader's decoded buffer as an untagged 16-bit TIFF. " + "No pipeline processing, no color management, no scaling. " + "Pakon RAW, LinearRaw DNG (SilverFast/VueScan), and camera RAW." + ) + box.addWidget(self.linear_hint_label) + + expansion_row = QHBoxLayout() + expansion_row.setContentsMargins(0, 0, 0, 0) + self.linear_expansion_label = field_label("Expansion") + expansion_row.addWidget(self.linear_expansion_label) + self.linear_expansion_combo = QComboBox() + self.linear_expansion_combo.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + expansion_row.addWidget(self.linear_expansion_combo) + self.linear_expansion_row = QWidget() + self.linear_expansion_row.setLayout(expansion_row) + self.linear_expansion_row.setVisible(False) + box.addWidget(self.linear_expansion_row) + self.linear_expansion_hint = hint_label("Leave at the default unless you know why you need to change it.") + self.linear_expansion_hint.setVisible(False) + box.addWidget(self.linear_expansion_hint) + self.linear_expansion_combo.currentIndexChanged.connect(self._on_linear_expansion_changed) + self.layout.addWidget(container) def _sync_flat_enabled(self) -> None: - on = self.intent_flat_btn.isChecked() + flat_on = self.intent_flat_btn.isChecked() + linear_on = self.intent_linear_btn.isChecked() if hasattr(self, "form"): - self.form.set_flat_mode(on) - self.flat_hint_label.setVisible(on) - self.flat_peek_btn.setVisible(on) + self.form.set_flat_mode(flat_on) + self.form.setVisible(not linear_on) + self.flat_hint_label.setVisible(flat_on) + self.flat_peek_btn.setVisible(flat_on) + self.linear_hint_label.setVisible(linear_on) + if hasattr(self, "linear_expansion_row"): + self.linear_expansion_row.setVisible(linear_on) + self.linear_expansion_hint.setVisible(linear_on) + if linear_on: + self._refresh_linear_expansion_combo() + if hasattr(self, "_presets_section"): + self._presets_section.setVisible(not linear_on) + if hasattr(self, "_sidecars_section"): + self._sidecars_section.setVisible(not linear_on) self._sync_flat_roll_warning() if hasattr(self, "form"): self._refresh_export_enabled() @@ -559,18 +604,72 @@ def _sync_flat_roll_warning(self) -> None: def _on_flat_output_toggled(self, btn_id: int, checked: bool) -> None: if checked: - self.controller.set_flat_output(btn_id == 1) + if btn_id == 2: + self.controller.set_linear_output(True) + else: + self.controller.set_linear_output(False) + self.controller.set_flat_output(btn_id == 1) self._sync_flat_enabled() def _on_flat_output_changed(self, enabled: bool) -> None: self.intent_btn_group.blockSignals(True) if enabled: self.intent_flat_btn.setChecked(True) - else: + elif not self.state.linear_output: + self.intent_print_btn.setChecked(True) + self.intent_btn_group.blockSignals(False) + self._sync_flat_enabled() + + def _on_linear_output_changed(self, enabled: bool) -> None: + self.intent_btn_group.blockSignals(True) + if enabled: + self.intent_linear_btn.setChecked(True) + elif not self.state.flat_output: self.intent_print_btn.setChecked(True) self.intent_btn_group.blockSignals(False) self._sync_flat_enabled() + _EXPANSION_OPTIONS: dict[str, list[tuple[str, float | None]]] = { + "pakon": [("4× (default)", None), ("2×", 2.0), ("Off", 1.0)], + "pakon_f335": [("Off (default)", None), ("2×", 2.0), ("4×", 4.0)], + "dng": [("Off (default)", None), ("2×", 2.0), ("4×", 4.0)], + "camera": [], + "unsupported": [], + } + + def _refresh_linear_expansion_combo(self) -> None: + from negpy.services.export.linear_output import linear_output_source_type + + path = self.state.current_file_path or "" + source_type = linear_output_source_type(path) if path else "unsupported" + options = self._EXPANSION_OPTIONS.get(source_type, []) + + combo = self.linear_expansion_combo + combo.blockSignals(True) + combo.clear() + if not options: + combo.addItem("N/A") + combo.setEnabled(False) + else: + for label, _val in options: + combo.addItem(label) + combo.setEnabled(True) + current = self.state.linear_expansion + for i, (_label, val) in enumerate(options): + if val == current: + combo.setCurrentIndex(i) + break + else: + combo.setCurrentIndex(0) + combo.blockSignals(False) + self._current_expansion_source_type = source_type + + def _on_linear_expansion_changed(self, index: int) -> None: + source_type = getattr(self, "_current_expansion_source_type", "unsupported") + options = self._EXPANSION_OPTIONS.get(source_type, []) + if 0 <= index < len(options): + self.state.linear_expansion = options[index][1] + def _on_flat_peek_changed(self, active: bool) -> None: self.flat_peek_btn.blockSignals(True) self.flat_peek_btn.setChecked(active) @@ -670,11 +769,13 @@ def _add_sidecars_section(self) -> None: repo = self.controller.session.repo expanded = bool(repo.get_global_setting("section_expanded_export_sidecars", default=False)) - section = CollapsibleSection("Sidecars", expanded=expanded, icon=qta.icon("fa5s.file-export", color="#aaa")) - section.setToolTip("Optional plain-file copies of edits next to your sources, for archival. SQLite stays primary.") - section.set_content(content) - section.expanded_changed.connect(lambda checked: repo.save_global_setting("section_expanded_export_sidecars", checked)) - self.layout.addWidget(section) + self._sidecars_section = CollapsibleSection("Sidecars", expanded=expanded, icon=qta.icon("fa5s.file-export", color="#aaa")) + self._sidecars_section.setToolTip("Optional plain-file copies of edits next to your sources, for archival. SQLite stays primary.") + self._sidecars_section.set_content(content) + self._sidecars_section.expanded_changed.connect( + lambda checked: repo.save_global_setting("section_expanded_export_sidecars", checked) + ) + self.layout.addWidget(self._sidecars_section) # --- Batch --------------------------------------------------------------- @@ -758,6 +859,25 @@ def _set_export_scope(self, key: str, persist: bool = True) -> None: self.controller.session.repo.save_global_setting("export_scope", key) def _on_export_clicked(self) -> None: + if self.state.linear_output: + scope = self._export_scope + if scope in ("all_current", "all_saved"): + files = [ + self.state.uploaded_files[i] + for i in self.controller.session.asset_model.visible_actual_indices_ordered() + if not self.state.uploaded_files[i].get("excluded") + ] + self.controller.request_linear_output_export(files=files) + elif scope == "selected": + files = [ + self.state.uploaded_files[i] + for i in self.state.selected_indices + if 0 <= i < len(self.state.uploaded_files) and not self.state.uploaded_files[i].get("excluded") + ] + self.controller.request_linear_output_export(files=files) + else: + self.controller.request_linear_output_export() + return scope = self._export_scope if scope == "selected": self.controller.request_export_selected() @@ -927,9 +1047,20 @@ def _refresh_proof_mismatch_warning(self) -> None: def _refresh_export_enabled(self) -> None: """Disable the Export action when the current format/colour-space pairing can't be encoded (JPEG XL only tags a subset of colour spaces).""" - blocked = self.form.is_export_blocked() - self.export_main_btn.setEnabled(not blocked) - self.export_menu_btn.setEnabled(not blocked) + linear_on = self.state.linear_output + if linear_on: + from negpy.services.export.linear_output import is_linear_output_supported + + path = self.state.current_file_path or "" + supported = bool(path) and is_linear_output_supported(path) + self.export_main_btn.setEnabled(supported) + self.export_menu_btn.setEnabled(True) + if hasattr(self, "linear_expansion_row"): + self._refresh_linear_expansion_combo() + else: + blocked = self.form.is_export_blocked() + self.export_main_btn.setEnabled(not blocked) + self.export_menu_btn.setEnabled(not blocked) def sync_ui(self) -> None: conf = self.state.config.export @@ -957,7 +1088,9 @@ def sync_ui(self) -> None: self.cs_template_combo.setCurrentText(saved_template) else: self.cs_template_combo.setCurrentText(ContactSheetTemplates.DEFAULT_NAME) - if self.state.flat_output: + if self.state.linear_output: + self.intent_linear_btn.setChecked(True) + elif self.state.flat_output: self.intent_flat_btn.setChecked(True) else: self.intent_print_btn.setChecked(True) diff --git a/negpy/services/export/linear_output.py b/negpy/services/export/linear_output.py new file mode 100644 index 00000000..a4150ef3 --- /dev/null +++ b/negpy/services/export/linear_output.py @@ -0,0 +1,446 @@ +"""Linear Output: export a loader's decoded buffer as an untagged 16-bit TIFF. + +Bypasses the entire darkroom pipeline — no normalization, exposure, lab, +toning, finish, flatfield, or sensor-crosstalk correction. The output is +the closest thing to "what the scanner/camera actually captured" that NegPy +can produce, with only lossless geometry (EXIF orientation + user rotation/flip) +baked in. +""" + +import io +import os +from dataclasses import dataclass +from typing import Optional + +import numpy as np +import rawpy +import tifffile as _tifffile + +from negpy.features.geometry.models import GeometryConfig +from negpy.infrastructure.loaders.constants import SUPPORTED_JPEG_EXTENSIONS, SUPPORTED_RAW_EXTENSIONS, SUPPORTED_TIFF_EXTENSIONS +from negpy.infrastructure.loaders.helpers import NonStandardFileWrapper, get_best_demosaic_algorithm, read_orientation +from negpy.infrastructure.loaders.pakon_loader import PakonLoader +from negpy.infrastructure.loaders.rawpy_loader import ( + _find_linearraw_page, + _is_dng, + _peek_hdri_ir_page, + _peek_linearraw_4ch, +) +from negpy.kernel.image.logic import _to_uint16_jit, apply_exif_orientation, ensure_rgb, uint16_to_float32 + + +@dataclass(frozen=True) +class _CameraWB: + """Camera white balance multipliers extracted before the unity-WB decode.""" + + as_shot: tuple[float, float, float, float] + daylight: tuple[float, float, float, float] + + +@dataclass(frozen=True) +class _SourceMeta: + """Device and timestamp metadata from the source file.""" + + make: Optional[str] = None + model: Optional[str] = None + datetime: Optional[str] = None + + +def _read_source_meta_tiff(file_path: str) -> _SourceMeta: + try: + with _tifffile.TiffFile(file_path) as tif: + tags = tif.pages[0].tags + make = tags.get("Make") + model = tags.get("Model") + dt = tags.get("DateTime") + return _SourceMeta( + make=str(make.value).strip() if make else None, + model=str(model.value).strip() if model else None, + datetime=str(dt.value).strip() if dt else None, + ) + except Exception: + return _read_source_meta_exif(file_path) + + +def _read_source_meta_exif(file_path: str) -> _SourceMeta: + """Fallback: scan for an embedded EXIF TIFF block (works for RAF, ORF, etc.).""" + try: + with open(file_path, "rb") as f: + header = f.read(4096) + marker = header.find(b"Exif\x00\x00") + if marker < 0: + return _SourceMeta() + tiff_bytes = header[marker + 6 :] + import logging + + prev = logging.getLogger("tifffile").level + logging.getLogger("tifffile").setLevel(logging.CRITICAL) + try: + tif = _tifffile.TiffFile(io.BytesIO(tiff_bytes)) + finally: + logging.getLogger("tifffile").setLevel(prev) + tags = tif.pages[0].tags + make = tags.get("Make") + model = tags.get("Model") + dt = tags.get("DateTime") + return _SourceMeta( + make=str(make.value).strip() if make else None, + model=str(model.value).strip() if model else None, + datetime=str(dt.value).strip() if dt else None, + ) + except Exception: + return _SourceMeta() + + +def _is_camera_raw(file_path: str) -> bool: + ext = os.path.splitext(file_path)[1].lower() + if ext in SUPPORTED_TIFF_EXTENSIONS | SUPPORTED_JPEG_EXTENSIONS: + return False + if PakonLoader.can_handle(file_path): + return False + return ext in SUPPORTED_RAW_EXTENSIONS + + +def is_linear_output_supported(file_path: str) -> bool: + if PakonLoader.can_handle(file_path): + return True + if _is_dng(file_path): + return _is_linearraw_dng(file_path) or _is_camera_raw(file_path) + if _is_camera_raw(file_path): + return True + return False + + +def linear_output_source_type(file_path: str) -> str: + """Classify a file for Linear Output expansion options. + + Returns ``"pakon"``, ``"dng"``, ``"camera"``, or ``"unsupported"``. + """ + if PakonLoader.can_handle(file_path): + return "pakon_f335" if _is_pakon_f335(file_path) else "pakon" + if _is_dng(file_path) and _is_linearraw_dng(file_path): + return "dng" + if _is_camera_raw(file_path): + return "camera" + return "unsupported" + + +def _is_linearraw_dng(file_path: str) -> bool: + """True if the DNG contains a LinearRaw IFD (3 or 4 samples).""" + try: + with _tifffile.TiffFile(file_path) as tif: + return _find_linearraw_page(tif, samples=4) is not None or _find_linearraw_page(tif, samples=3) is not None + except Exception: + return False + + +def _apply_geometry(f32: np.ndarray, orientation: int, geometry: Optional[GeometryConfig]) -> np.ndarray: + f32 = apply_exif_orientation(f32, orientation) + if geometry is not None: + if geometry.rotation != 0: + f32 = np.rot90(f32, k=geometry.rotation) + if geometry.flip_horizontal: + f32 = np.ascontiguousarray(np.fliplr(f32)) + if geometry.flip_vertical: + f32 = np.ascontiguousarray(np.flipud(f32)) + return f32 + + +def _decode_linear( + file_path: str, geometry: Optional[GeometryConfig] = None, expansion: Optional[float] = None +) -> tuple[np.ndarray, Optional[np.ndarray], Optional[_CameraWB], _SourceMeta]: + """Decode to an oriented float32 buffer. Returns (rgb, ir_or_none, camera_wb_or_none, source_meta).""" + if PakonLoader.can_handle(file_path): + rgb, ir = _decode_pakon(file_path, geometry, expansion=expansion) + meta = _SourceMeta(make="Pakon", model=_pakon_spec_desc(file_path)) + return rgb, ir, None, meta + if _is_dng(file_path): + meta = _read_source_meta_tiff(file_path) + if _is_linearraw_dng(file_path): + rgb, ir = _decode_dng(file_path, geometry, expansion=expansion) + return rgb, ir, None, meta + if _is_camera_raw(file_path): + rgb, ir, wb = _decode_camera_raw(file_path, geometry) + return rgb, ir, wb, meta + if _is_camera_raw(file_path): + meta = _read_source_meta_tiff(file_path) + rgb, ir, wb, decode_meta = _decode_camera_raw(file_path, geometry) + merged = _SourceMeta( + make=meta.make or decode_meta.make, + model=meta.model or decode_meta.model, + datetime=meta.datetime or decode_meta.datetime, + ) + return rgb, ir, wb, merged + raise ValueError(f"Linear Output is not supported for this file type: {file_path}") + + +PAKON_EXPANSION = 4.0 +_F335_SIZE = 72000000 + + +def _is_pakon_f335(file_path: str) -> bool: + try: + return abs(os.path.getsize(file_path) - _F335_SIZE) < 1024 + except OSError: + return False + + +def _default_pakon_expansion(file_path: str) -> float: + # F335 is 16-bit; all others assumed 14-bit (confirmed for F135, unverified for 2k Square / Panoram). + return 1.0 if _is_pakon_f335(file_path) else PAKON_EXPANSION + + +def _pakon_spec_desc(file_path: str) -> str: + try: + file_size = os.path.getsize(file_path) + spec = next((s for s in PakonLoader.PAKON_SPECS if abs(file_size - s["size"]) < 1024), None) + return spec["desc"] if spec else "Unknown" + except OSError: + return "Unknown" + + +def _decode_pakon(file_path: str, geometry: Optional[GeometryConfig] = None, expansion: Optional[float] = None) -> tuple[np.ndarray, None]: + loader = PakonLoader() + ctx_mgr, metadata = loader.load(file_path) + with ctx_mgr as wrapper: + if not isinstance(wrapper, NonStandardFileWrapper): + raise TypeError("Expected NonStandardFileWrapper from PakonLoader") + f32 = wrapper.data + factor = expansion if expansion is not None else _default_pakon_expansion(file_path) + if factor > 1.0: + f32 = np.clip(f32 * factor, 0.0, 1.0) + f32 = _apply_geometry(f32, metadata.get("orientation", 0), geometry) + return f32, None + + +def _decode_dng( + file_path: str, geometry: Optional[GeometryConfig] = None, expansion: Optional[float] = None +) -> tuple[np.ndarray, Optional[np.ndarray]]: + peeked_4ch = _peek_linearraw_4ch(file_path) + if peeked_4ch is not None: + rgb, ir = peeked_4ch + if expansion is not None and expansion > 1.0: + rgb = np.clip(rgb * expansion, 0.0, 1.0) + orientation = read_orientation(file_path) + rgb = _apply_geometry(rgb, orientation, geometry) + ir = _apply_geometry(ir, orientation, geometry) + return rgb, ir + + # 3-channel LinearRaw (SilverFast HDRi): read directly via tifffile. + try: + with _tifffile.TiffFile(file_path) as tif: + page = _find_linearraw_page(tif, samples=3) + if page is None: + raise ValueError(f"No LinearRaw IFD found in {file_path}") + arr = page.asarray() + except ValueError: + raise + except Exception as e: + raise ValueError(f"Failed to read LinearRaw data from {file_path}: {e}") from e + + if arr.dtype == np.uint16: + scale = 1.0 / 65535.0 + elif arr.dtype == np.uint8: + scale = 1.0 / 255.0 + else: + scale = 1.0 + rgb = np.clip(arr.astype(np.float32) * scale, 0.0, 1.0) + if expansion is not None and expansion > 1.0: + rgb = np.clip(rgb * expansion, 0.0, 1.0) + + ir = _peek_hdri_ir_page(file_path) + orientation = read_orientation(file_path) + rgb = _apply_geometry(rgb, orientation, geometry) + if ir is not None: + ir = _apply_geometry(ir, orientation, geometry) + return rgb, ir + + +def _decode_camera_raw(file_path: str, geometry: Optional[GeometryConfig] = None) -> tuple[np.ndarray, None, _CameraWB, _SourceMeta]: + raw = rawpy.imread(file_path) + wb = _CameraWB( + as_shot=tuple(raw.camera_whitebalance), # type: ignore[arg-type] + daylight=tuple(raw.daylight_whitebalance), # type: ignore[arg-type] + ) + ts = raw.other.timestamp + dt_str = ts.strftime("%Y:%m:%d %H:%M:%S") if ts else None + algo = get_best_demosaic_algorithm(raw) + rgb = raw.postprocess( + gamma=(1, 1), + no_auto_bright=True, + user_wb=[1, 1, 1, 1], + output_bps=16, + output_color=rawpy.ColorSpace.raw, + demosaic_algorithm=algo, + user_flip=0, + adjust_maximum_thr=0.0, + ) + raw.close() + rgb = ensure_rgb(rgb) + f32 = uint16_to_float32(rgb) + orientation = read_orientation(file_path) + f32 = _apply_geometry(f32, orientation, geometry) + meta = _SourceMeta(datetime=dt_str) + return f32, None, wb, meta + + +def _normalize_wb_rgb(wb: tuple[float, float, float, float]) -> tuple[float, float, float]: + """Normalize RGGB multipliers to green=1, return (R, G, B).""" + g = wb[1] if wb[1] > 0 else 1.0 + return (wb[0] / g, 1.0, wb[2] / g) + + +def _build_xmp(source_path: str, wb: _CameraWB) -> bytes: + r, g, b = _normalize_wb_rgb(wb.as_shot) + raw_name = os.path.basename(source_path) + xmp = ( + "\n" + "\n" + "\n" + " \n" + f" {raw_name}\n" + " \n" + " \n" + " \n" + " \n" + f" RAW-WB: {r:.6f} {g:.6f} {b:.6f}\n" + " \n" + " \n" + " \n" + "\n" + "\n" + "" + ) + return xmp.encode("utf-8") + + +def _effective_expansion(file_path: str, expansion: Optional[float]) -> float: + if PakonLoader.can_handle(file_path): + factor = expansion if expansion is not None else _default_pakon_expansion(file_path) + return factor if factor > 1.0 else 1.0 + if _is_dng(file_path) and _is_linearraw_dng(file_path): + return expansion if (expansion is not None and expansion > 1.0) else 1.0 + return 1.0 + + +def _source_format_label(file_path: str) -> str: + if PakonLoader.can_handle(file_path): + return f"Pakon {_pakon_spec_desc(file_path)}" + if _is_dng(file_path) and _is_linearraw_dng(file_path): + return "DNG LinearRaw" + if _is_camera_raw(file_path): + return "camera RAW" + return "unknown" + + +def _write_tiff( + f32: np.ndarray, + dest, + source_name: str, + camera_wb: Optional[_CameraWB] = None, + source_path: Optional[str] = None, + source_meta: Optional[_SourceMeta] = None, + expansion: float = 1.0, + source_format: str = "", +) -> None: + """Write a float32 buffer as an untagged 16-bit TIFF to *dest* (path or file-like).""" + u16 = _to_uint16_jit(np.ascontiguousarray(f32, dtype=np.float32)) + photometric = "rgb" if f32.ndim == 3 else "minisblack" + parts = [f"source: {source_format or source_name}"] + if expansion > 1.0: + parts.append(f"expansion: x{expansion:g}") + else: + parts.append("no scaling") + if camera_wb is not None: + r, g, b = _normalize_wb_rgb(camera_wb.as_shot) + parts.append(f"no WB applied (as-shot: {r:.3f} {g:.3f} {b:.3f})") + else: + parts.append("no WB applied") + parts.append("no color management") + description = f"NegPy Linear Output -- {', '.join(parts)}." + + extratags: list[tuple] = [] + if camera_wb is not None and source_path is not None: + xmp_bytes = _build_xmp(source_path, camera_wb) + extratags.append((700, 1, len(xmp_bytes), xmp_bytes, True)) + + if source_meta is not None: + if source_meta.make: + extratags.append((271, 2, 0, source_meta.make, True)) + if source_meta.model: + extratags.append((272, 2, 0, source_meta.model, True)) + + dt = (source_meta.datetime if source_meta else None) or None + + _tifffile.imwrite( + dest, + u16, + photometric=photometric, + compression="zlib", + predictor=True, + description=description, + software="NegPy", + datetime=dt, + extratags=extratags or None, + metadata=None, + ) + + +def _write_ir_tiff(ir: np.ndarray, dest, source_name: str) -> None: + """Write a single-channel IR buffer as an untagged 16-bit grayscale TIFF.""" + u16 = _to_uint16_jit(np.ascontiguousarray(ir[:, :, np.newaxis] if ir.ndim == 2 else ir, dtype=np.float32)) + if u16.ndim == 3 and u16.shape[2] == 1: + u16 = u16[:, :, 0] + description = f"NegPy Linear Output -- infrared channel. Source: {source_name}" + _tifffile.imwrite( + dest, + u16, + photometric="minisblack", + compression="zlib", + predictor=True, + description=description, + ) + + +def export_linear_output( + file_path: str, output_path: str, geometry: Optional[GeometryConfig] = None, expansion: Optional[float] = None +) -> None: + """Decode *file_path* and write an untagged linear 16-bit TIFF to *output_path*. + + Lossless geometry (90-degree rotation, horizontal/vertical flip) from *geometry* + is applied; fine rotation is ignored (it resamples). + + *expansion* scales the linear data before writing (e.g. 4.0 for Pakon's 14-bit + sensor → 16-bit range). ``None`` uses the source-type default; values <= 1.0 disable. + + If the source has an IR channel, it is written as a separate grayscale TIFF + with an ``_ir`` suffix next to the RGB output. + """ + eff = _effective_expansion(file_path, expansion) + fmt = _source_format_label(file_path) + f32, ir, camera_wb, meta = _decode_linear(file_path, geometry, expansion=expansion) + os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) + _write_tiff( + f32, output_path, os.path.basename(file_path), camera_wb, source_path=file_path, source_meta=meta, expansion=eff, source_format=fmt + ) + + if ir is not None: + stem, ext = os.path.splitext(output_path) + ir_path = f"{stem}_ir{ext}" + _write_ir_tiff(ir, ir_path, os.path.basename(file_path)) + + +def export_linear_output_bytes(file_path: str, geometry: Optional[GeometryConfig] = None) -> tuple[bytes, str]: + """Like export_linear_output but returns (tiff_bytes, filename_stem) for in-memory use. + + IR is not included in the returned bytes (use export_linear_output for IR). + """ + eff = _effective_expansion(file_path, None) + fmt = _source_format_label(file_path) + f32, _ir, camera_wb, meta = _decode_linear(file_path, geometry) + buf = io.BytesIO() + _write_tiff(f32, buf, os.path.basename(file_path), camera_wb, source_path=file_path, source_meta=meta, expansion=eff, source_format=fmt) + stem = os.path.splitext(os.path.basename(file_path))[0] + return buf.getvalue(), stem diff --git a/tests/test_linear_output.py b/tests/test_linear_output.py new file mode 100644 index 00000000..058c7a9a --- /dev/null +++ b/tests/test_linear_output.py @@ -0,0 +1,542 @@ +"""Tests for the Linear Output export feature.""" + +import io +import os + +import numpy as np +import pytest +import tifffile + +from negpy.features.geometry.models import GeometryConfig +from negpy.kernel.image.logic import apply_exif_orientation +from negpy.services.export.linear_output import ( + _CameraWB, + _SourceMeta, + _build_xmp, + _default_pakon_expansion, + _effective_expansion, + _is_camera_raw, + _normalize_wb_rgb, + _write_tiff, + export_linear_output, + export_linear_output_bytes, + is_linear_output_supported, + linear_output_source_type, +) + + +_LINEAR_RAW = 34892 + + +def _make_linearraw_dng_4ch(tmp_dir: str, h: int = 100, w: int = 150) -> str: + """Create a synthetic 4-channel LinearRaw DNG (RGB + IR).""" + rng = np.random.RandomState(99) + data = rng.randint(0, 40000, size=(h, w, 4), dtype=np.uint16) + path = os.path.join(tmp_dir, "scan_4ch.dng") + with tifffile.TiffWriter(path) as tw: + tw.write(data, photometric=_LINEAR_RAW, planarconfig="contig") + return path + + +def _make_linearraw_dng_3ch(tmp_dir: str, h: int = 100, w: int = 150) -> str: + """Create a synthetic 3-channel LinearRaw DNG (RGB, no IR).""" + rng = np.random.RandomState(77) + data = rng.randint(0, 40000, size=(h, w, 3), dtype=np.uint16) + path = os.path.join(tmp_dir, "scan_3ch.dng") + with tifffile.TiffWriter(path) as tw: + tw.write(data, photometric=_LINEAR_RAW, planarconfig="contig") + return path + + +def _make_pakon_raw(tmp_dir: str, h: int = 1000, w: int = 1500) -> str: + """Create a minimal synthetic Pakon RAW file (F135 Plus Low Res, 9 MB).""" + data = np.random.RandomState(42).randint(0, 32768, size=(h, w, 3), dtype=np.uint16) + path = os.path.join(tmp_dir, "test_scan.raw") + data.tofile(path) + assert os.path.getsize(path) == h * w * 3 * 2 # 9000000 + return path + + +def _make_pakon_f335_raw(tmp_dir: str) -> str: + """Create a synthetic F335 RAW file (4000×3000, 72 MB).""" + data = np.random.RandomState(55).randint(0, 65535, size=(4000, 3000, 3), dtype=np.uint16) + path = os.path.join(tmp_dir, "f335_scan.raw") + data.tofile(path) + assert os.path.getsize(path) == 72000000 + return path + + +class TestIsLinearOutputSupported: + def test_pakon_raw_supported(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + assert is_linear_output_supported(path) + + def test_regular_tiff_not_supported(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.tiff") + arr = np.zeros((10, 10, 3), dtype=np.uint16) + tifffile.imwrite(path, arr) + assert not is_linear_output_supported(path) + + def test_nonexistent_raw_supported_by_extension(self) -> None: + """A .raw extension is in SUPPORTED_RAW_EXTENSIONS; support is a format check.""" + assert is_linear_output_supported("/nonexistent/file.raw") + + def test_nonexistent_unknown_ext(self) -> None: + assert not is_linear_output_supported("/nonexistent/file.xyz") + + +class TestExportLinearOutput: + def test_basic_roundtrip(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(raw_path, out_path) + + assert os.path.exists(out_path) + with tifffile.TiffFile(out_path) as tf: + page = tf.pages[0] + arr = page.asarray() + assert arr.dtype == np.uint16 + assert arr.shape == (1000, 1500, 3) + assert page.photometric.name == "RGB" + assert page.iccprofile is None + + def test_no_icc_profile(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(raw_path, out_path) + + with tifffile.TiffFile(out_path) as tf: + assert tf.pages[0].iccprofile is None + + def test_image_description(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(raw_path, out_path) + + with tifffile.TiffFile(out_path) as tf: + desc = tf.pages[0].description + assert "NegPy Linear Output" in desc + assert "no color management" in desc + assert "no WB applied" in desc + assert "Pakon" in desc + assert "F135" in desc + assert "x4" in desc + + def test_pixel_values_roundtrip(self, tmp_path: str) -> None: + """The output uint16 values should be the expanded float32 loader output * 65535, rounded.""" + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + from negpy.infrastructure.loaders.pakon_loader import PakonLoader + from negpy.services.export.linear_output import PAKON_EXPANSION + + loader = PakonLoader() + ctx_mgr, _meta = loader.load(raw_path) + with ctx_mgr as wrapper: + expected_f32 = wrapper.data.copy() + + export_linear_output(raw_path, out_path) + + with tifffile.TiffFile(out_path) as tf: + actual_u16 = tf.pages[0].asarray() + + expected_u16 = np.clip(expected_f32 * PAKON_EXPANSION * 65535.0, 0, 65535).astype(np.uint16) + np.testing.assert_allclose(actual_u16.astype(np.int32), expected_u16.astype(np.int32), atol=1) + + def test_rejects_unsupported_file(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.tiff") + tifffile.imwrite(path, np.zeros((10, 10, 3), dtype=np.uint16)) + out = os.path.join(str(tmp_path), "out.tiff") + with pytest.raises(ValueError, match="not supported"): + export_linear_output(path, out) + + def test_creates_output_directory(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "subdir", "nested", "output.tiff") + + export_linear_output(raw_path, out_path) + assert os.path.exists(out_path) + + +class TestExportLinearOutputBytes: + def test_returns_valid_tiff(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + + tiff_bytes, stem = export_linear_output_bytes(raw_path) + + assert stem == "test_scan" + assert len(tiff_bytes) > 0 + with tifffile.TiffFile(io.BytesIO(tiff_bytes)) as tf: + arr = tf.pages[0].asarray() + assert arr.dtype == np.uint16 + assert arr.shape == (1000, 1500, 3) + + +class TestOrientationHandling: + """Verify that apply_exif_orientation is applied correctly in the export path. + + Pakon always reports orientation=0 (no-op), but the code should handle + nonzero values correctly if a future source provides them. + """ + + def test_orientation_zero_is_identity(self) -> None: + arr = np.arange(24, dtype=np.float32).reshape(2, 4, 3) + result = apply_exif_orientation(arr, 0) + np.testing.assert_array_equal(result, arr) + + def test_orientation_one_is_identity(self) -> None: + arr = np.arange(24, dtype=np.float32).reshape(2, 4, 3) + result = apply_exif_orientation(arr, 1) + np.testing.assert_array_equal(result, arr) + + def test_orientation_6_rotates_cw(self) -> None: + arr = np.arange(24, dtype=np.float32).reshape(2, 4, 3) + result = apply_exif_orientation(arr, 6) + assert result.shape == (4, 2, 3) + expected = np.rot90(arr, 3) + np.testing.assert_array_equal(result, expected) + + def test_orientation_8_rotates_ccw(self) -> None: + arr = np.arange(24, dtype=np.float32).reshape(2, 4, 3) + result = apply_exif_orientation(arr, 8) + assert result.shape == (4, 2, 3) + expected = np.rot90(arr, 1) + np.testing.assert_array_equal(result, expected) + + def test_orientation_3_rotates_180(self) -> None: + arr = np.arange(24, dtype=np.float32).reshape(2, 4, 3) + result = apply_exif_orientation(arr, 3) + assert result.shape == (2, 4, 3) + expected = np.rot90(arr, 2) + np.testing.assert_array_equal(result, expected) + + +class TestGeometryHandling: + """Verify that user rotation/flip from GeometryConfig is applied.""" + + def test_rotation_90cw(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + geo = GeometryConfig(rotation=1) + + export_linear_output(raw_path, out_path, geometry=geo) + + with tifffile.TiffFile(out_path) as tf: + arr = tf.pages[0].asarray() + assert arr.shape == (1500, 1000, 3) + + def test_rotation_180(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + geo = GeometryConfig(rotation=2) + + export_linear_output(raw_path, out_path, geometry=geo) + + with tifffile.TiffFile(out_path) as tf: + arr = tf.pages[0].asarray() + assert arr.shape == (1000, 1500, 3) + + def test_flip_horizontal(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_no_flip = os.path.join(str(tmp_path), "no_flip.tiff") + out_flip = os.path.join(str(tmp_path), "flip.tiff") + + export_linear_output(raw_path, out_no_flip) + export_linear_output(raw_path, out_flip, geometry=GeometryConfig(flip_horizontal=True)) + + with tifffile.TiffFile(out_no_flip) as tf: + arr_orig = tf.pages[0].asarray() + with tifffile.TiffFile(out_flip) as tf: + arr_flip = tf.pages[0].asarray() + + np.testing.assert_array_equal(arr_flip, arr_orig[:, ::-1, :]) + + def test_flip_vertical(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_no_flip = os.path.join(str(tmp_path), "no_flip.tiff") + out_flip = os.path.join(str(tmp_path), "flip.tiff") + + export_linear_output(raw_path, out_no_flip) + export_linear_output(raw_path, out_flip, geometry=GeometryConfig(flip_vertical=True)) + + with tifffile.TiffFile(out_no_flip) as tf: + arr_orig = tf.pages[0].asarray() + with tifffile.TiffFile(out_flip) as tf: + arr_flip = tf.pages[0].asarray() + + np.testing.assert_array_equal(arr_flip, arr_orig[::-1, :, :]) + + def test_fine_rotation_ignored(self, tmp_path: str) -> None: + """Fine rotation involves resampling and should be skipped.""" + raw_path = _make_pakon_raw(str(tmp_path)) + out_plain = os.path.join(str(tmp_path), "plain.tiff") + out_fine = os.path.join(str(tmp_path), "fine.tiff") + + export_linear_output(raw_path, out_plain) + export_linear_output(raw_path, out_fine, geometry=GeometryConfig(fine_rotation=5.0)) + + with tifffile.TiffFile(out_plain) as tf: + arr_plain = tf.pages[0].asarray() + with tifffile.TiffFile(out_fine) as tf: + arr_fine = tf.pages[0].asarray() + + np.testing.assert_array_equal(arr_plain, arr_fine) + + def test_no_geometry_is_identity(self, tmp_path: str) -> None: + raw_path = _make_pakon_raw(str(tmp_path)) + out_none = os.path.join(str(tmp_path), "none.tiff") + out_default = os.path.join(str(tmp_path), "default.tiff") + + export_linear_output(raw_path, out_none) + export_linear_output(raw_path, out_default, geometry=GeometryConfig()) + + with tifffile.TiffFile(out_none) as tf: + arr_none = tf.pages[0].asarray() + with tifffile.TiffFile(out_default) as tf: + arr_default = tf.pages[0].asarray() + + np.testing.assert_array_equal(arr_none, arr_default) + + +class TestDngSupport: + def test_4ch_dng_supported(self, tmp_path: str) -> None: + path = _make_linearraw_dng_4ch(str(tmp_path)) + assert is_linear_output_supported(path) + + def test_3ch_dng_supported(self, tmp_path: str) -> None: + path = _make_linearraw_dng_3ch(str(tmp_path)) + assert is_linear_output_supported(path) + + def test_non_linearraw_dng_supported_as_camera(self, tmp_path: str) -> None: + """A camera DNG (no LinearRaw IFD) is supported via the rawpy path.""" + path = os.path.join(str(tmp_path), "camera.dng") + tifffile.imwrite(path, np.zeros((10, 10, 3), dtype=np.uint16), photometric="rgb") + assert is_linear_output_supported(path) + + def test_4ch_dng_roundtrip(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_4ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(dng_path, out_path) + + assert os.path.exists(out_path) + with tifffile.TiffFile(out_path) as tf: + page = tf.pages[0] + arr = page.asarray() + assert arr.dtype == np.uint16 + assert arr.shape == (100, 150, 3) + assert page.photometric.name == "RGB" + assert page.iccprofile is None + + def test_4ch_dng_ir_written(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_4ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(dng_path, out_path) + + ir_path = os.path.join(str(tmp_path), "output_ir.tiff") + assert os.path.exists(ir_path) + with tifffile.TiffFile(ir_path) as tf: + ir_arr = tf.pages[0].asarray() + assert ir_arr.dtype == np.uint16 + assert ir_arr.shape == (100, 150) + assert "infrared" in tf.pages[0].description + + def test_4ch_dng_pixel_values(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_4ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + source = tifffile.imread(dng_path) + expected_rgb = np.clip(source[:, :, :3].astype(np.float32) / 65535.0 * 65535.0, 0, 65535).astype(np.uint16) + + export_linear_output(dng_path, out_path) + + with tifffile.TiffFile(out_path) as tf: + actual = tf.pages[0].asarray() + np.testing.assert_allclose(actual.astype(np.int32), expected_rgb.astype(np.int32), atol=1) + + def test_3ch_dng_roundtrip(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_3ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(dng_path, out_path) + + assert os.path.exists(out_path) + with tifffile.TiffFile(out_path) as tf: + arr = tf.pages[0].asarray() + assert arr.dtype == np.uint16 + assert arr.shape == (100, 150, 3) + assert tf.pages[0].iccprofile is None + + def test_3ch_dng_no_ir_file(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_3ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + + export_linear_output(dng_path, out_path) + + ir_path = os.path.join(str(tmp_path), "output_ir.tiff") + assert not os.path.exists(ir_path) + + def test_dng_geometry_applied(self, tmp_path: str) -> None: + dng_path = _make_linearraw_dng_4ch(str(tmp_path)) + out_path = os.path.join(str(tmp_path), "output.tiff") + geo = GeometryConfig(rotation=1) + + export_linear_output(dng_path, out_path, geometry=geo) + + with tifffile.TiffFile(out_path) as tf: + arr = tf.pages[0].asarray() + assert arr.shape == (150, 100, 3) + + ir_path = os.path.join(str(tmp_path), "output_ir.tiff") + with tifffile.TiffFile(ir_path) as tf: + ir_arr = tf.pages[0].asarray() + assert ir_arr.shape == (150, 100) + + +class TestCameraRawSupport: + def test_is_camera_raw_nef(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.nef") + open(path, "wb").close() + assert _is_camera_raw(path) + + def test_is_camera_raw_cr2(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.cr2") + open(path, "wb").close() + assert _is_camera_raw(path) + + def test_is_camera_raw_arw(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.arw") + open(path, "wb").close() + assert _is_camera_raw(path) + + def test_tiff_not_camera_raw(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.tiff") + open(path, "wb").close() + assert not _is_camera_raw(path) + + def test_jpeg_not_camera_raw(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.jpg") + open(path, "wb").close() + assert not _is_camera_raw(path) + + def test_pakon_not_camera_raw(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + assert not _is_camera_raw(path) + + def test_dng_is_camera_raw(self, tmp_path: str) -> None: + """A DNG is in SUPPORTED_RAW_EXTENSIONS and not in TIFF/JPEG sets.""" + path = os.path.join(str(tmp_path), "photo.dng") + open(path, "wb").close() + assert _is_camera_raw(path) + + def test_normalize_wb_rgb(self) -> None: + r, g, b = _normalize_wb_rgb((398.0, 302.0, 873.0, 0.0)) + assert g == 1.0 + assert abs(r - 398.0 / 302.0) < 1e-6 + assert abs(b - 873.0 / 302.0) < 1e-6 + + def test_build_xmp_maketiff_format(self) -> None: + wb = _CameraWB( + as_shot=(398.0, 302.0, 873.0, 0.0), + daylight=(1.94, 0.94, 1.38, 0.0), + ) + xmp = _build_xmp("/path/to/DSCF3404.RAF", wb) + text = xmp.decode("utf-8") + assert "RAW-WB:" in text + assert "1.000000" in text + assert "crs:RawFileName" in text + assert "DSCF3404.RAF" in text + assert "dc:description" in text + + def test_camera_raw_supported(self, tmp_path: str) -> None: + """A .nef file should be supported for linear output.""" + path = os.path.join(str(tmp_path), "photo.nef") + open(path, "wb").close() + assert is_linear_output_supported(path) + + def test_write_tiff_with_source_meta(self, tmp_path: str) -> None: + f32 = np.random.RandomState(0).rand(10, 10, 3).astype(np.float32) + out = os.path.join(str(tmp_path), "meta.tiff") + meta = _SourceMeta(make="Plustek", model="OpticFilm 8100", datetime="2025:01:15 12:00:00") + _write_tiff(f32, out, "test.dng", source_meta=meta) + with tifffile.TiffFile(out) as tf: + tags = tf.pages[0].tags + assert tags["Make"].value == "Plustek" + assert tags["Model"].value == "OpticFilm 8100" + assert "2025:01:15" in tags["DateTime"].value + assert tags["Software"].value == "NegPy" + + def test_write_tiff_software_always_set(self, tmp_path: str) -> None: + f32 = np.random.RandomState(0).rand(10, 10, 3).astype(np.float32) + out = os.path.join(str(tmp_path), "sw.tiff") + _write_tiff(f32, out, "test.raw") + with tifffile.TiffFile(out) as tf: + assert tf.pages[0].tags["Software"].value == "NegPy" + + +class TestF335Detection: + def test_f335_detected_by_size(self, tmp_path: str) -> None: + path = _make_pakon_f335_raw(str(tmp_path)) + assert _default_pakon_expansion(path) == 1.0 + + def test_f135_gets_4x(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + assert _default_pakon_expansion(path) == 4.0 + + def test_source_type_f335(self, tmp_path: str) -> None: + path = _make_pakon_f335_raw(str(tmp_path)) + assert linear_output_source_type(path) == "pakon_f335" + + def test_source_type_f135(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + assert linear_output_source_type(path) == "pakon" + + def test_f335_export_no_expansion(self, tmp_path: str) -> None: + path = _make_pakon_f335_raw(str(tmp_path)) + out = os.path.join(str(tmp_path), "output.tiff") + export_linear_output(path, out) + with tifffile.TiffFile(out) as tf: + arr = tf.pages[0].asarray() + assert arr.dtype == np.uint16 + assert arr.shape == (4000, 3000, 3) + assert arr.max() > 0 + desc = tf.pages[0].description + assert "no scaling" in desc + assert "F335" in desc + + def test_f135_description_records_expansion(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + out = os.path.join(str(tmp_path), "output.tiff") + export_linear_output(path, out) + with tifffile.TiffFile(out) as tf: + desc = tf.pages[0].description + assert "x4" in desc + assert "F135" in desc + + def test_effective_expansion_camera_raw(self, tmp_path: str) -> None: + path = os.path.join(str(tmp_path), "photo.nef") + open(path, "wb").close() + assert _effective_expansion(path, None) == 1.0 + assert _effective_expansion(path, 2.0) == 1.0 + + def test_pakon_make_model_tags(self, tmp_path: str) -> None: + path = _make_pakon_raw(str(tmp_path)) + out = os.path.join(str(tmp_path), "output.tiff") + export_linear_output(path, out) + with tifffile.TiffFile(out) as tf: + tags = tf.pages[0].tags + assert tags["Make"].value == "Pakon" + assert "F135 Plus Low Res" in tags["Model"].value + + def test_f335_make_model_tags(self, tmp_path: str) -> None: + path = _make_pakon_f335_raw(str(tmp_path)) + out = os.path.join(str(tmp_path), "output.tiff") + export_linear_output(path, out) + with tifffile.TiffFile(out) as tf: + tags = tf.pages[0].tags + assert tags["Make"].value == "Pakon" + assert "F335" in tags["Model"].value