diff --git a/docs/api_reference.md b/docs/api_reference.md index 7c710189..30a99939 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -38,6 +38,16 @@ and .. autopydantic_model:: mdio.GridOverrides ``` +### SEG-Y File Headers + +```{eval-rst} +.. automodule:: mdio.segy.headers + :members: + +.. automodule:: mdio.segy.text_header + :members: +``` + ## Core Functionality ### Dimensions diff --git a/docs/tutorials/file_headers.ipynb b/docs/tutorials/file_headers.ipynb new file mode 100644 index 00000000..8b5ff37d --- /dev/null +++ b/docs/tutorials/file_headers.ipynb @@ -0,0 +1,227 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Edit SEG-Y File Headers on MDIO\n", + "\n", + "```{article-info}\n", + ":author: Brian Michell\n", + ":date: \"{sub-ref}`today`\"\n", + ":read-time: \"{sub-ref}`wordcount-minutes` min read\"\n", + ":class-container: sd-p-0 sd-outline-muted sd-rounded-3 sd-font-weight-light\n", + "```\n", + "\n", + "SEG-Y text and binary file headers live on the scalar `segy_file_header` variable as\n", + "`textHeader` and `binaryHeader` attributes. They are metadata only. Updating them does\n", + "not read or rewrite traces.\n", + "\n", + "Ingest may skip these fields (`MDIO__IMPORT__SAVE_SEGY_FILE_HEADER` defaults to off).\n", + "Export back to SEG-Y needs both attributes. This tutorial shows how to add them when\n", + "they are missing, and how to replace or patch text and binary headers in place. Edits\n", + "range from one character to a completely new header; you always pass the payload you\n", + "want stored.\n", + "\n", + "```{warning}\n", + "Update **values** of existing binary-header keys. Adding or removing keys can break\n", + "`mdio_to_segy`: the export factory only encodes fields from the SEG-Y spec, and it\n", + "requires `sample_interval`, `samples_per_trace`, and revision major/minor.\n", + "```\n" + ], + "id": "file-headers-intro" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## A tiny MDIO file without headers\n", + "\n", + "Build a small in-memory dataset and write it with `to_mdio`. This is the same shape as a\n", + "file ingested with file-header saving turned off.\n" + ], + "id": "file-headers-setup" + }, + { + "cell_type": "code", + "metadata": {}, + "source": [ + "from pathlib import Path\n", + "\n", + "from mdio import open_mdio\n", + "from mdio import to_mdio\n", + "from mdio import update_segy_file_headers\n", + "from mdio.builder.template_registry import get_template\n", + "from mdio.builder.xarray_builder import to_xarray_dataset\n", + "\n", + "mdio_path = Path(\"headers_demo.mdio\")\n", + "\n", + "template = get_template(\"PostStack2DTime\")\n", + "template.full_chunk_shape = (4, 8)\n", + "mdio_ds = template.build_dataset(name=\"headers-demo\", sizes=(4, 8))\n", + "to_mdio(to_xarray_dataset(mdio_ds), mdio_path, mode=\"w\")\n", + "\n", + "opened = open_mdio(mdio_path)\n", + "print(\"has segy_file_header:\", \"segy_file_header\" in opened)" + ], + "execution_count": null, + "outputs": [], + "id": "file-headers-create" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Add headers when they are missing\n", + "\n", + "`update_segy_file_headers` creates `segy_file_header` if needed and fills defaults.\n", + "With no `template`, defaults follow SEG-Y Revision 1.0. `samples_per_trace` is taken\n", + "from the default data variable (here, 8 time samples).\n", + "\n", + "Pass `template=get_segy_standard(2.0)` (or any `SegySpec`) when the file should carry\n", + "that revision's binary field set.\n" + ], + "id": "file-headers-add" + }, + { + "cell_type": "code", + "metadata": {}, + "source": [ + "headers = update_segy_file_headers(mdio_path)\n", + "\n", + "dataset = open_mdio(mdio_path)\n", + "print(dataset[\"segy_file_header\"].attrs[\"textHeader\"].split(\"\\n\")[0])\n", + "print(\"samples_per_trace\", headers.binary_header[\"samples_per_trace\"])\n", + "print(\"revision\", headers.binary_header[\"segy_revision_major\"], headers.binary_header[\"segy_revision_minor\"])" + ], + "execution_count": null, + "outputs": [], + "id": "file-headers-add-code" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "To seed defaults from a different spec instead of Rev 1.0:\n", + "\n", + "```python\n", + "update_segy_file_headers(mdio_path, template=get_segy_standard(2.0))\n", + "```\n" + ], + "id": "file-headers-template" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Replace or patch in place\n", + "\n", + "Call the same function again. `None` arguments leave that header unchanged.\n", + "\n", + "- **Text:** whatever string you pass replaces `textHeader`. Edit one character, one\n", + " card, or supply a new 40x80 header — same call.\n", + "- **Binary:** keys you pass are merged onto the stored mapping. That is the safe way\n", + " to change a value. A completely new binary header is a full mapping; do not drop\n", + " keys the export spec needs.\n" + ], + "id": "file-headers-edit" + }, + { + "cell_type": "code", + "metadata": {}, + "source": [ + "update_segy_file_headers(mdio_path, binary_header={\"job_id\": 42, \"line_num\": 7})\n", + "\n", + "binary = open_mdio(mdio_path)[\"segy_file_header\"].attrs[\"binaryHeader\"]\n", + "print(\"job_id\", binary[\"job_id\"])\n", + "print(\"line_num\", binary[\"line_num\"])\n", + "print(\"samples_per_trace still\", binary[\"samples_per_trace\"])" + ], + "execution_count": null, + "outputs": [], + "id": "file-headers-edit-code" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Edit the text header in Python, then write it back\n", + "\n", + "The stored value is one string (40 cards of 80 characters, joined by newlines).\n", + "Change as little or as much as you want, then pass the whole string.\n" + ], + "id": "file-headers-line" + }, + { + "cell_type": "code", + "metadata": {}, + "source": [ + "text = open_mdio(mdio_path)[\"segy_file_header\"].attrs[\"textHeader\"]\n", + "rows = text.split(\"\\n\")\n", + "\n", + "# one character\n", + "rows[0] = rows[0][:4] + \"X\" + rows[0][5:]\n", + "\n", + "# one card\n", + "rows[1] = \"C02 LINE: DEMO-001\".ljust(80)\n", + "\n", + "# or assign a completely new 40-line header to `rows` / `text`\n", + "update_segy_file_headers(mdio_path, text_header=\"\\n\".join(rows))\n", + "\n", + "updated = open_mdio(mdio_path)[\"segy_file_header\"].attrs[\"textHeader\"]\n", + "print(updated.split(\"\\n\")[0])\n", + "print(updated.split(\"\\n\")[1])" + ], + "execution_count": null, + "outputs": [], + "id": "file-headers-line-code" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Binary headers: update values, do not reshape the field set\n", + "\n", + "Safe:\n", + "\n", + "```python\n", + "update_segy_file_headers(mdio_path, binary_header={\"job_id\": 99})\n", + "```\n", + "\n", + "Risky:\n", + "\n", + "- **Adding** a key that is not in the SEG-Y spec used at export. `SegyFactory` raises\n", + " when it cannot place the field.\n", + "- **Removing** a required key (`sample_interval`, `samples_per_trace`,\n", + " `segy_revision_major`, `segy_revision_minor`). Export then fails or writes a bad file.\n", + "- Switching revision by stuffing extra Rev 2 fields into a Rev 1 header (or the reverse)\n", + " without passing a matching `template` the next time defaults are built.\n", + "\n", + "This helper never deletes keys: a partial `binary_header` is merged onto what is already\n", + "stored. To change revision layout, pass `template` when the file still lacks a binary\n", + "header, or replace the whole mapping only if you know the export `SegySpec`.\n" + ], + "id": "file-headers-binary-warn" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Recap\n", + "\n", + "- `update_segy_file_headers` adds or writes `textHeader` / `binaryHeader` without touching traces.\n", + "- Missing fields get Rev 1.0 defaults, or a `SegySpec` you pass as `template`.\n", + "- Text: pass the full string, whether you changed one character or all 40 cards.\n", + "- Binary: merge value updates. Do not add or remove keys unless the export spec matches.\n" + ], + "id": "file-headers-recap" + } + ], + "metadata": { + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/tutorials/index.md b/docs/tutorials/index.md index df5ec7d3..cc77084a 100644 --- a/docs/tutorials/index.md +++ b/docs/tutorials/index.md @@ -14,4 +14,5 @@ compression rechunking corrupt_files custom_template +file_headers ``` diff --git a/src/mdio/__init__.py b/src/mdio/__init__.py index efd76492..4f19231f 100644 --- a/src/mdio/__init__.py +++ b/src/mdio/__init__.py @@ -12,6 +12,8 @@ from mdio.optimize.access_pattern import OptimizedAccessPatternConfig from mdio.optimize.access_pattern import optimize_access_patterns from mdio.segy.geometry import GridOverrides +from mdio.segy.headers import SegyFileHeaders +from mdio.segy.headers import update_segy_file_headers try: __version__ = metadata.version("multidimio") @@ -22,8 +24,10 @@ __all__ = [ "__version__", "GridOverrides", + "SegyFileHeaders", "open_mdio", "to_mdio", + "update_segy_file_headers", "mdio_to_segy", "segy_to_mdio", "OptimizedAccessPatternConfig", diff --git a/src/mdio/segy/__init__.py b/src/mdio/segy/__init__.py index 4347a73e..296c0ed2 100644 --- a/src/mdio/segy/__init__.py +++ b/src/mdio/segy/__init__.py @@ -1 +1,9 @@ """SEG-Y specific implementation module.""" + +from mdio.segy.headers import SegyFileHeaders +from mdio.segy.headers import update_segy_file_headers + +__all__ = [ + "SegyFileHeaders", + "update_segy_file_headers", +] diff --git a/src/mdio/segy/creation.py b/src/mdio/segy/creation.py index 2a588087..a1fdeef9 100644 --- a/src/mdio/segy/creation.py +++ b/src/mdio/segy/creation.py @@ -95,8 +95,8 @@ def mdio_spec_to_segy( if "segy_file_header" not in dataset: msg = ( - "MDIO does not contain SEG-Y file headers to write to output. Please add a dummy segy_file_header " - "variable and fill its metadata (.attrs) with `textHeader` and `binaryHeader`." + "MDIO does not contain SEG-Y file headers to write to output. " + "Call mdio.update_segy_file_headers(...) to add textHeader and binaryHeader." ) raise MDIOMissingVariableError(msg) diff --git a/src/mdio/segy/headers.py b/src/mdio/segy/headers.py new file mode 100644 index 00000000..3d907bea --- /dev/null +++ b/src/mdio/segy/headers.py @@ -0,0 +1,260 @@ +"""In-place updates for SEG-Y text and binary file headers on an MDIO dataset. + +Headers live as attributes on the scalar ``segy_file_header`` variable. Reads use +:func:`mdio.open_mdio`. Writes go through Zarr because xarray does not persist attribute +updates on an existing array. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import numpy as np +import zarr +from segy.factory import SegyFactory +from segy.factory import get_default_text +from segy.standards import get_segy_standard + +from mdio.api.io import _normalize_path +from mdio.api.io import _normalize_storage_options +from mdio.api.io import open_mdio +from mdio.constants import ZarrFormat +from mdio.core.zarr_io import zarr_warnings_suppress_unstable_structs_v3 +from mdio.exceptions import MDIONotFoundError +from mdio.segy.text_header import sanitize_text_header +from mdio.segy.text_header import validate_text_header + +if TYPE_CHECKING: + from collections.abc import Mapping + from pathlib import Path + + from segy.schema import SegySpec + from upath import UPath + from xarray import Dataset as xr_Dataset + from zarr import Array as ZarrArray + from zarr import Group as ZarrGroup + + +logger = logging.getLogger(__name__) + +SEGY_FILE_HEADER_VARIABLE = "segy_file_header" +TEXT_HEADER_ATTR = "textHeader" +BINARY_HEADER_ATTR = "binaryHeader" + +_DEFAULT_SAMPLE_INTERVAL = 4000 +_DEFAULT_SAMPLES_PER_TRACE = 1500 + + +@dataclass(frozen=True, slots=True) +class SegyFileHeaders: + """Resolved SEG-Y text and binary file headers stored on an MDIO dataset. + + Attributes: + text_header: Sanitized 40x80 textual file header. + binary_header: Binary header fields as a JSON-friendly integer mapping. + """ + + text_header: str + binary_header: dict[str, int] + + +def update_segy_file_headers( + mdio_path: UPath | Path | str, + *, + text_header: str | None = None, + binary_header: Mapping[str, int] | None = None, + template: SegySpec | None = None, +) -> SegyFileHeaders: + """Update SEG-Y text and/or binary headers on an existing MDIO file. + + Reads the dataset with :func:`mdio.open_mdio` (lazy chunks) and writes only + ``segy_file_header`` attributes. Trace data is not read or rewritten. + + ``None`` arguments leave that header unchanged when it already exists. If the variable + or an attribute is missing, a default is written. Defaults come from ``template`` when + given, otherwise SEG-Y Revision 1.0. ``samples_per_trace`` is taken from the default + data variable shape when that metadata is present. + + User ``binary_header`` values are merged onto the existing header, or onto the default + header when the file has none. Prefer updating values of keys that already exist. + Adding or removing binary fields can break SEG-Y export. + + Args: + mdio_path: Local or remote path to the MDIO store. + text_header: Replacement textual header. Sanitized to the 40x80 ASCII card layout. + binary_header: Binary header fields to set or overlay. Integer-valued mapping. + template: SEG-Y spec used to build defaults when the file lacks header fields. + + Returns: + The text and binary headers stored after the update. + + Raises: + MDIONotFoundError: If ``mdio_path`` does not exist. + ValueError: If a provided text header cannot be sanitized or a binary field is not + an integer. + """ + path = _normalize_path(mdio_path) + if not path.exists(): + msg = f"MDIO file not found: {path}" + raise MDIONotFoundError(msg) + dataset = open_mdio(path, chunks={}) + + existing_text = _read_text_header(dataset) + existing_binary = _read_binary_header(dataset) + user_text = sanitize_text_header(text_header) if text_header is not None else None + user_binary = _coerce_binary_header(binary_header) if binary_header is not None else None + if user_binary is not None: + _warn_unknown_binary_keys(user_binary, template) + + default_text: str | None = None + default_binary: dict[str, int] | None = None + if (user_text is None and existing_text is None) or existing_binary is None: + spec = _resolve_spec(template) + sample_interval, samples_per_trace = _resolve_factory_params(dataset, existing_binary) + default_text, default_binary = _build_defaults(spec, sample_interval, samples_per_trace) + + if user_text is not None: + resolved_text = user_text + elif existing_text is not None: + resolved_text = existing_text + else: + resolved_text = default_text + + base_binary = existing_binary if existing_binary is not None else default_binary + if resolved_text is None or base_binary is None: + msg = "Failed to resolve SEG-Y file headers" + raise ValueError(msg) + resolved_binary = {**base_binary, **user_binary} if user_binary is not None else base_binary + + validate_text_header(resolved_text) + _write_header_attrs(path, resolved_text, resolved_binary) + return SegyFileHeaders(text_header=resolved_text, binary_header=resolved_binary) + + +def _read_text_header(dataset: xr_Dataset) -> str | None: + """Return the stored text header when it is a string.""" + if SEGY_FILE_HEADER_VARIABLE not in dataset: + return None + value = dataset[SEGY_FILE_HEADER_VARIABLE].attrs.get(TEXT_HEADER_ATTR) + return value if isinstance(value, str) else None + + +def _read_binary_header(dataset: xr_Dataset) -> dict[str, int] | None: + """Return the stored binary header when it is a mapping of integers.""" + if SEGY_FILE_HEADER_VARIABLE not in dataset: + return None + value = dataset[SEGY_FILE_HEADER_VARIABLE].attrs.get(BINARY_HEADER_ATTR) + if not isinstance(value, dict): + return None + return _coerce_binary_header(value) + + +def _resolve_spec(template: SegySpec | None) -> SegySpec: + """Return a writable spec copy. Rev 1.0 when the caller did not pass a template.""" + if template is None: + return get_segy_standard(1.0) + return template.model_copy(deep=True) + + +def _resolve_factory_params(dataset: xr_Dataset, existing_binary: dict[str, int] | None) -> tuple[int, int]: + """Pick sample interval and samples-per-trace for default header generation.""" + sample_interval = _DEFAULT_SAMPLE_INTERVAL + samples_per_trace = _infer_samples_per_trace(dataset) or _DEFAULT_SAMPLES_PER_TRACE + if existing_binary is not None: + sample_interval = existing_binary.get("sample_interval", sample_interval) + samples_per_trace = existing_binary.get("samples_per_trace", samples_per_trace) + return sample_interval, samples_per_trace + + +def _infer_samples_per_trace(dataset: xr_Dataset) -> int | None: + """Return the last-axis length of the default data variable, if present.""" + attributes = dataset.attrs.get("attributes") + if not isinstance(attributes, dict): + return None + variable_name = attributes.get("defaultVariableName") + if not isinstance(variable_name, str) or variable_name not in dataset: + return None + data = dataset[variable_name] + if not data.dims: + return None + return int(data.sizes[data.dims[-1]]) + + +def _build_defaults(spec: SegySpec, sample_interval: int, samples_per_trace: int) -> tuple[str, dict[str, int]]: + """Build default text and binary headers from a SEG-Y spec.""" + factory = SegyFactory(spec=spec, sample_interval=sample_interval, samples_per_trace=samples_per_trace) + text_header = get_default_text(factory.spec) + binary_header = _binary_bytes_to_mdio_dict(factory.create_binary_header(), factory.spec) + return text_header, binary_header + + +def _binary_bytes_to_mdio_dict(raw: bytes, spec: SegySpec) -> dict[str, int]: + """Decode factory binary-header bytes into the MDIO attribute mapping.""" + parsed = np.frombuffer(raw, dtype=spec.binary_header.dtype) + header = {name: int(parsed[name][0]) for name in parsed.dtype.names or ()} + return _normalize_revision_keys(header) + + +def _normalize_revision_keys(binary_header: dict[str, int]) -> dict[str, int]: + """Store revision as major/minor, matching SEG-Y ingest.""" + normalized = dict(binary_header) + if "segy_revision_major" in normalized and "segy_revision_minor" in normalized: + normalized.pop("segy_revision", None) + return normalized + if "segy_revision" in normalized: + code = normalized.pop("segy_revision") + normalized["segy_revision_major"] = (code >> 8) & 0xFF + normalized["segy_revision_minor"] = code & 0xFF + return normalized + + +def _coerce_binary_header(binary_header: Mapping[str, object]) -> dict[str, int]: + """Integer-coerce a user or stored binary header mapping.""" + coerced: dict[str, int] = {} + for key, value in binary_header.items(): + try: + coerced[key] = int(value) + except (TypeError, ValueError) as exc: + msg = f"Binary header field {key!r} must be an integer, got {value!r}" + raise ValueError(msg) from exc + return _normalize_revision_keys(coerced) + + +def _warn_unknown_binary_keys(user_binary: dict[str, int], template: SegySpec | None) -> None: + """Warn when the user adds fields the SEG-Y template cannot encode.""" + spec = _resolve_spec(template) + known = set(spec.binary_header.names) + known.discard("segy_revision") + known.update({"segy_revision_major", "segy_revision_minor"}) + unknown = sorted(set(user_binary) - known) + if unknown: + logger.warning( + "Binary header fields %s are not in the SEG-Y template; adding or removing fields can break export.", + unknown, + ) + + +def _write_header_attrs(path: UPath, text_header: str, binary_header: dict[str, int]) -> None: + """Persist header attributes through Zarr.""" + storage_options = _normalize_storage_options(path) + zarr_format = zarr.config.get("default_zarr_format") + group = zarr.open_group( + path.as_posix(), + mode="r+", + storage_options=storage_options, + use_consolidated=zarr_format == ZarrFormat.V2, + ) + header_array = _ensure_header_array(group) + header_array.attrs.update({TEXT_HEADER_ATTR: text_header, BINARY_HEADER_ATTR: binary_header}) + if zarr_format == ZarrFormat.V2: + zarr.consolidate_metadata(group.store) + + +def _ensure_header_array(group: ZarrGroup) -> ZarrArray: + """Return the scalar header variable, creating it when the store has none.""" + if SEGY_FILE_HEADER_VARIABLE in group: + return group[SEGY_FILE_HEADER_VARIABLE] + with zarr_warnings_suppress_unstable_structs_v3(): + return group.create_array(SEGY_FILE_HEADER_VARIABLE, shape=(), dtype="U1", fill_value="") diff --git a/tests/unit/test_update_segy_file_headers.py b/tests/unit/test_update_segy_file_headers.py new file mode 100644 index 00000000..041ece0d --- /dev/null +++ b/tests/unit/test_update_segy_file_headers.py @@ -0,0 +1,211 @@ +"""Tests for cheap in-place SEG-Y file header updates on MDIO stores.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import numpy as np +import pytest +import xarray as xr +from segy.standards import get_segy_standard + +from mdio import open_mdio +from mdio import to_mdio +from mdio import update_segy_file_headers +from mdio.exceptions import MDIONotFoundError +from mdio.segy.headers import BINARY_HEADER_ATTR +from mdio.segy.headers import SEGY_FILE_HEADER_VARIABLE +from mdio.segy.headers import TEXT_HEADER_ATTR +from mdio.segy.text_header import EXPECTED_COLS +from mdio.segy.text_header import EXPECTED_ROWS +from mdio.segy.text_header import validate_text_header + +if TYPE_CHECKING: + from pathlib import Path + + +def _well_formed_header(prefix: str = "C") -> str: + """Build a 40x80 header with a distinctive prefix on each card.""" + rows = [f"{prefix}{i:02d}".ljust(EXPECTED_COLS) for i in range(1, EXPECTED_ROWS + 1)] + return "\n".join(rows) + + +def _write_minimal_store( + path: Path, + *, + with_header_var: bool = True, + text_header: str | None = None, + binary_header: dict[str, int] | None = None, + samples: int = 8, +) -> Path: + """Create a tiny MDIO store through the public write API.""" + dataset = xr.Dataset( + {"amplitude": (("sample",), np.arange(samples, dtype=np.float32))}, + attrs={"attributes": {"defaultVariableName": "amplitude"}}, + ) + if with_header_var: + dataset[SEGY_FILE_HEADER_VARIABLE] = ((), "") + attrs: dict[str, object] = {} + if text_header is not None: + attrs[TEXT_HEADER_ATTR] = text_header + if binary_header is not None: + attrs[BINARY_HEADER_ATTR] = binary_header + if attrs: + dataset[SEGY_FILE_HEADER_VARIABLE].attrs.update(attrs) + to_mdio(dataset, path, mode="w") + return path + + +def _stored_headers(path: Path) -> tuple[str, dict[str, int]]: + """Read header attrs back through ``open_mdio``.""" + attrs = open_mdio(path)[SEGY_FILE_HEADER_VARIABLE].attrs + return attrs[TEXT_HEADER_ATTR], dict(attrs[BINARY_HEADER_ATTR]) + + +class TestUpdateSegyFileHeaders: + """In-place header updates write attrs only and fill missing fields.""" + + def test_missing_path_raises(self, tmp_path: Path) -> None: + """Unknown store path raises ``MDIONotFoundError``.""" + with pytest.raises(MDIONotFoundError, match="not found"): + update_segy_file_headers(tmp_path / "missing.mdio") + + def test_updates_text_header_only(self, tmp_path: Path) -> None: + """Provided text replaces the stored text; binary stays put.""" + original_binary = {"job_id": 7, "sample_interval": 2000, "samples_per_trace": 8} + store = _write_minimal_store( + tmp_path / "headers.mdio", + text_header=_well_formed_header("A"), + binary_header=original_binary, + ) + new_text = _well_formed_header("B") + + result = update_segy_file_headers(store, text_header=new_text) + + stored_text, stored_binary = _stored_headers(store) + assert result.text_header == new_text + assert stored_text == new_text + assert stored_binary == original_binary + + def test_merges_binary_header_fields(self, tmp_path: Path) -> None: + """User binary fields overlay the existing mapping.""" + store = _write_minimal_store( + tmp_path / "headers.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1, "line_num": 2, "sample_interval": 2000}, + ) + + result = update_segy_file_headers(store, binary_header={"job_id": 99}) + + _, stored_binary = _stored_headers(store) + assert result.binary_header["job_id"] == 99 + assert stored_binary["job_id"] == 99 + assert stored_binary["line_num"] == 2 + assert stored_binary["sample_interval"] == 2000 + + def test_sanitizes_short_text_header(self, tmp_path: Path) -> None: + """Short user text is padded to the 40x80 card layout.""" + store = _write_minimal_store( + tmp_path / "headers.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1}, + ) + + result = update_segy_file_headers(store, text_header="C01 CLIENT") + + validate_text_header(result.text_header) + assert result.text_header.split("\n")[0].startswith("C01 CLIENT") + + def test_creates_missing_variable_with_defaults(self, tmp_path: Path) -> None: + """Store without ``segy_file_header`` gets a scalar var plus Rev1 defaults.""" + store = _write_minimal_store(tmp_path / "no_headers.mdio", with_header_var=False, samples=8) + + result = update_segy_file_headers(store) + + validate_text_header(result.text_header) + stored_text, stored_binary = _stored_headers(store) + assert stored_text == result.text_header + assert stored_binary["samples_per_trace"] == 8 + assert stored_binary["segy_revision_major"] == 1 + assert stored_binary["segy_revision_minor"] == 0 + assert "sample_interval" in stored_binary + + def test_fills_missing_attrs_on_existing_variable(self, tmp_path: Path) -> None: + """Empty header variable receives default text and binary attrs.""" + store = _write_minimal_store(tmp_path / "empty_attrs.mdio", with_header_var=True) + + result = update_segy_file_headers(store) + + validate_text_header(result.text_header) + _, stored_binary = _stored_headers(store) + assert stored_binary["samples_per_trace"] == 8 + + def test_user_binary_overlays_defaults_when_missing(self, tmp_path: Path) -> None: + """Partial user binary sits on top of the generated default header.""" + store = _write_minimal_store(tmp_path / "partial.mdio", with_header_var=False) + + result = update_segy_file_headers(store, binary_header={"job_id": 42}) + + assert result.binary_header["job_id"] == 42 + assert result.binary_header["samples_per_trace"] == 8 + assert result.binary_header["segy_revision_major"] == 1 + + def test_template_controls_default_revision(self, tmp_path: Path) -> None: + """Optional SegySpec template drives default binary field set and revision.""" + store = _write_minimal_store(tmp_path / "rev2.mdio", with_header_var=False) + + result = update_segy_file_headers(store, template=get_segy_standard(2.0)) + + assert result.binary_header["segy_revision_major"] == 2 + assert result.binary_header["segy_revision_minor"] == 0 + assert "extended_samples_per_trace" in result.binary_header + + def test_does_not_rewrite_amplitude_payload(self, tmp_path: Path) -> None: + """Header update leaves the data array bytes unchanged.""" + store = _write_minimal_store( + tmp_path / "payload.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1}, + ) + before = open_mdio(store)["amplitude"].values.copy() + + update_segy_file_headers(store, binary_header={"job_id": 2}) + + after = open_mdio(store)["amplitude"].values + np.testing.assert_array_equal(before, after) + + def test_rejects_non_integer_binary_value(self, tmp_path: Path) -> None: + """Non-integer binary field values raise ``ValueError``.""" + store = _write_minimal_store( + tmp_path / "bad.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1}, + ) + with pytest.raises(ValueError, match="must be an integer"): + update_segy_file_headers(store, binary_header={"job_id": "not-a-number"}) # type: ignore[dict-item] + + def test_normalizes_encoded_revision(self, tmp_path: Path) -> None: + """Encoded ``segy_revision`` is stored as major/minor like ingest.""" + store = _write_minimal_store( + tmp_path / "rev.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1}, + ) + + result = update_segy_file_headers(store, binary_header={"segy_revision": 256}) + + assert "segy_revision" not in result.binary_header + assert result.binary_header["segy_revision_major"] == 1 + assert result.binary_header["segy_revision_minor"] == 0 + + def test_unknown_binary_key_warns(self, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Fields outside the SEG-Y template warn; export can break if they are added.""" + store = _write_minimal_store( + tmp_path / "unknown.mdio", + text_header=_well_formed_header(), + binary_header={"job_id": 1}, + ) + with caplog.at_level(logging.WARNING, logger="mdio.segy.headers"): + update_segy_file_headers(store, binary_header={"not_a_segy_field": 1}) + assert any("not_a_segy_field" in record.message for record in caplog.records)