diff --git a/docs/archive-format.md b/docs/archive-format.md
index dec86f3..48e45cf 100644
--- a/docs/archive-format.md
+++ b/docs/archive-format.md
@@ -403,14 +403,22 @@ That is the front-end's job.
```bash
ymf2xmf run.ymf # writes run.xmf beside it
ymf2xmf -v run.ymf -o view.xmf # -v summarizes what the archive holds
+ymf2xmf --show run.ymf # summarizes and converts nothing
+ymf2xmf run.ymf -c c0p2 -f u -s -1 -o last.xmf # one space, one field, the last step
+ymf2xmf -i run.ymf # asks which collections, fields and steps
```
```text
run.ymf
- Mesh Spatial_Domain: 6 steps, t=0..50, Triangle topology, fields: T, q, side
+ 1. Mesh Spatial_Domain: 6 steps, t=0..50, Triangle topology, fields: T, q, side
wrote view.xmf
```
+`-c/--collection` (a name or part of one), `-f/--field` and `-s/--steps`
+(Python indices and slices: `-1`, `0:10`, `::5`) cherry-pick what the `.xmf`
+describes; the `.ymf` is never changed. In code the same selection is
+`ymf.archive.select_domain(domain, collections=, fields=, steps=)`.
+
ParaView and VisIt open the `.xmf` directly. `extra` is carried into it as
a base64-encoded JSON `` element, a
child of ``, which the XDMF DTD permits
diff --git a/src/ymf/archive.py b/src/ymf/archive.py
index 8efeb40..d80dbab 100644
--- a/src/ymf/archive.py
+++ b/src/ymf/archive.py
@@ -158,6 +158,8 @@
"inline_domain",
"domain_arrays",
"rebase_domain",
+ "select_domain",
+ "step_indices",
"read_ymf",
"dump_grid",
"load_grid",
@@ -1240,6 +1242,118 @@ def _data_items(domain: Dict[str, Any]):
yield "%s/%s/%s" % (where, a.get("center", "Node"), a["name"]), a
+def step_indices(spec: Any, n: int) -> List[int]:
+ """The step indices ``spec`` picks out of ``n`` steps, in order.
+
+ ``spec`` is a comma-separated list of Python indices and slices --
+ ``"-1"`` the last step, ``"0,5,9"``, ``"2:8"``, ``"::10"`` every tenth,
+ ``"-3:"`` the last three -- or a sequence of ints. Negative indices
+ count from the end as in Python; an index outside ``range(n)`` is
+ refused rather than silently dropped, since it names a step the
+ archive does not have.
+ """
+ if isinstance(spec, str):
+ parts = [p.strip() for p in spec.split(",") if p.strip()]
+ else:
+ parts = [str(int(i)) for i in spec]
+ chosen: List[int] = []
+ for part in parts:
+ if ":" in part:
+ try:
+ bounds = [int(b) if b.strip() else None for b in part.split(":")]
+ except ValueError:
+ raise YmfArchiveError("step selection %r is not a slice" % (part,)) from None
+ if len(bounds) > 3:
+ raise YmfArchiveError("step selection %r has too many colons" % (part,))
+ indices = list(range(n))[slice(*bounds)]
+ else:
+ try:
+ index = int(part)
+ except ValueError:
+ raise YmfArchiveError(
+ "step selection %r is neither an index nor a slice" % (part,)) from None
+ if not -n <= index < n:
+ raise YmfArchiveError(
+ "step %d does not exist; the archive holds %d step(s)" % (index, n))
+ indices = [index % n]
+ for i in indices:
+ if i not in chosen:
+ chosen.append(i)
+ return chosen
+
+
+def select_domain(
+ domain: Dict[str, Any],
+ *,
+ collections: Optional[Sequence[str]] = None,
+ fields: Optional[Sequence[str]] = None,
+ steps: Any = None,
+) -> Dict[str, Any]:
+ """A copy of ``domain`` holding only the chosen collections, fields and steps.
+
+ An archive of a long run with several spaces and a dozen fields is more
+ than a viewer wants to open at once, and more than anyone wants to copy;
+ this is the cherry-picking behind ``ymf2xmf --collection/--field/--steps``.
+
+ ``collections`` names the time collections to keep, each either a
+ collection's exact name or a substring of one (``"c0p2"`` for
+ ``Mesh_c0p2_Lagrange``). ``fields`` names the attributes to keep on
+ every grid; ``steps`` is a :func:`step_indices` selection applied to
+ every collection. ``None`` keeps everything. A name that matches
+ nothing is refused, with what the archive does hold, rather than
+ yielding an emptier archive than was asked for.
+ """
+ import copy
+
+ all_collections = domain.get("time_collections", [])
+ names = [c.get("name", "") for c in all_collections]
+ if collections is not None:
+ keep = []
+ for want in collections:
+ matched = [c for c in all_collections
+ if c.get("name", "") == want or want in c.get("name", "")]
+ if not matched:
+ raise YmfArchiveError(
+ "no collection matches %r; the archive holds %s" % (want, names))
+ for c in matched:
+ if c not in keep:
+ keep.append(c)
+ all_collections = keep
+
+ def grids_of(step: Dict[str, Any]):
+ if "spatial_collection" in step:
+ return step["spatial_collection"]
+ return [step]
+
+ if fields is not None:
+ present = set()
+ for c in all_collections:
+ for step in c.get("data", []):
+ for g in grids_of(step):
+ present.update(a["name"] for a in g.get("attributes", []))
+ missing = [f for f in fields if f not in present]
+ if missing:
+ raise YmfArchiveError(
+ "no field named %s; the archive holds %s"
+ % (", ".join(repr(f) for f in missing), sorted(present)))
+
+ chosen = []
+ for c in all_collections:
+ data = c.get("data", [])
+ indices = (step_indices(steps, len(data)) if steps is not None
+ else list(range(len(data))))
+ new_steps = []
+ for i in indices:
+ step = copy.deepcopy(data[i])
+ if fields is not None:
+ for g in grids_of(step):
+ g["attributes"] = [a for a in g.get("attributes", [])
+ if a["name"] in fields]
+ new_steps.append(step)
+ chosen.append(dict(c, data=new_steps))
+ return dict(domain, time_collections=chosen)
+
+
def inline_domain(domain: Dict[str, Any], base_dir: str | Path = ".") -> Dict[str, Any]:
"""A copy of ``domain`` with every referenced array copied in as ``Values``.
diff --git a/src/ymf/cli.py b/src/ymf/cli.py
index 15fabc5..bec8de7 100644
--- a/src/ymf/cli.py
+++ b/src/ymf/cli.py
@@ -21,10 +21,11 @@
from pathlib import Path
from typing import List, Optional, Sequence
-from ymf.archive import YmfArchiveError, canonicalize_domain, read_ymf, validate_domain
+from ymf.archive import (YmfArchiveError, canonicalize_domain, read_ymf, select_domain,
+ validate_domain)
from ymf.xdmf import write_xdmf
-__all__ = ["archive2xmf", "ymf2xmf", "ymf2xmf_main"]
+__all__ = ["archive2xmf", "describe", "ymf2xmf", "ymf2xmf_main"]
def _default_output(source: Path) -> Path:
@@ -37,6 +38,9 @@ def ymf2xmf(
output: Optional[str | Path] = None,
*,
validate: bool = True,
+ collections: Optional[Sequence[str]] = None,
+ fields: Optional[Sequence[str]] = None,
+ steps: Optional[str] = None,
) -> Path:
"""Write the ``.xmf`` for a ``.ymf`` archive. Returns the output path.
@@ -50,6 +54,11 @@ def ymf2xmf(
Check the archive's structure before converting. On by default: a
malformed archive is better reported here, with the offending
field named, than as a file a viewer silently mis-draws.
+ collections, fields, steps:
+ Convert only these: see :func:`ymf.archive.select_domain`. The
+ ``.xmf`` then describes a part of the archive, which is what a
+ viewer opening one space, one field or one instant of a long run
+ wants; the ``.ymf`` is untouched.
The archive's ``extra`` payload -- ``Problem``, ``solution_paths``,
``vvuq`` and anything else with no XDMF equivalent -- is carried into
@@ -64,6 +73,8 @@ def ymf2xmf(
raise YmfArchiveError(
"%s holds no domain, so there is nothing to convert" % (source,))
domain = validate_domain(domain) if validate else canonicalize_domain(domain)
+ if collections is not None or fields is not None or steps is not None:
+ domain = select_domain(domain, collections=collections, fields=fields, steps=steps)
destination = Path(output) if output is not None else _default_output(source)
write_xdmf(domain, destination, ymf_extra=extra)
return destination
@@ -121,10 +132,25 @@ def _is_archive(source: Path) -> bool:
return False
-def _describe(domain) -> List[str]:
- """A short summary of what an archive holds, for --verbose."""
+def _fields_of(collection) -> List[str]:
+ """Every field name any grid of the collection carries, in first-seen order."""
+ names: List[str] = []
+ for step in collection.get("data", []):
+ grids = step.get("spatial_collection", [step])
+ for g in grids:
+ for a in g.get("attributes", []):
+ if a["name"] not in names:
+ names.append(a["name"])
+ return names
+
+
+def describe(domain) -> List[str]:
+ """What an archive holds, one line per collection, for ``--show`` and ``--verbose``.
+
+ Each collection is numbered so the interactive prompt can refer to it.
+ """
lines = []
- for collection in domain.get("time_collections", []):
+ for i, collection in enumerate(domain.get("time_collections", []), start=1):
steps = collection["data"]
times = [step["time"] for step in steps]
span = ("no steps" if not times
@@ -136,14 +162,70 @@ def _describe(domain) -> List[str]:
if "spatial_collection" in first:
detail = ", %d subdomains" % len(first["spatial_collection"])
first = first["spatial_collection"][0]
- fields = [a["name"] for a in first.get("attributes", [])]
detail += ", %s topology" % first["topology"]["type"]
+ fields = _fields_of(collection)
if fields:
detail += ", fields: %s" % ", ".join(fields)
- lines.append(" %s: %s%s" % (collection["name"], span, detail))
+ lines.append(" %d. %s: %s%s" % (i, collection["name"], span, detail))
return lines
+_describe = describe
+
+
+def _pick(prompt: str, options: Sequence[str], ask=None) -> Optional[List[str]]:
+ """Ask for some of ``options`` by number; blank keeps them all (None)."""
+ ask = ask or input
+ for i, option in enumerate(options, start=1):
+ print(" %d. %s" % (i, option))
+ answer = ask(prompt).strip()
+ if not answer:
+ return None
+ chosen = []
+ for token in answer.replace(",", " ").split():
+ try:
+ index = int(token)
+ except ValueError:
+ raise YmfArchiveError("%r is not a number" % (token,)) from None
+ if not 1 <= index <= len(options):
+ raise YmfArchiveError("%d is not one of 1..%d" % (index, len(options)))
+ if options[index - 1] not in chosen:
+ chosen.append(options[index - 1])
+ return chosen
+
+
+def interactive_selection(domain, ask=None):
+ """Ask which collections, fields and steps to convert.
+
+ Returns ``(collections, fields, steps)`` as :func:`ymf2xmf` takes
+ them, each ``None`` where everything was kept. ``ask`` defaults to
+ ``input`` (looked up at call time, so a test can replace it); tests
+ pass their own.
+ """
+ ask = ask or input
+ collections = domain.get("time_collections", [])
+ names = [c["name"] for c in collections]
+ print("Collections (blank for all):")
+ chosen_names = _pick("collections> ", names, ask)
+ kept = [c for c in collections if chosen_names is None or c["name"] in chosen_names]
+ fields: List[str] = []
+ for c in kept:
+ for f in _fields_of(c):
+ if f not in fields:
+ fields.append(f)
+ chosen_fields = None
+ if fields:
+ print("Fields (blank for all):")
+ chosen_fields = _pick("fields> ", fields, ask)
+ n_steps = max((len(c.get("data", [])) for c in kept), default=0)
+ steps = None
+ if n_steps > 1:
+ answer = ask("Steps of %d, as indices or slices, e.g. -1, 0:10, ::5 (blank for all)> "
+ % (n_steps,)).strip()
+ steps = answer or None
+ return chosen_names, chosen_fields, steps
+
+
def ymf2xmf_main(argv: Optional[Sequence[str]] = None) -> int:
"""``ymf2xmf`` console entry point."""
parser = argparse.ArgumentParser(
@@ -162,6 +244,18 @@ def ymf2xmf_main(argv: Optional[Sequence[str]] = None) -> int:
help="skip the structural check before converting")
parser.add_argument("-v", "--verbose", action="store_true",
help="describe what the archive holds")
+ parser.add_argument("--show", action="store_true",
+ help="describe what the archive holds and stop; convert nothing")
+ parser.add_argument("-c", "--collection", action="append", metavar="NAME",
+ help="convert only this time collection, by name or part of one "
+ "(repeatable; default: all)")
+ parser.add_argument("-f", "--field", action="append", metavar="NAME",
+ help="convert only this field (repeatable; default: all)")
+ parser.add_argument("-s", "--steps", metavar="SPEC",
+ help="convert only these steps: indices and slices, e.g. -1, "
+ "0:10, ::5 (default: all)")
+ parser.add_argument("-i", "--interactive", action="store_true",
+ help="ask which collections, fields and steps to convert")
args = parser.parse_args(argv)
source = Path(args.archive)
@@ -174,13 +268,21 @@ def ymf2xmf_main(argv: Optional[Sequence[str]] = None) -> int:
validate=not args.no_validate):
print("wrote %s" % (destination,))
return 0
- if args.verbose:
+ collections, fields, steps = args.collection, args.field, args.steps
+ if args.verbose or args.show or args.interactive:
domain, _ = read_ymf(source)
+ domain = canonicalize_domain(domain)
print("%s" % (source,))
- for line in _describe(canonicalize_domain(domain)):
- print(line)
+ if not args.interactive:
+ for line in describe(domain):
+ print(line)
+ if args.show:
+ return 0
+ if args.interactive:
+ collections, fields, steps = interactive_selection(domain)
destination = ymf2xmf(source, args.output,
- validate=not args.no_validate)
+ validate=not args.no_validate,
+ collections=collections, fields=fields, steps=steps)
except YmfArchiveError as exc:
# A structural problem in the archive, named. Not a traceback:
# the user asked to convert a file, and the file is the problem.
diff --git a/tests/test_select.py b/tests/test_select.py
new file mode 100644
index 0000000..607391f
--- /dev/null
+++ b/tests/test_select.py
@@ -0,0 +1,217 @@
+"""Cherry-picking an archive: ymf2xmf --collection/--field/--steps, --show and -i."""
+from xml.etree.ElementTree import parse
+
+import pytest
+
+from ymf.archive import (
+ YmfArchiveError,
+ add_spatial_step,
+ add_uniform_step,
+ attribute,
+ data_item,
+ geometry,
+ grid,
+ new_domain,
+ read_ymf,
+ select_domain,
+ step_indices,
+ topology,
+ write_ymf,
+)
+from ymf.cli import describe, interactive_selection, ymf2xmf, ymf2xmf_main
+
+
+def two_spaces(n_steps=4, spatial=False):
+ """Two collections, several fields, n_steps instants; a Proteus-shaped archive."""
+ domain = new_domain("Mesh Spatial_Domain")
+ quadratic = {"name": "Mesh_c0p2_Lagrange", "data": []}
+ domain["time_collections"].append(quadratic)
+ for i in range(n_steps):
+ t = 0.1 * i
+ def fields(names):
+ return [attribute(n, data_item([9], "out.h5:/%s_t%d" % (n, i), precision=8))
+ for n in names]
+ if spatial:
+ add_spatial_step(domain, t, [
+ grid(topology("Triangle", 10, data_item([10, 3], "out.h5:/e_p%d" % r, data_type="Int")),
+ geometry(data_item([9, 3], "out.h5:/n_p%d" % r, precision=8)),
+ fields(["u", "p"]), name="p%d" % r)
+ for r in range(2)], collection=0)
+ else:
+ add_uniform_step(domain, t,
+ topology("Triangle", 10, data_item([10, 3], "out.h5:/e", data_type="Int")),
+ geometry(data_item([9, 3], "out.h5:/n", precision=8)),
+ fields(["u", "p", "vof"]), collection=0)
+ add_uniform_step(domain, t,
+ topology("Tri_6", 10, data_item([10, 6], "out.h5:/e2", data_type="Int")),
+ geometry(data_item([25, 3], "out.h5:/n2", precision=8)),
+ fields(["u2"]), collection=1)
+ return domain
+
+
+def names_of(domain):
+ return [c["name"] for c in domain["time_collections"]]
+
+
+def fields_of(step):
+ return [a["name"] for a in step.get("attributes", [])]
+
+
+# --------------------------------------------------------------------------
+# step_indices
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("spec, expected", [
+ ("-1", [3]), ("0", [0]), ("0,2", [0, 2]), ("1:3", [1, 2]), ("::2", [0, 2]),
+ ("-2:", [2, 3]), ("2,0:2", [2, 0, 1]), ([3, 1], [3, 1]), (" 1 , 1 ", [1]),
+])
+def test_step_indices_read_like_python(spec, expected):
+ assert step_indices(spec, 4) == expected
+
+
+def test_a_step_the_archive_does_not_have_is_refused():
+ with pytest.raises(YmfArchiveError, match="step 7 does not exist"):
+ step_indices("7", 4)
+ with pytest.raises(YmfArchiveError, match="neither an index nor a slice"):
+ step_indices("last", 4)
+
+
+# --------------------------------------------------------------------------
+# select_domain
+# --------------------------------------------------------------------------
+
+
+def test_everything_is_kept_by_default():
+ domain = two_spaces()
+ assert select_domain(domain) == domain
+
+
+def test_a_collection_is_picked_by_name_or_part_of_it():
+ domain = two_spaces()
+ assert names_of(select_domain(domain, collections=["Mesh Spatial_Domain"])) == ["Mesh Spatial_Domain"]
+ assert names_of(select_domain(domain, collections=["c0p2"])) == ["Mesh_c0p2_Lagrange"]
+ with pytest.raises(YmfArchiveError, match="no collection matches 'dgp1'"):
+ select_domain(domain, collections=["dgp1"])
+
+
+def test_fields_are_picked_on_every_grid():
+ picked = select_domain(two_spaces(), fields=["p"])
+ base, quadratic = picked["time_collections"]
+ assert all(fields_of(step) == ["p"] for step in base["data"])
+ # the quadratic space has no p: its grids keep their mesh and lose their fields
+ assert all(fields_of(step) == [] for step in quadratic["data"])
+ assert quadratic["data"][0]["topology"]["type"] == "Tri_6"
+
+
+def test_fields_are_picked_inside_spatial_collections():
+ picked = select_domain(two_spaces(spatial=True), fields=["u"], collections=["Spatial"])
+ for step in picked["time_collections"][0]["data"]:
+ assert [fields_of(g) for g in step["spatial_collection"]] == [["u"], ["u"]]
+
+
+def test_an_unknown_field_is_refused_with_what_there_is():
+ with pytest.raises(YmfArchiveError, match="no field named 'phi'.*'p', 'u', 'u2', 'vof'"):
+ select_domain(two_spaces(), fields=["phi"])
+
+
+def test_steps_apply_to_every_collection():
+ picked = select_domain(two_spaces(), steps="-1")
+ for c in picked["time_collections"]:
+ assert [s["time"] for s in c["data"]] == [pytest.approx(0.3)]
+ picked = select_domain(two_spaces(), steps="::2")
+ assert [s["time"] for s in picked["time_collections"][0]["data"]] == [0.0, pytest.approx(0.2)]
+
+
+def test_the_source_is_not_modified():
+ domain = two_spaces()
+ select_domain(domain, collections=["c0p2"], fields=["u2"], steps="0")
+ assert names_of(domain) == ["Mesh Spatial_Domain", "Mesh_c0p2_Lagrange"]
+ assert fields_of(domain["time_collections"][0]["data"][3]) == ["u", "p", "vof"]
+
+
+# --------------------------------------------------------------------------
+# ymf2xmf with a selection, --show, -i
+# --------------------------------------------------------------------------
+
+
+@pytest.fixture
+def archive(tmp_path):
+ path = tmp_path / "run.ymf"
+ write_ymf(two_spaces(), path)
+ return path
+
+
+def grids_in(xmf):
+ root = parse(xmf).getroot()
+ return root.find("Domain").findall("Grid")
+
+
+def test_the_conversion_can_cherry_pick(archive, tmp_path):
+ out = ymf2xmf(archive, tmp_path / "part.xmf", collections=["Spatial"],
+ fields=["vof"], steps="-2:")
+ (temporal,) = grids_in(out)
+ assert temporal.attrib["Name"] == "Mesh Spatial_Domain"
+ steps = temporal.findall("Grid")
+ assert len(steps) == 2
+ assert [a.attrib["Name"] for a in steps[0].findall("Attribute")] == ["vof"]
+ # the archive itself is untouched
+ domain, _ = read_ymf(archive)
+ assert len(domain["time_collections"]) == 2
+
+
+def test_the_flags_reach_the_conversion(archive, tmp_path, capsys):
+ out = tmp_path / "part.xmf"
+ assert ymf2xmf_main([str(archive), "-o", str(out), "-c", "c0p2", "-f", "u2", "-s", "0"]) == 0
+ (temporal,) = grids_in(out)
+ assert temporal.attrib["Name"] == "Mesh_c0p2_Lagrange"
+ assert len(temporal.findall("Grid")) == 1
+
+
+def test_a_bad_selection_is_an_error_not_a_traceback(archive, capsys):
+ assert ymf2xmf_main([str(archive), "-f", "phi"]) == 1
+ assert "no field named 'phi'" in capsys.readouterr().err
+
+
+def test_show_describes_and_converts_nothing(archive, capsys):
+ assert ymf2xmf_main([str(archive), "--show"]) == 0
+ out = capsys.readouterr().out
+ assert "1. Mesh Spatial_Domain: 4 steps, t=0..0.3, Triangle topology, fields: u, p, vof" in out
+ assert "2. Mesh_c0p2_Lagrange" in out and "fields: u2" in out
+ assert not archive.with_suffix(".xmf").exists()
+
+
+def test_describe_lists_every_field_any_step_carries():
+ domain = two_spaces(n_steps=2)
+ # a field that appears only at the second step is still listed
+ domain["time_collections"][0]["data"][1]["attributes"].append(
+ attribute("late", data_item([9], "out.h5:/late", precision=8)))
+ (base, _) = describe(domain)
+ assert base.endswith("fields: u, p, vof, late")
+
+
+def test_the_interactive_prompt_builds_the_selection(capsys):
+ answers = iter(["1", "2 3", "-1"])
+ collections, fields, steps = interactive_selection(two_spaces(), ask=lambda prompt: next(answers))
+ assert (collections, fields, steps) == (["Mesh Spatial_Domain"], ["p", "vof"], "-1")
+ out = capsys.readouterr().out
+ assert "1. Mesh Spatial_Domain" in out and "1. u" in out
+
+
+def test_blank_answers_keep_everything():
+ answers = iter(["", "", ""])
+ assert interactive_selection(two_spaces(), ask=lambda prompt: next(answers)) == (None, None, None)
+
+
+def test_an_answer_that_is_not_a_choice_is_refused():
+ with pytest.raises(YmfArchiveError, match="not one of 1..2"):
+ interactive_selection(two_spaces(), ask=lambda prompt: "3")
+
+
+def test_interactive_from_the_command_line(archive, tmp_path, monkeypatch):
+ answers = iter(["2", "", ""])
+ monkeypatch.setattr("builtins.input", lambda prompt="": next(answers))
+ out = tmp_path / "part.xmf"
+ assert ymf2xmf_main([str(archive), "-i", "-o", str(out)]) == 0
+ (temporal,) = grids_in(out)
+ assert temporal.attrib["Name"] == "Mesh_c0p2_Lagrange"