Lint and tighten up code

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: BrianMichell

src/mdio/core/config.py

@@ -17,15 +17,6 @@
     SAVE_SEGY_FILE_HEADER_STRICT,
     SAVE_SEGY_FILE_HEADER_LENIENT,
 ]
-"""Mode for ``MDIO__IMPORT__SAVE_SEGY_FILE_HEADER``.
-
-* ``0`` (also accepts ``False`` / ``"false"``): do not save SEG-Y file headers.
-* ``1`` (also accepts ``True`` / ``"true"``): save SEG-Y file headers and raise
-  on a malformed text header.
-* ``2``: save SEG-Y file headers and, on a malformed text header, log a
-  warning and correct it (non-ASCII or non-printable characters become spaces
-  and the header is padded to 80x40).
-"""
 
 _SAVE_HEADER_TRUE_STRINGS = frozenset({"true", "yes", "on"})
 _SAVE_HEADER_FALSE_STRINGS = frozenset({"false", "no", "off"})

src/mdio/segy/creation.py

@@ -31,32 +31,18 @@
 
 
 def _ensure_exportable_text_header(text_header: str) -> str:
-    """Validate the stored text header and repair it if it cannot be encoded.
-
-    MDIO stores the text header as a wrapped 40x80 string. Stores written by
-    older versions of MDIO may contain non-ASCII characters (typically
-    ``U+FFFD`` from a lossy EBCDIC import) that cannot be re-encoded to ASCII
-    by the SEG-Y factory. To keep export usable for those stores this helper
-    runs the validator and, on failure, sanitizes the header in place and logs
-    a warning rather than aborting the export.
+    """Validate the stored text header; repair and warn if it cannot be ASCII-encoded.
 
     Args:
         text_header: The ``textHeader`` attribute as stored on the MDIO dataset.
 
     Returns:
-        A text header string that satisfies
-        :func:`mdio.segy.text_header.validate_text_header` and is therefore
-        guaranteed to round-trip through ``factory.create_textual_header``.
+        A text header string that satisfies :func:`validate_text_header`.
     """
     try:
         validate_text_header(text_header)
     except ValueError as exc:
-        logger.warning(
-            "Stored MDIO text header is not exportable as-is and will be repaired: %s. "
-            "The repair replaces non-ASCII or non-printable characters with spaces and "
-            "forces the 80x40 card layout. Re-ingest the source SEG-Y to remove this warning.",
-            exc,
-        )
+        logger.warning("Stored MDIO text header is not exportable as-is and will be repaired: %s", exc)
         return sanitize_text_header(text_header)
     return text_header
 

src/mdio/segy/text_header.py

@@ -1,69 +1,22 @@
-"""SEG-Y textual file header validation and sanitization helpers.
-
-The SEG-Y standard defines the textual file header as a 3200-byte block
-organized as 40 cards of 80 characters each, encoded as either ASCII or
-EBCDIC. Both encodings used by ``TGSAI/segy`` ultimately require the
-in-memory string to be 7-bit ASCII (``ord(c) <= 127``) before bytes can be
-written. The MDIO on-disk representation is the wrapped form: 40 lines of
-exactly 80 characters joined by ``"\\n"``.
-
-When the source bytes were ingested through a lossy EBCDIC decode, MDIO
-typically receives ``U+FFFD`` (``"\uFFFD"``) replacement characters and other
-non-ASCII codepoints. Those characters round-trip through MDIO storage but
-fail when ``segy.factory.create_textual_header`` tries to re-encode the
-header to ASCII for SEG-Y export. The helpers in this module exist to detect
-that situation up-front and, when requested, repair it deterministically.
-
-Repairs are conservative: any character that is either non-ASCII
-(``ord(c) > 127``) or non-printable per :py:meth:`str.isprintable` is replaced
-with an ASCII space, and the card grid is forced to exactly 40 rows of 80
-columns. Newlines (``"\\n"``) are treated only as row separators; other
-Unicode line-break characters (``"\\v"``, ``"\\f"``, ``"\\x85"``, ``"\u2028"``,
-``"\u2029"``) are treated as content and replaced rather than re-splitting
-the layout. Sanitization additionally collapses runs of two or more
-``"\\n"`` to one so headers that were written with ``"\\n\\n"`` between
-cards are not silently truncated to half their length.
-"""
+"""SEG-Y textual file header validation and sanitization helpers."""
 
 from __future__ import annotations
 
 import re
 
 EXPECTED_ROWS = 40
 EXPECTED_COLS = 80
-EXPECTED_LENGTH = EXPECTED_ROWS * EXPECTED_COLS
 ASCII_MAX_ORD = 127
 
 _REPORT_LIMIT = 5
 _NEWLINE_RUN = re.compile(r"\n{2,}")
 
 
 def _is_safe_char(char: str) -> bool:
-    """Return True if a char is safe to round-trip through SEG-Y ASCII/EBCDIC.
-
-    A char is "safe" when it is both 7-bit ASCII (``ord <= 127``) and printable
-    per :py:meth:`str.isprintable`. ASCII space passes; ``U+FFFD``, accented
-    Latin characters, control characters and tabs do not.
-    """
+    """Return True if char is 7-bit ASCII and printable."""
     return ord(char) <= ASCII_MAX_ORD and char.isprintable()
 
 
-def _split_rows(text_header: str) -> list[str]:
-    """Split a wrapped text header into rows on ``"\\n"`` only.
-
-    Other Unicode line-break characters (``"\\v"``, ``"\\f"``, ``"\u0085"``, etc.)
-    are intentionally left in place so that lossy decodes do not silently
-    re-shape the card grid. They will surface as unsafe characters during
-    validation and be replaced during sanitization.
-    """
-    return text_header.split("\n")
-
-
-def _find_unsafe(row: str) -> list[int]:
-    """Return positions of characters that are not :func:`_is_safe_char`."""
-    return [i for i, c in enumerate(row) if not _is_safe_char(c)]
-
-
 def _summarize(mapping: dict[int, list[int]], limit: int = _REPORT_LIMIT) -> str:
     """Format ``{row: [positions]}`` for an error message, capped for readability."""
     if not mapping:
@@ -80,18 +33,15 @@ def _summarize(mapping: dict[int, list[int]], limit: int = _REPORT_LIMIT) -> str
 
 
 def validate_text_header(text_header: str) -> None:
-    """Validate a SEG-Y textual file header is 40 rows of 80 ASCII-printable characters.
+    r"""Validate a SEG-Y textual file header is 40 rows of 80 ASCII-printable characters.
 
     Args:
-        text_header: Decoded textual file header string in the wrapped form
-            (40 rows of 80 characters joined by ``"\\n"``).
+        text_header: Decoded text header in wrapped form (40 rows of 80 chars joined by ``\n``).
 
     Raises:
-        ValueError: If the header does not split into exactly 40 rows on
-            ``"\\n"``, any row is not 80 characters wide, or any character is
-            not safe to encode as 7-bit ASCII (see :func:`_is_safe_char`).
+        ValueError: If row count, row width, or any character fails the SEG-Y ASCII contract.
     """
-    rows = _split_rows(text_header)
+    rows = text_header.split("\n")
 
     if len(rows) != EXPECTED_ROWS:
         err = f"Invalid text header line count: expected {EXPECTED_ROWS}, got {len(rows)}"
@@ -106,46 +56,30 @@ def validate_text_header(text_header: str) -> None:
 
     bad_chars: dict[int, list[int]] = {}
     for i, row in enumerate(rows):
-        positions = _find_unsafe(row)
+        positions = [j for j, c in enumerate(row) if not _is_safe_char(c)]
         if positions:
             bad_chars[i] = positions
 
     if bad_chars:
-        err = (
-            "Invalid text header characters: non-ASCII or non-printable at "
-            f"{_summarize(bad_chars)}"
-        )
+        err = f"Invalid text header characters: non-ASCII or non-printable at {_summarize(bad_chars)}"
         raise ValueError(err)
 
 
 def sanitize_text_header(text_header: str) -> str:
-    """Coerce a SEG-Y textual file header into the 40x80 ASCII-printable card layout.
-
-    Pre-processing collapses runs of two or more ``"\\n"`` into one. Some SEG-Y
-    writers terminate each card with ``"\\n\\n"``, which yields 80 rows on a
-    naive ``split("\\n")`` and would silently drop cards 21-40 when the row
-    list is sliced to 40. Collapsing runs of newlines recovers the intended
-    card layout for that common case while leaving properly-wrapped headers
-    untouched.
-
-    The normalized input is then split on ``"\\n"`` and each row is independently:
-
-    1. Stripped of unsafe characters (any non-ASCII or non-printable codepoint
-       is replaced with a single ASCII space).
-    2. Right-padded with spaces or truncated to exactly 80 characters.
+    r"""Coerce a SEG-Y textual file header into the 40x80 ASCII-printable card layout.
 
-    Rows beyond 40 are dropped. Missing rows are appended as 80-space blanks
-    so the result always contains exactly 40 lines.
+    Runs of two or more ``\n`` collapse to one (some writers terminate cards with ``\n\n``).
+    Each row gets unsafe characters replaced with spaces and is padded/truncated to 80 chars.
+    The result always has exactly 40 rows.
 
     Args:
         text_header: Decoded textual file header string.
 
     Returns:
-        Sanitized header string with rows joined by ``"\\n"``. The output is
-        guaranteed to satisfy :func:`validate_text_header`.
+        Sanitized header that satisfies :func:`validate_text_header`.
     """
     normalized = _NEWLINE_RUN.sub("\n", text_header)
-    rows = _split_rows(normalized)
+    rows = normalized.split("\n")
 
     sanitized: list[str] = []
     for row in rows[:EXPECTED_ROWS]:

tests/unit/ingestion/test_segy_file_headers.py

@@ -81,7 +81,7 @@ def test_invalid_row_count_raises(self) -> None:
         ds = _empty_dataset()
         with (
             patch.dict(os.environ, {"MDIO__IMPORT__SAVE_SEGY_FILE_HEADER": "true"}),
-            pytest.raises(ValueError, match="Invalid text header count"),
+            pytest.raises(ValueError, match="Invalid text header line count"),
         ):
             _add_segy_file_headers(ds, info)
 
@@ -93,6 +93,6 @@ def test_invalid_column_count_raises(self) -> None:
         ds = _empty_dataset()
         with (
             patch.dict(os.environ, {"MDIO__IMPORT__SAVE_SEGY_FILE_HEADER": "true"}),
-            pytest.raises(ValueError, match="Invalid text header columns"),
+            pytest.raises(ValueError, match="Invalid text header line widths"),
         ):
             _add_segy_file_headers(ds, info)

tests/unit/test_segy_export_text_header.py

@@ -1,27 +1,26 @@
-"""Tests for export-side text-header guarding in ``mdio.segy.creation``.
-
-These cover the second half of issue #814: existing MDIO stores written by an
-older version of MDIO may carry a malformed text header (typically scattered
-``U+FFFD`` characters from a lossy EBCDIC import). The export path must not
-crash on those stores; it should repair the header and warn instead.
-"""
+"""Tests for export-side text header guarding in ``mdio.segy.creation``."""
 
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 
-import pytest
 from segy.factory import SegyFactory
 from segy.standards import get_segy_standard
 
 from mdio.segy.creation import _ensure_exportable_text_header
 
+if TYPE_CHECKING:
+    import pytest
+
 
 def _well_formed_header() -> str:
+    """Build a 40x80 header where each row reads ``Cnn ...spaces``."""
     return "\n".join([f"C{i:02d}".ljust(80) for i in range(1, 41)])
 
 
 def _replacement_char_header() -> str:
+    """Build a 40x80 header with U+FFFD scattered through the last three cards."""
     rows = [f"C{i:02d}".ljust(80) for i in range(1, 41)]
     rows[37] = "\ufffdC38" + " " * 76
     rows[38] = "\ufffdC39" + " " * 76
@@ -33,21 +32,23 @@ class TestEnsureExportableTextHeader:
     """The export guard repairs malformed headers and warns; otherwise no-op."""
 
     def test_passthrough_when_well_formed(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Well-formed input is returned unchanged with no warning."""
         header = _well_formed_header()
         with caplog.at_level(logging.WARNING, logger="mdio.segy.creation"):
             result = _ensure_exportable_text_header(header)
         assert result == header
         assert not any("repaired" in record.message for record in caplog.records)
 
     def test_repairs_replacement_char_and_warns(self, caplog: pytest.LogCaptureFixture) -> None:
+        """U+FFFD is repaired and a warning is logged."""
         with caplog.at_level(logging.WARNING, logger="mdio.segy.creation"):
             result = _ensure_exportable_text_header(_replacement_char_header())
         assert "\ufffd" not in result
-        result.replace("\n", "").encode("ascii")  # raises if any non-ASCII char survived
+        result.replace("\n", "").encode("ascii")
         assert any("repaired" in record.message for record in caplog.records)
 
     def test_repairs_short_layout(self, caplog: pytest.LogCaptureFixture) -> None:
-        """A header with fewer than 40 cards is padded out so export can proceed."""
+        """Header with fewer than 40 cards is padded out to 40 rows of 80 chars."""
         short = "\n".join(["C01".ljust(80)] * 5)
         with caplog.at_level(logging.WARNING, logger="mdio.segy.creation"):
             result = _ensure_exportable_text_header(short)
@@ -57,12 +58,7 @@ def test_repairs_short_layout(self, caplog: pytest.LogCaptureFixture) -> None:
         assert any("repaired" in record.message for record in caplog.records)
 
     def test_repaired_header_is_accepted_by_segy_factory(self) -> None:
-        """End-to-end proof that repair output is round-trippable via the SEG-Y factory.
-
-        Regression guard for issue #814: a malformed header that previously
-        crashed ``factory.create_textual_header`` must produce a 3200-byte
-        textual block after going through the export guard.
-        """
+        """Repair output round-trips through ``factory.create_textual_header`` to 3200 bytes."""
         spec = get_segy_standard(1.0)
         factory = SegyFactory(spec=spec, sample_interval=2000, samples_per_trace=1)
 
@@ -72,14 +68,7 @@ def test_repaired_header_is_accepted_by_segy_factory(self) -> None:
         assert len(encoded) == 3200
 
     def test_repairs_double_newline_wrapped(self, caplog: pytest.LogCaptureFixture) -> None:
-        """Legacy stores wrapped with ``\\n\\n`` per card must export with all 40 cards intact.
-
-        This is the second real-world malformed sample seen in the wild
-        (file ``260418_A4_…``): each card is terminated with ``\\n\\n``, which
-        previously caused naive splitting to lose cards 21-40 silently. The
-        export guard must collapse the double newlines and emit a 3200-byte
-        textual block whose 40 cards all carry their original ``Cnn`` prefix.
-        """
+        r"""Cards terminated with ``\n\n`` keep all 40 ``Cnn`` prefixes after repair."""
         cards = [f"C{i:02d}".ljust(80) for i in range(1, 41)]
         wrapped = "\n\n".join(cards) + "\n"