mf6 synchronization

flopy4/cli.py

@@ -0,0 +1,241 @@
+"""Top-level CLI entry point for flopy4."""
+
+import argparse
+import logging
+import shutil
+import sys
+import warnings
+from pathlib import Path
+
+_PROJ_ROOT = Path(__file__).parents[1].expanduser().resolve()
+_MF6_ROOT = _PROJ_ROOT / "flopy4" / "mf6"
+_DFN_SCHEMA_VERSION = "2.0.0.dev1"
+
+
+def _resolve_release_id(release_id: str | None, verbose: bool = False) -> str:
+    """Resolve a release_id, falling back to the discovered binary's version
+    or the latest release if no release_id is given.
+    """
+    if release_id is not None:
+        return release_id
+
+    # Try to read the version from a binary on PATH.
+    from flopy4.mf6._compat import _query_mf6_version
+
+    for name in ("mf6", "mf6.exe"):
+        exe = shutil.which(name)
+        if exe:
+            version = _query_mf6_version(exe)
+            if version:
+                if verbose:
+                    print(f"Detected MF6 {version} at {exe}")
+                return f"MODFLOW-ORG/modflow6@{version}"
+
+    # Fall back to @latest.
+    if verbose:
+        print("No MF6 binary found on PATH; resolving to latest release.")
+    return "MODFLOW-ORG/modflow6@latest"
+
+
+def _cmd_sync(args: argparse.Namespace) -> None:
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", message=".*modflow_devtools.programs.*experimental.*")
+        warnings.filterwarnings("ignore", message=".*modflow_devtools.dfns.*experimental.*")
+        from modflow_devtools.dfns import LocalDfnRegistry, RemoteDfnRegistry
+        # from modflow_devtools.programs import install_program
+
+    from flopy4.mf6.utils.codegen.dfn2py import make
+
+    release_id = _resolve_release_id(args.release_id, verbose=args.verbose)
+
+    # For remote release IDs, install the matching binary unless --no-install.
+    from pathlib import Path
+
+    # is_local = Path(release_id).expanduser().is_dir()
+    # if not is_local and not args.no_install:
+    #     _, tag = release_id.split("@", 1)
+    #     if tag not in ("latest",):
+    #         version_arg = tag
+    #     else:
+    #         version_arg = None  # let install_program resolve latest
+    #     if args.verbose:
+    #         label = version_arg or "latest"
+    #         print(f"Installing mf6 {label} ...")
+    #     try:
+    #         installed = install_program(
+    #             "mf6",
+    #             version=version_arg,
+    #             force=args.force,
+    #             verbose=args.verbose,
+    #         )
+    #         if args.verbose and installed:
+    #             print(f"Installed: {', '.join(str(p) for p in installed)}")
+    #     except Exception as exc:
+    #         print(f"Warning: binary installation failed: {exc}", file=sys.stderr)
+    #         print("Continuing with class generation only.", file=sys.stderr)
+
+    if args.verbose:
+        print(f"Generating flopy4.mf6 from {release_id} ...")
+
+    def _populate_remote_cache(
+        registry: RemoteDfnRegistry, release_id: str, force: bool = False
+    ) -> None:
+        """Populate the registry's DFN cache.
+
+        Tries ``registry.sync()`` first (requires a ``dfns.zip`` release asset).
+        Falls back to fetching DFNs from the repository tree when that asset is
+        not yet available — writes directly into the registry's cache path so
+        ``registry.spec()`` finds the files in the expected location.
+
+        Parameters
+        ----------
+        force :
+            Re-fetch even if the cache already exists. Should be True for
+            branch refs (mutable) and False for pinned release tags.
+        """
+        if not force and registry.cache_path.exists() and any(registry.cache_path.iterdir()):
+            return
+        try:
+            registry.sync()
+        except Exception:
+            # dfns.zip not yet a release asset; fetch from the git tree instead.
+            from modflow_devtools.dfn import get_dfns
+
+            owner_repo, tag = release_id.split("@", 1)
+            owner, repo = owner_repo.split("/")
+            registry.cache_path.mkdir(parents=True, exist_ok=True)
+            get_dfns(owner=owner, repo=repo, ref=tag, outdir=registry.cache_path)
+
+    def _write_contract(outdir: Path, mf6_version: str) -> None:
+        (outdir / "_contract.py").write_text(
+            "# autogenerated file, do not modify\n"
+            f'MF6_VERSION = "{mf6_version}"\n'
+            f'DFN_SCHEMA_VERSION = "{_DFN_SCHEMA_VERSION}"\n'
+        )
+
+    path = Path(release_id).expanduser()
+    if path.exists() and path.is_dir():
+        registry = LocalDfnRegistry(path=path)
+        effective_version = "unknown"
+    else:
+        if "@" not in release_id or "/" not in release_id.split("@")[0]:
+            raise ValueError(
+                f"release_id must be 'owner/repo@tag' or a local path; got: {release_id!r}"
+            )
+        registry = RemoteDfnRegistry(release_id=release_id)
+        effective_version = args.mf6_version or registry.latest_tag() or "unknown"
+
+    if effective_version == "unknown":
+        warnings.warn(
+            f"MF6 version is unknown for local DFN path '{release_id}'. "
+            "Pass --mf6-version to record it explicitly in _contract.py."
+        )
+
+    if isinstance(registry, RemoteDfnRegistry):
+        # Branch refs are mutable: always re-fetch unless the caller explicitly
+        # passes a pinned semver tag (first char is a digit).
+        is_branch = not effective_version[0].isdigit() if effective_version else True
+        _populate_remote_cache(registry, release_id, force=args.force or is_branch)
+
+    make(
+        dfndir=registry.cache_path if isinstance(registry, RemoteDfnRegistry) else registry.path,
+        outdir=_MF6_ROOT,
+        existing_only=not args.all_packages,
+        makedirs=args.all_packages,
+        fmt=not args.no_format,
+    )
+    _write_contract(_MF6_ROOT, effective_version)
+
+    if effective_version != "unknown":
+        print(f"Synced flopy4.mf6 to MF6 version: {effective_version}")
+
+
+def _cmd_status(args: argparse.Namespace) -> None:
+    try:
+        from flopy4.mf6._contract import DFN_SCHEMA_VERSION, MF6_VERSION
+    except ImportError:
+        MF6_VERSION = "unknown"
+        DFN_SCHEMA_VERSION = "unknown"
+
+    from flopy4.mf6._compat import _query_mf6_version
+
+    exe = shutil.which("mf6") or shutil.which("mf6.exe")
+    binary_version = _query_mf6_version(exe) if exe else None
+
+    synced = binary_version is not None and binary_version == MF6_VERSION
+    status = "(✓ in sync)" if synced else "(! mismatch)" if binary_version else "(not found)"
+
+    print(f"flopy4.mf6 synced to : {MF6_VERSION}")
+    print(f"DFN schema version   : {DFN_SCHEMA_VERSION}")
+    if exe:
+        print(f"Discovered binary    : {binary_version}  [{exe}]  {status}")
+    else:
+        print("Discovered binary    : none  (not found on PATH)")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="flopy4")
+    sub = parser.add_subparsers(dest="command")
+
+    mf6_p = sub.add_parser("mf6", help="MF6 sync and status commands.")
+    mf6_sub = mf6_p.add_subparsers(dest="subcommand")
+
+    # flopy4 mf6 sync
+    sync_p = mf6_sub.add_parser(
+        "sync",
+        help="Sync flopy4.mf6 to an MF6 release: install binary and regenerate classes.",
+    )
+    sync_p.add_argument(
+        "release_id",
+        nargs="?",
+        default=None,
+        help="Remote release ID (owner/repo@tag) or local path to .dfn files. "
+        "Defaults to the discovered binary's version, or latest if no binary found.",
+    )
+    sync_p.add_argument(
+        "--mf6-version",
+        default=None,
+        dest="mf6_version",
+        help="Override MF6 version in _contract.py (useful with local DFN paths).",
+    )
+    sync_p.add_argument(
+        "--no-install",
+        action="store_true",
+        help="Skip binary installation; regenerate classes only.",
+    )
+    sync_p.add_argument(
+        "--all-packages",
+        action="store_true",
+        dest="all_packages",
+        help="Generate all packages, including ones not yet on disk. "
+        "By default sync only updates already-generated files.",
+    )
+    sync_p.add_argument(
+        "--no-format",
+        action="store_true",
+        help="Skip ruff formatting of generated files.",
+    )
+    sync_p.add_argument("--force", action="store_true", help="Force binary reinstallation.")
+    sync_p.add_argument("--verbose", action="store_true")
+    sync_p.set_defaults(func=_cmd_sync)
+
+    # flopy4 mf6 status
+    status_p = mf6_sub.add_parser(
+        "status",
+        help="Show current sync state: synced version vs. discovered binary.",
+    )
+    status_p.set_defaults(func=_cmd_status)
+
+    args = parser.parse_args()
+    if not hasattr(args, "func"):
+        parser.print_help()
+        sys.exit(1)
+
+    if getattr(args, "verbose", False):
+        logging.basicConfig(level=logging.INFO)
+
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

flopy4/mf6/__init__.py

@@ -5,8 +5,15 @@
 from tomli import load as load_toml
 from tomli_w import dump as dump_toml
 
+try:
+    from flopy4.mf6._contract import DFN_SCHEMA_VERSION, MF6_VERSION
+except ImportError:
+    DFN_SCHEMA_VERSION = "unknown"
+    MF6_VERSION = "unknown"
+
 # Import submodules to make them accessible via flopy4.mf6.*
 from flopy4.mf6 import gwe, gwf, gwt, prt, simulation, solution, utils
+from flopy4.mf6._compat import check_mf6_compatibility
 from flopy4.mf6.codec import dump as dump_mf6
 from flopy4.mf6.codec import load as load_mf6
 from flopy4.mf6.component import Component
@@ -36,6 +43,8 @@
     "NetCDFModel",
     "Tdis",
     "Simulation",
+    "MF6_VERSION",
+    "DFN_SCHEMA_VERSION",
 ]
 
 
@@ -98,3 +107,5 @@ def _write_toml(component: Component) -> None:
 DEFAULT_REGISTRY.register_writer(Component, "mf6", _write_mf6)
 DEFAULT_REGISTRY.register_writer(Component, "json", _write_json)
 DEFAULT_REGISTRY.register_writer(Component, "toml", _write_toml)
+
+check_mf6_compatibility()

flopy4/mf6/_compat.py

@@ -0,0 +1,53 @@
+import re
+import shutil
+import subprocess
+import warnings
+
+_VERSION_RE = re.compile(r"version\s+([\d]+\.[\d]+\.[\d]+(?:\.\S+)?)", re.I)
+
+
+def _query_mf6_version(exe: str) -> str | None:
+    try:
+        out = subprocess.check_output([exe, "-v"], text=True, stderr=subprocess.STDOUT)
+        m = _VERSION_RE.search(out)
+        return m.group(1) if m else None
+    except Exception:
+        return None
+
+
+def check_mf6_compatibility(exe: str | None = None) -> None:
+    """Warn if a discovered MF6 binary doesn't match the synced version.
+
+    Does nothing when the synced version is ``"unknown"`` or a branch
+    name (non-semver).
+
+    Parameters
+    ----------
+    exe :
+        Path to an MF6 executable. If None, searches PATH for ``mf6``
+        or ``mf6.exe``. Does nothing if no binary is found.
+    """
+    try:
+        from flopy4.mf6._contract import MF6_VERSION
+    except ImportError:
+        return
+
+    # Skip if version is unknown or a branch name rather than a semver tag.
+    if not MF6_VERSION or not MF6_VERSION[0].isdigit():
+        return
+
+    if exe is None:
+        exe = shutil.which("mf6") or shutil.which("mf6.exe")
+    if exe is None:
+        return
+
+    binary_version = _query_mf6_version(exe)
+    if binary_version is None or binary_version == MF6_VERSION:
+        return
+
+    warnings.warn(
+        f"flopy4.mf6 is synced to MF6 {MF6_VERSION} but the binary at '{exe}' "
+        f"reports {binary_version}. Run `flopy4 mf6 sync` to re-sync.",
+        UserWarning,
+        stacklevel=3,
+    )

flopy4/mf6/_contract.py

@@ -0,0 +1,3 @@
+# autogenerated file, do not modify
+MF6_VERSION = "unknown"
+DFN_SCHEMA_VERSION = "2.0.0.dev1"

flopy4/mf6/codec/reader/dfn2lark.py

@@ -1,10 +1,10 @@
-"""Convert (TOML/v2) DFNs to Lark grammars."""
+"""Generate Lark grammars from DFN files."""
 
 import argparse
 from os import PathLike
 from pathlib import Path
 
-from modflow_devtools.dfn import load_flat, map
+import modflow_devtools.dfn as dfn
 
 from flopy4.mf6.codec.reader.grammar import make_grammars
 
@@ -17,9 +17,8 @@ def make(dfndir: str | PathLike, outdir: str | PathLike):
     dfndir = Path(dfndir).expanduser().absolute()
     outdir = Path(outdir).expanduser().absolute()
     outdir.mkdir(exist_ok=True, parents=True)
-    dfns_v1 = load_flat(dfndir)
-    dfns_v2 = {name: map(dfn, schema_version=2) for name, dfn in dfns_v1.items()}
-    make_grammars(dfns_v2, outdir)
+    dfns = dfn.Dfn.load_all(dfndir, schema_version="2.0.0.dev1")
+    make_grammars(dfns, outdir)
 
 
 def main():

flopy4/mf6/utils/codegen/__init__.py

@@ -1,3 +1,3 @@
-from .generate_classes import generate_classes
+from flopy4.mf6.utils.codegen.dfn2py import make
 
-__all__ = ["generate_classes"]
+__all__ = ["make"]