Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 36 additions & 1 deletion autotest/test_dfn.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from modflow_devtools.dfn.fetch import fetch_dfns
from modflow_devtools.dfn.schema.v1 import FieldV1
from modflow_devtools.dfn.schema.v2 import FieldV2
from modflow_devtools.dfn2toml import convert
from modflow_devtools.dfn2toml import convert, validate
from modflow_devtools.markers import requires_pkg

PROJ_ROOT = Path(__file__).parents[1]
Expand Down Expand Up @@ -348,3 +348,38 @@ def test_dfn_from_dict_with_already_deserialized_fields():
dfn = Dfn.from_dict(d)
assert dfn.blocks is not None
assert dfn.blocks["options"]["test"] is field


@requires_pkg("boltons")
def test_validate_directory():
"""Test validation on a directory of DFN files."""
assert validate(DFN_DIR) is True


@requires_pkg("boltons")
def test_validate_single_file(dfn_name):
"""Test validation on a single DFN file."""
if dfn_name == "common":
pytest.skip("common.dfn is handled separately")
assert validate(DFN_DIR / f"{dfn_name}.dfn") is True


@requires_pkg("boltons")
def test_validate_common_file():
"""Test validation on common.dfn."""
assert validate(DFN_DIR / "common.dfn") is True


@requires_pkg("boltons")
def test_validate_invalid_file(function_tmpdir):
"""Test validation on an invalid DFN file."""
invalid_dfn = function_tmpdir / "invalid.dfn"
invalid_dfn.write_text("invalid content")
assert validate(invalid_dfn) is False


@requires_pkg("boltons")
def test_validate_nonexistent_file(function_tmpdir):
"""Test validation on a nonexistent file."""
nonexistent = function_tmpdir / "nonexistent.dfn"
assert validate(nonexistent) is False
6 changes: 5 additions & 1 deletion docs/md/dfn.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,12 @@ Where legacy DFNs are flat lists of variables, with comments demarcating blocks,

The `dfn` dependency group is necessary to use the TOML conversion utility.

To convert definition files to TOML, use:
To convert legacy format definition files to TOML, use:

```shell
python -m modflow_devtools.dfn.dfn2toml -i <dfn dir path> -o <output dir path>
```

The tool may also be used on individual files.

To validate legacy format definition files, use the `--validate` flag.
132 changes: 98 additions & 34 deletions modflow_devtools/dfn2toml.py
Original file line number Diff line number Diff line change
@@ -1,54 +1,107 @@
"""Convert DFNs to TOML."""

import argparse
import sys
from dataclasses import asdict
from os import PathLike
from pathlib import Path

import tomli_w as tomli
from boltons.iterutils import remap

from modflow_devtools.dfn import load_flat, map, to_flat, to_tree
from modflow_devtools.dfn import Dfn, load, load_flat, map, parse_dfn, to_flat, to_tree
from modflow_devtools.dfn.schema.block import block_sort_key
from modflow_devtools.misc import drop_none_or_empty

# mypy: ignore-errors


def convert(indir: PathLike, outdir: PathLike, schema_version: str = "2") -> None:
indir = Path(indir).expanduser().absolute()
def validate(path: str | PathLike) -> bool:
"""Validate DFN file(s) by attempting to parse them."""
path = Path(path).expanduser().absolute()
try:
if not path.exists():
raise FileNotFoundError(f"Path does not exist: {path}")

if path.is_file():
if path.name == "common.dfn":
with path.open() as f:
parse_dfn(f)
else:
common_path = path.parent / "common.dfn"
if common_path.exists():
with common_path.open() as f:
common, _ = parse_dfn(f)
else:
common = {}
with path.open() as f:
load(f, name=path.stem, common=common, format="dfn")
else:
load_flat(path)
return True
except Exception as e:
print(f"Validation failed: {e}")
return False


def convert(inpath: PathLike, outdir: PathLike, schema_version: str = "2") -> None:
inpath = Path(inpath).expanduser().absolute()
outdir = Path(outdir).expanduser().absolute()
outdir.mkdir(exist_ok=True, parents=True)

dfns = {
name: map(dfn, schema_version=schema_version)
for name, dfn in load_flat(indir).items()
}
tree = to_tree(dfns)
flat = to_flat(tree)
for dfn_name, dfn in flat.items():
with Path.open(outdir / f"{dfn_name}.toml", "wb") as f:
# TODO if we start using c/attrs, swap out
# all this for a custom unstructuring hook
dfn_dict = asdict(dfn)
dfn_dict["schema_version"] = str(dfn_dict["schema_version"])
if dfn_dict.get("blocks"):
blocks = dfn_dict.pop("blocks")
for block_name, block_fields in blocks.items():
if block_name not in dfn_dict:
dfn_dict[block_name] = {}
for field_name, field_data in block_fields.items():
dfn_dict[block_name][field_name] = field_data

tomli.dump(
dict(
sorted(
remap(dfn_dict, visit=drop_none_or_empty).items(),
key=block_sort_key,
)
),
f,
)
if inpath.is_file():
if inpath.name == "common.dfn":
raise ValueError("Cannot convert common.dfn as a standalone file")

common_path = inpath.parent / "common.dfn"
if common_path.exists():
with common_path.open() as f:
from modflow_devtools.dfn import parse_dfn

common, _ = parse_dfn(f)
else:
common = {}

with inpath.open() as f:
dfn = load(f, name=inpath.stem, common=common, format="dfn")

dfn = map(dfn, schema_version=schema_version)
_convert(outdir / f"{inpath.stem}.toml", dfn)
else:
dfns = {
name: map(dfn, schema_version=schema_version)
for name, dfn in load_flat(inpath).items()
}
tree = to_tree(dfns)
flat = to_flat(tree)
for dfn_name, dfn in flat.items():
_convert(outdir / f"{dfn_name}.toml", dfn)


def _convert(outpath: Path, dfn: Dfn) -> None:
"""Write a DFN object to a TOML file."""
with Path.open(outpath, "wb") as f:
# TODO if we start using c/attrs, swap out
# all this for a custom unstructuring hook
dfn_dict = asdict(dfn)
dfn_dict["schema_version"] = str(dfn_dict["schema_version"])
if dfn_dict.get("blocks"):
blocks = dfn_dict.pop("blocks")
for block_name, block_fields in blocks.items():
if block_name not in dfn_dict:
dfn_dict[block_name] = {}
for field_name, field_data in block_fields.items():
dfn_dict[block_name][field_name] = field_data

tomli.dump(
dict(
sorted(
remap(dfn_dict, visit=drop_none_or_empty).items(),
key=block_sort_key,
)
),
f,
)


if __name__ == "__main__":
Expand All @@ -62,7 +115,7 @@ def convert(indir: PathLike, outdir: PathLike, schema_version: str = "2") -> Non
"--indir",
"-i",
type=str,
help="Directory containing DFN files.",
help="Directory containing DFN files, or a single DFN file.",
)
parser.add_argument(
"--outdir",
Expand All @@ -76,5 +129,16 @@ def convert(indir: PathLike, outdir: PathLike, schema_version: str = "2") -> Non
default="2",
help="Schema version to convert to.",
)
parser.add_argument(
"--validate",
"-v",
action="store_true",
help="Validate DFN files without converting them.",
)
args = parser.parse_args()
convert(args.indir, args.outdir, args.schema_version)

if args.validate:
if not validate(args.indir):
sys.exit(1)
else:
convert(args.indir, args.outdir, args.schema_version)