test(io): harden the ascii writer

.vscode/settings.json

@@ -13,6 +13,7 @@
         }
     },
     "mypy-type-checker.importStrategy": "fromEnvironment",
+    "python.analysis.typeCheckingMode": "off",
     "files.exclude": {
         "**/.git": true,
         "**/.svn": true,

flopy4/mf6/adapters.py

@@ -264,7 +264,6 @@ def plottable(self):
 
     @property
     def has_stress_period_data(self):
-        # TODO oc returns true? is stress package?
         return "nper" in self._data.dims
 
     def check(self, f=None, verbose=True, level=1, checktype=None):

flopy4/mf6/binding.py

@@ -0,0 +1,51 @@
+from attrs import define
+
+from flopy4.mf6.component import Component
+from flopy4.mf6.exchange import Exchange
+from flopy4.mf6.model import Model
+from flopy4.mf6.package import Package
+from flopy4.mf6.solution import Solution
+
+
+@define
+class Binding:
+    """
+    An MF6 component binding: a record representation of the
+    component for writing to a parent component's name file.
+    """
+
+    type: str
+    fname: str
+    terms: tuple[str, ...] | None = None
+
+    def to_tuple(self) -> tuple[str, ...]:
+        if self.terms and any(self.terms):
+            return (self.type, self.fname, *self.terms)
+        else:
+            return (self.type, self.fname)
+
+    @classmethod
+    def from_component(cls, component: Component) -> "Binding":
+        def _get_binding_type(component: Component) -> str:
+            cls_name = component.__class__.__name__
+            if isinstance(component, Exchange):
+                return f"{'-'.join([cls_name[:2], cls_name[3:]]).upper()}6"
+            elif isinstance(component, Solution):
+                return f"{component.slntype}6"
+            else:
+                return f"{cls_name.upper()}6"
+
+        def _get_binding_terms(component: Component) -> tuple[str, ...] | None:
+            if isinstance(component, Exchange):
+                return (component.exgmnamea, component.exgmnameb)  # type: ignore
+            elif isinstance(component, Solution):
+                return tuple(component.models)
+            elif isinstance(component, (Model, Package)):
+                return (component.name,)  # type: ignore
+            return None
+
+        return cls(
+            type=_get_binding_type(component),
+            fname=component.filename or component.default_filename(),
+            terms=_get_binding_terms(component),
+        )

flopy4/mf6/codec/writer/__init__.py

@@ -13,8 +13,9 @@
 )
 _JINJA_ENV.filters["field_type"] = filters.field_type
 _JINJA_ENV.filters["array_how"] = filters.array_how
-_JINJA_ENV.filters["array_chunks"] = filters.array_chunks
+_JINJA_ENV.filters["array2chunks"] = filters.array2chunks
 _JINJA_ENV.filters["array2string"] = filters.array2string
+_JINJA_ENV.filters["array2const"] = filters.array2const
 _JINJA_ENV.filters["data2list"] = filters.data2list
 _JINJA_TEMPLATE_NAME = "blocks.jinja"
 _PRINT_OPTIONS = {

flopy4/mf6/codec/writer/filters.py

@@ -6,6 +6,7 @@
 import xarray as xr
 from modflow_devtools.dfn.schema.v2 import FieldType
 from numpy.typing import NDArray
+from xattree import Scalar
 
 from flopy4.mf6.constants import FILL_DNODATA
 
@@ -33,13 +34,13 @@ def field_type(value: Any) -> FieldType:
 
 
 def array_how(value: xr.DataArray) -> str:
-    # TODO
-    # - detect constant arrays?
-    # - above certain size, use external?
+    # TODO above certain size, use external?
+    if value.max() == value.min():
+        return "constant"
     return "internal"
 
 
-def array_chunks(value: xr.DataArray, chunks: Mapping[Hashable, int] | None = None):
+def array2chunks(value: xr.DataArray, chunks: Mapping[Hashable, int] | None = None):
     """
     Yield chunks from a dask-backed array of up to 3 dimensions.
     If it's not already chunked, split it into chunks of the
@@ -128,18 +129,28 @@ def nonempty(value: NDArray | xr.DataArray) -> NDArray:
     return mask
 
 
-def data2list(value: list | tuple | dict | xr.Dataset | xr.DataArray):
+def array2const(value: xr.DataArray) -> Scalar:
+    if np.issubdtype(value.dtype, np.integer):
+        return value.max().item()
+    if np.issubdtype(value.dtype, np.floating):
+        return f"{value.max().item():.8f}"
+    return value.ravel()[0]
+
+
+def data2list(value: list | dict | xr.Dataset | xr.DataArray):
     """
     Yield records (tuples) from data in a `list`, `dict`, `DataArray` or `Dataset`.
     """
 
-    if isinstance(value, (list, tuple)):
+    if isinstance(value, list):
         for rec in value:
+            if not isinstance(rec, tuple):
+                raise ValueError(f"Unsupported record type: {type(rec)}")
             yield rec
         return
 
     if isinstance(value, dict):
-        for name, val in value.values():
+        for name, val in value.items():
             yield (name, val)
         return
 

flopy4/mf6/codec/writer/templates/macros.jinja

@@ -32,14 +32,14 @@
 {{ inset ~ name.upper() }}{% if "layered" in how %} LAYERED{% endif %}
 
 {% if how == "constant" %}
-CONSTANT {{ value.item() }}
+{{ inset }}CONSTANT {{ value|array2const -}}
 {% elif how == "layered constant" %}
 {% for layer in value -%}
-CONSTANT {{ layer.item() }}
+{{ inset }}CONSTANT {{ layer|array2const -}}
 {%- endfor %}
 {% elif how == "internal" %}
 INTERNAL
-{% for chunk in value|array_chunks -%}
+{% for chunk in value|array2chunks -%}
 {{ (2 * inset) ~ chunk|array2string }}
 {%- endfor %}
 {% elif how == "external" %}

flopy4/mf6/component.py

@@ -1,12 +1,13 @@
 from abc import ABC
 from collections.abc import MutableMapping
 from pathlib import Path
-from typing import ClassVar
+from typing import Any, ClassVar
 
 import numpy as np
 from attrs import fields
 from modflow_devtools.dfn import Dfn, Field
 from packaging.version import Version
+from xattree import asdict as xattree_asdict
 from xattree import xattree
 
 from flopy4.mf6.constants import FILL_DNODATA, MF6
@@ -197,3 +198,19 @@ def write(self, format: str = MF6) -> None:
         self._write(format=format)
         for child in self.children.values():  # type: ignore
             child.write(format=format)
+
+    def to_dict(self, blocks: bool = False) -> dict[str, Any]:
+        """Convert the component to a dictionary representation."""
+        data = xattree_asdict(self)
+        data.pop("filename")
+        data.pop("workspace", None)
+        data.pop("nodes", None)  # TODO: find a better way to omit
+        if blocks:
+            blocks_ = {}  # type: ignore
+            for field_name, field_value in data.items():
+                block_name = self.dfn.fields[field_name].block
+                if block_name not in blocks_:
+                    blocks_[block_name] = {}
+                blocks_[block_name][field_name] = field_value
+            return blocks_
+        return data

flopy4/mf6/config.py

@@ -1,3 +1,3 @@
 # TODO use https://environ-config.readthedocs.io/en/stable/?
 
-SPARSE_THRESHOLD = 1000
+SPARSE_THRESHOLD = 100000

flopy4/mf6/constants.py

@@ -1,5 +1,6 @@
 import numpy as np
 
 MF6 = "mf6"
+PERIOD = "period"
 FILL_DEFAULT = np.nan
 FILL_DNODATA = 1e30