fix(datafile): fix mixed-text file behavior

autotest/test_binaryfile.py

@@ -3,6 +3,7 @@
 See also test_cellbudgetfile.py for similar tests.
 """
 
+import warnings
 from itertools import repeat
 
 import numpy as np
@@ -15,6 +16,7 @@
 import flopy
 from flopy.utils import (
     BinaryHeader,
+    BinaryLayerFile,
     CellBudgetFile,
     HeadFile,
     HeadUFile,
@@ -99,6 +101,8 @@ def test_headfile_build_index(example_data_path):
     assert hds.ncol == 20
     assert hds.nlay == 3
     assert not hasattr(hds, "nper")
+    assert hds.text == "head"
+    assert hds.text_bytes == b"HEAD".rjust(16)
     assert hds.totalbytes == 10_676_004
     assert len(hds.recordarray) == 3291
     assert type(hds.recordarray) == np.ndarray
@@ -145,7 +149,149 @@ def test_headfile_build_index(example_data_path):
     )
 
 
-def test_concentration_build_index(example_data_path):
+def test_headfile_examples(example_data_path):
+    # HeadFile with default text='head'
+    pth = example_data_path / "mf6-freyberg/freyberg.hds"
+    with HeadFile(pth) as obj:
+        assert obj.precision == "double"
+        assert (obj.nlay, obj.nrow, obj.ncol) == (1, 40, 20)
+        assert obj.text == "head"
+        assert obj.text_bytes == b"HEAD".ljust(16)
+        assert len(obj) == 1
+
+    # HeadFile with explicit text='drawdown' for a drawdown file
+    pth = example_data_path / "mfusg_test/03A_conduit_unconfined/output/ex3A.ddn"
+    with HeadFile(pth, text="drawdown") as obj:
+        assert obj.precision == "single"
+        assert (obj.nlay, obj.nrow, obj.ncol) == (2, 100, 100)
+        assert obj.text == "drawdown"
+        assert obj.text_bytes == b"DRAWDOWN".rjust(16)
+        assert len(obj) == 2
+
+    # HeadFile with default text='head' raises on non-head file
+    with pytest.raises(ValueError, match="no records with text='head'"):
+        HeadFile(pth)
+
+
+@pytest.mark.parametrize(
+    "pth, expected",
+    [
+        pytest.param(
+            "mf6/create_tests/test_transport/expected_output/gwt_mst03.ucn",
+            {
+                "precision": "double",
+                "nlay, nrow, ncol": (1, 1, 1),
+                "text": "concentration",
+                "text_bytes": b"CONCENTRATION".ljust(16),
+                "len(obj)": 28,
+            },
+            id="gwt_mst03.ucn",
+        ),
+        pytest.param(
+            "mfusg_test/03A_conduit_unconfined/output/ex3A.cln.hds",
+            {
+                "precision": "single",
+                "nlay, nrow, ncol": (1, 1, 2),
+                "text": "cln_heads",
+                "text_bytes": b"CLN HEADS".rjust(16),
+                "len(obj)": 1,
+            },
+            id="ex3A.cln.hds",
+        ),
+        pytest.param(
+            "mfusg_test/03A_conduit_unconfined/output/ex3A.ddn",
+            {
+                "precision": "single",
+                "nlay, nrow, ncol": (2, 100, 100),
+                "text": "drawdown",
+                "text_bytes": b"DRAWDOWN".rjust(16),
+                "len(obj)": 2,
+            },
+            id="ex3A.ddn",
+        ),
+    ],
+)
+def test_binarylayerfile_examples(example_data_path, pth, expected):
+    # BinaryLayerFile auto-detects text from file
+    with BinaryLayerFile(example_data_path / pth) as obj:
+        assert obj.precision == expected["precision"]
+        assert (obj.nlay, obj.nrow, obj.ncol) == expected["nlay, nrow, ncol"]
+        assert obj.text == expected["text"]
+        assert obj.text_bytes == expected["text_bytes"]
+        assert len(obj) == expected["len(obj)"]
+
+
+def _write_binary_layer_record(f, data, kstp=1, kper=1, totim=1.0, text="HEAD"):
+    """Write one single-precision binary layer record to open file f."""
+    nrow, ncol = data.shape
+    text_bytes = text.encode("ascii").ljust(16)[:16]
+    header = np.array(
+        [(kstp, kper, totim, totim, text_bytes, ncol, nrow, 1)],
+        dtype=[
+            ("kstp", "<i4"),
+            ("kper", "<i4"),
+            ("pertim", "<f4"),
+            ("totim", "<f4"),
+            ("text", "S16"),
+            ("ncol", "<i4"),
+            ("nrow", "<i4"),
+            ("ilay", "<i4"),
+        ],
+    )
+    header.tofile(f)
+    data.astype(np.float32).tofile(f)
+
+
+def test_binarylayerfile_mixed_text(tmp_path):
+    """BinaryLayerFile warns on multiple text types and scopes to first found."""
+    fname = tmp_path / "mixed.bin"
+    data = np.ones((3, 3), dtype=np.float32)
+    with open(fname, "wb") as f:
+        _write_binary_layer_record(
+            f, data, kstp=1, kper=1, totim=1.0, text="            HEAD"
+        )
+        _write_binary_layer_record(
+            f, data, kstp=1, kper=1, totim=1.0, text="        DRAWDOWN"
+        )
+
+    with pytest.warns(UserWarning, match="multiple record types"):
+        obj = BinaryLayerFile(fname)
+
+    assert obj.text == "head"
+    assert len(obj) == 1  # only HEAD record in recordarray
+    assert len(obj.headers) == 2  # both records in headers DataFrame
+    assert set(obj.unique_records) == {"DRAWDOWN", "HEAD"}
+
+    # re-open scoped to drawdown
+    with BinaryLayerFile(fname, text="drawdown") as obj2:
+        assert obj2.text == "drawdown"
+        assert len(obj2) == 1
+    obj.close()
+
+
+def test_binarylayerfile_wrong_text(tmp_path):
+    """BinaryLayerFile raises clearly when requested text is absent."""
+    fname = tmp_path / "head_only.bin"
+    data = np.ones((3, 3), dtype=np.float32)
+    with open(fname, "wb") as f:
+        _write_binary_layer_record(
+            f, data, kstp=1, kper=1, totim=1.0, text="            HEAD"
+        )
+
+    with pytest.raises(ValueError, match="no records with text='drawdown'"):
+        BinaryLayerFile(fname, text="drawdown")
+
+
+def test_unique_records(example_data_path):
+    """unique_records returns sorted array of text labels in the file."""
+    pth = example_data_path / "mf6-freyberg/freyberg.hds"
+    with HeadFile(pth) as obj:
+        ur = obj.unique_records
+        assert isinstance(ur, np.ndarray)
+        assert list(ur) == ["HEAD"]
+
+
+def test_ucnfile_build_index(example_data_path):
     # test low-level BinaryLayerFile._build_index() method with UCN file
     pth = example_data_path / "mt3d_test/mf2005mt3d/P07/MT3D001.UCN"
     with UcnFile(pth) as ucn:
@@ -154,6 +300,8 @@ def test_concentration_build_index(example_data_path):
     assert ucn.ncol == 21
     assert ucn.nlay == 8
     assert not hasattr(ucn, "nper")
+    assert ucn.text == "concentration"
+    assert ucn.text_bytes == b"CONCENTRATION".ljust(16)
     assert ucn.totalbytes == 10_432
     assert len(ucn.recordarray) == 8
     assert type(ucn.recordarray) == np.ndarray
@@ -286,6 +434,8 @@ def test_headu_file_data(function_tmpdir, example_data_path):
     headobj = HeadUFile(fname)
     assert isinstance(headobj, HeadUFile)
     assert headobj.nlay == 3
+    assert headobj.text == "headu"
+    assert headobj.text_bytes == b"HEADU".rjust(16)
 
     # ensure recordarray is has correct data
     ra = headobj.recordarray

autotest/test_formattedfile.py

@@ -21,6 +21,7 @@ def test_headfile_build_index(example_data_path):
     assert hds.ncol == 10
     assert hds.nlay == 1
     assert not hasattr(hds, "nper")
+    assert hds.text == "head"
     assert hds.totalbytes == 1613
     assert len(hds.recordarray) == 1
     assert type(hds.recordarray) == np.ndarray

flopy/export/utils.py

@@ -159,6 +159,9 @@ def _add_output_nc_variable(
     array = np.zeros((len(times), shape3d[0], shape3d[1], shape3d[2]), dtype=np.float32)
     array[:] = np.nan
 
+    if isinstance(text, bytes):
+        text = text.decode("ascii")
+
     if isinstance(out_obj, ZBNetOutput):
         a = np.asarray(out_obj.zone_array, dtype=np.float32)
         if mask_array3d is not None:
@@ -177,7 +180,7 @@ def _add_output_nc_variable(
                     else:
                         a = out_obj.get_data(totim=t)
                 except Exception as e:
-                    nme = var_name + text.decode().strip().lower()
+                    nme = var_name + text
                     estr = f"error getting data for {nme} at time {t}:{e!s}"
                     if logger:
                         logger.warn(estr)
@@ -189,7 +192,7 @@ def _add_output_nc_variable(
                 try:
                     array[i, :, :, :] = a.astype(np.float32)
                 except Exception as e:
-                    nme = var_name + text.decode().strip().lower()
+                    nme = var_name + text
                     estr = f"error assigning {nme} data to array for time {t}:{e!s}"
                     if logger:
                         logger.warn(estr)
@@ -207,7 +210,7 @@ def _add_output_nc_variable(
 
     if isinstance(nc, dict):
         if text:
-            var_name = text.decode().strip().lower()
+            var_name = text
         nc[var_name] = array
         return nc
 
@@ -217,7 +220,7 @@ def _add_output_nc_variable(
     precision_str = "f4"
 
     if text:
-        var_name = text.decode().strip().lower()
+        var_name = text
     attribs = {"long_name": var_name}
     attribs["coordinates"] = "time layer latitude longitude"
     attribs["min"] = mn
@@ -426,7 +429,7 @@ def output_helper(
                     times,
                     shape3d,
                     out_obj,
-                    "concentration",
+                    out_obj.text,
                     logger=logger,
                     mask_vals=mask_vals,
                     mask_array3d=mask_array3d,
@@ -438,7 +441,7 @@ def output_helper(
                     times,
                     shape3d,
                     out_obj,
-                    out_obj.text.decode(),
+                    out_obj.text,
                     logger=logger,
                     mask_vals=mask_vals,
                     mask_array3d=mask_array3d,

flopy/export/vtk.py

@@ -1197,14 +1197,12 @@ def add_heads(self, hds, kstpkper=None, masked_values=None):
         kstpkpers = hds.get_kstpkper()
         self._totim = dict(zip(kstpkpers, times))
 
-        text = hds.text.decode()
-
         d = {}
         for ki in kstpkper:
             d[ki] = hds.get_data(ki)
 
         self.__transient_output_data = False
-        self.add_transient_array(d, name=text, masked_values=masked_values)
+        self.add_transient_array(d, name=hds.text, masked_values=masked_values)
         self.__transient_output_data = True
 
     def add_cell_budget(self, cbc, text=None, kstpkper=None, masked_values=None):

flopy/mf6/utils/binaryfile_utils.py

@@ -192,7 +192,7 @@ def _get_binary_file_object(self, path, bintype, key):
 
         elif bintype == "DDN":
             try:
-                return bf.HeadFile(path, text="drawdown", precision="double")
+                return bf.HeadFile(path, precision="double")
             except AssertionError:
                 raise AssertionError(f"{self.dataDict[key]} does not exist")
 
@@ -333,9 +333,7 @@ def _setbinarykeys(self, binarypathdict):
 
             elif key[1] == "DDN":
                 try:
-                    readddn = bf.HeadFile(
-                        path, text="drawdown", precision="double"
-                    )
+                    readddn = bf.HeadFile(path, precision="double")
                     self.dataDict[(key[0], key[1], "DRAWDOWN")] = path
                     readddn.close()
 

flopy/utils/__init__.py

@@ -23,7 +23,14 @@
 
 from .utl_import import import_optional_dependency  # isort:skip
 from . import get_modflow as get_modflow_module
-from .binaryfile import BinaryHeader, CellBudgetFile, HeadFile, HeadUFile, UcnFile
+from .binaryfile import (
+    BinaryHeader,
+    BinaryLayerFile,
+    CellBudgetFile,
+    HeadFile,
+    HeadUFile,
+    UcnFile,
+)
 from .check import check
 from .flopy_io import read_fixed_var, write_fixed_var
 from .formattedfile import FormattedHeadFile