From 2ae6b132f8b3b70706ee2f44de7366f9cbe2ad7d Mon Sep 17 00:00:00 2001 From: Lukas Date: Mon, 21 Sep 2026 10:24:02 +0200 Subject: [PATCH 1/4] feat(explorer): interactive front ends for exploring a simulation Adds fdsreader.explorer, a subpackage with one core and two front ends: explore() widgets for Jupyter fdsreader-explorer-cli a terminal application Both show a time bar, a 2D slice and any number of device or HRR curves. A desktop front end would sit beside them as explorer/gui.py. The core is importable with nothing but numpy, so the command line front end works on a machine with no plotting stack and no browser; matplotlib and ipywidgets are a new "notebook" extra, resolved lazily through a module-level __getattr__ so that importing fdsreader.explorer.cli never pulls them in. Slices are read one time step at a time. A slice file is a sequence of fixed-size records, so the wanted frame is seeked to directly: drawing a frame of a 749-step slice costs 0.3 MB and 0.1 ms, against 161 MB and 39 ms for materialising the whole series. The global colour scale is read from the .bnd files FDS writes beside the slice files, so it needs no field data at all. ExplorerState holds the selection and the rules acting on it, so a front end cannot invent its own meaning for "per step" or for the colour limits. fields.py assembles the stitched grid from the sub-slices rather than calling Slice.get_coordinates(), which raises for cell-centered slices (it takes the sub-slice dict keys, which are mesh ids, for meshes). Frames are verified to agree with Slice.to_global() cell for cell, and the terminal drawing is pinned by golden-text tests. 40 new tests, none of which need matplotlib or ipywidgets. --- docs/conf.py | 4 + docs/explorer.rst | 53 +++ docs/index.rst | 2 + fdsreader/explorer/__init__.py | 100 +++++ fdsreader/explorer/__main__.py | 5 + fdsreader/explorer/cli.py | 361 ++++++++++++++++ fdsreader/explorer/data.py | 230 +++++++++++ fdsreader/explorer/fields.py | 197 +++++++++ fdsreader/explorer/notebook.py | 731 +++++++++++++++++++++++++++++++++ fdsreader/explorer/plots.py | 191 +++++++++ fdsreader/explorer/render.py | 164 ++++++++ fdsreader/explorer/state.py | 96 +++++ pyproject.toml | 8 + tests/explorer/conftest.py | 99 +++++ tests/explorer/test_cli.py | 116 ++++++ tests/explorer/test_render.py | 106 +++++ tests/explorer/test_state.py | 70 ++++ 17 files changed, 2533 insertions(+) create mode 100644 docs/explorer.rst create mode 100644 fdsreader/explorer/__init__.py create mode 100644 fdsreader/explorer/__main__.py create mode 100644 fdsreader/explorer/cli.py create mode 100644 fdsreader/explorer/data.py create mode 100644 fdsreader/explorer/fields.py create mode 100644 fdsreader/explorer/notebook.py create mode 100644 fdsreader/explorer/plots.py create mode 100644 fdsreader/explorer/render.py create mode 100644 fdsreader/explorer/state.py create mode 100644 tests/explorer/conftest.py create mode 100644 tests/explorer/test_cli.py create mode 100644 tests/explorer/test_render.py create mode 100644 tests/explorer/test_state.py diff --git a/docs/conf.py b/docs/conf.py index 7f9d8bf..db4e4fb 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -37,6 +37,10 @@ # ones. extensions = ["sphinx.ext.autodoc", "sphinx.ext.viewcode", "autodocsumm"] +# fdsreader.explorer's Jupyter front end needs matplotlib and ipywidgets, which are an +# optional extra and are not installed for a docs build. +autodoc_mock_imports = ["matplotlib", "ipywidgets", "ipympl", "IPython"] + autoclass_content = "both" # List of patterns, relative to source directory, that match files and diff --git a/docs/explorer.rst b/docs/explorer.rst new file mode 100644 index 0000000..f68c77c --- /dev/null +++ b/docs/explorer.rst @@ -0,0 +1,53 @@ +Explorer +________ + +Front ends for looking at a simulation interactively. The core is importable with nothing +but numpy; the Jupyter front end additionally needs ``pip install "fdsreader[notebook]"``. + +.. automodule:: fdsreader.explorer + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.data + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.fields + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.state + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.plots + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.render + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.notebook + :autosummary: + :members: + :undoc-members: + :noindex: + +.. automodule:: fdsreader.explorer.cli + :autosummary: + :members: + :undoc-members: + :noindex: diff --git a/docs/index.rst b/docs/index.rst index fbf5545..d6c58fe 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -33,6 +33,7 @@ Source code, issue tracker and releases are hosted on fds_classes utils export + explorer .. include:: simulation.rst .. include:: slcf.rst @@ -48,3 +49,4 @@ Source code, issue tracker and releases are hosted on .. include:: fds_classes.rst .. include:: utils.rst .. include:: export.rst +.. include:: explorer.rst diff --git a/fdsreader/explorer/__init__.py b/fdsreader/explorer/__init__.py new file mode 100644 index 0000000..ca75ced --- /dev/null +++ b/fdsreader/explorer/__init__.py @@ -0,0 +1,100 @@ +"""Front ends for looking at an FDS simulation. + +Three of them share one core: + +* :mod:`~fdsreader.explorer.notebook` -- widgets for Jupyter, ``explore()`` +* :mod:`~fdsreader.explorer.cli` -- a terminal application, ``fdsreader-explorer-cli`` +* a desktop front end may follow; it would sit beside these two + +The shared parts are :mod:`~fdsreader.explorer.data` (reading devices, HRR and slices, +numpy only), :mod:`~fdsreader.explorer.plots` (matplotlib) and +:mod:`~fdsreader.explorer.render` (characters in a terminal). + +Importing this package pulls in nothing beyond numpy. The names that need matplotlib or +ipywidgets are resolved on first use, so:: + + from fdsreader.explorer.cli import main # numpy only + from fdsreader.explorer import explore # needs the notebook extras +""" + +from .data import ( + axis_label as axis_label, +) +from .data import ( + build_series as build_series, +) +from .data import ( + color_limits as color_limits, +) +from .data import ( + device_label as device_label, +) +from .data import ( + device_time as device_time, +) +from .data import ( + hrr_units as hrr_units, +) +from .data import ( + list_devices as list_devices, +) +from .data import ( + load_device_data as load_device_data, +) +from .data import ( + prepare_slice as prepare_slice, +) +from .data import ( + quantity_of as quantity_of, +) +from .data import ( + slice_label as slice_label, +) +from .data import ( + unit_of as unit_of, +) + +#: Names that live in a module with heavier requirements, resolved on first access. +_LAZY = { + "NotebookExplorer": "notebook", + "SimulationBrowser": "notebook", + "explore": "notebook", + "DIVERGING_CMAPS": "plots", + "SEQUENTIAL_CMAPS": "plots", + "plot_device": "plots", + "plot_series": "plots", + "plot_slice": "plots", + "RAMP": "render", + "series_lines": "render", + "slice_lines": "render", +} + + +def __getattr__(name): + """Import the front-end modules only when something from them is asked for.""" + module = _LAZY.get(name) + if module is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + from importlib import import_module + + return getattr(import_module(f".{module}", __name__), name) + + +def __dir__(): + return sorted(list(globals()) + list(_LAZY)) + + +__all__ = sorted(_LAZY) + [ + "axis_label", + "build_series", + "color_limits", + "device_label", + "device_time", + "hrr_units", + "list_devices", + "load_device_data", + "prepare_slice", + "quantity_of", + "slice_label", + "unit_of", +] diff --git a/fdsreader/explorer/__main__.py b/fdsreader/explorer/__main__.py new file mode 100644 index 0000000..a6aaf63 --- /dev/null +++ b/fdsreader/explorer/__main__.py @@ -0,0 +1,5 @@ +"""Run the command line front end with ``python -m fdsreader.explorer``.""" + +from .cli import main + +raise SystemExit(main()) diff --git a/fdsreader/explorer/cli.py b/fdsreader/explorer/cli.py new file mode 100644 index 0000000..64aadd7 --- /dev/null +++ b/fdsreader/explorer/cli.py @@ -0,0 +1,361 @@ +"""Command line front end. + +A one-shot renderer: it prints what a simulation contains, or draws one slice time step +or a set of curves, as text. Needs nothing but numpy, so it works over SSH on a machine +that has no plotting stack and no browser. + + fdsreader-explorer-cli CASE what is in it + fdsreader-explorer-cli CASE --slice 0 --time 4.4 draw a slice + fdsreader-explorer-cli CASE --curve T_1.0 --curve HRR +""" + +import argparse +import json +import sys + +import numpy as np + +import fdsreader +from fdsreader import settings + +from .data import build_series, list_devices, quantity_of, unit_of +from .fields import fields_of +from .render import RAMP, series_lines, slice_lines +from .state import ExplorerState + +PROG = "fdsreader-explorer-cli" + + +def _load(path, caching=False): + settings.ENABLE_CACHING = caching + return fdsreader.Simulation(str(path)) + + +def _state_from(args, fields): + """Build the shared state from the command line arguments.""" + state = ExplorerState(scale_mode="step") + if args.slice is not None: + state.field_index = args.slice + if args.scale in ("global", "step"): + state.scale_mode = args.scale + else: + try: + low, high = (float(part) for part in args.scale.split(",")) + except ValueError: + raise SystemExit(f"{PROG}: --scale wants 'global', 'step' or 'LOW,HIGH', not {args.scale!r}") + if not low < high: + raise SystemExit(f"{PROG}: --scale LOW must be below HIGH") + state.scale_mode, state.manual_limits = "manual", (low, high) + if args.time is not None and fields: + state.seek(fields, args.time) + return state + + +def overview(sim): + """What the simulation holds, as text lines.""" + devices = list_devices(sim) + series = build_series(sim) + n_hrr = sum(1 for s in series if s["source"] == "hrr") + + starts = [s["times"][0] for s in series if len(s["times"])] + ends = [s["times"][-1] for s in series if len(s["times"])] + + lines = [f"{sim.chid} {sim.root_path}", ""] + lines.append(f" meshes {len(sim.meshes)}") + lines.append(f" devices (DEVC) {len(devices)}") + lines.append(f" HRR quantities {n_hrr}") + lines.append(f" slices (SLCF) {len(fields_of(sim))}") + if starts: + lines.append(f" time {min(starts):g} … {max(ends):g} s") + + groups = {} + for device in devices: + groups.setdefault(quantity_of(device) or "—", []).append(unit_of(device)) + if groups: + lines += ["", " quantity # unit"] + for name, units in sorted(groups.items()): + shown = ", ".join(sorted(set(units) - {""})) or "—" + lines.append(f" {name:<28} {len(units):>4} {shown}") + return lines + + +def list_contents(sim): + """Every slice and every curve, with the names the other options take.""" + fields = fields_of(sim) + lines = ["slices:"] + if not fields: + lines.append(" (none)") + for i, field in enumerate(fields): + lines.append( + f" --slice {i:<3} {field.label} {len(field.times)} steps, t = {field.times[0]:g} … {field.times[-1]:g} s" + ) + + lines.append("") + lines.append("curves:") + series = build_series(sim) + if not series: + lines.append(" (none)") + for entry in series: + lines.append(f" --curve {entry['name']:<22} {entry['label']}") + return lines + + +def render_slice(fields, state, height, width, ramp): + field = state.field(fields) + if field is None: + raise SystemExit(f"{PROG}: this simulation has no slice output") + + timestep = state.clamp(fields) + frame = field.frame(timestep) + vmin, vmax, warning = state.limits(field, frame=frame) + + lines = [f"{field.label} · t = {field.times[timestep]:.3f} s (step {timestep + 1} of {len(field.times)})", ""] + if warning: + lines.append(f" note: {warning}") + lines += slice_lines(field, timestep, vmin, vmax, height=height, width=width, ramp=ramp, frame=frame) + return lines + + +def save_figure(fields, state, series, path, height, width): + """Write the current view to an image or an animation, using matplotlib.""" + try: + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt + except ModuleNotFoundError: + raise SystemExit(f'{PROG}: --save needs matplotlib: pip install "fdsreader[notebook]"') + from .plots import plot_series, plot_slice + + field = state.field(fields) + animated = path.suffix.lower() in (".gif", ".mp4") + + if animated: + if field is None: + raise SystemExit(f"{PROG}: --save {path.suffix} needs a slice to animate") + from matplotlib import animation + + steps = range(0, len(field.times), max(1, len(field.times) // 120)) + fig, ax = plt.subplots() + + def draw(step): + ax.clear() + frame = field.frame(step) + vmin, vmax, _ = state.limits(field, frame=frame) + plot_slice( + field, + step, + ax=ax, + vmin=vmin, + vmax=vmax, + cmap=state.colormap(field), + render=state.render, + n_levels=state.n_levels, + frame=frame, + ) + + writer = "pillow" if path.suffix.lower() == ".gif" else "ffmpeg" + anim = animation.FuncAnimation(fig, draw, frames=list(steps), interval=80) + try: + anim.save(str(path), writer=writer) + except Exception as exc: + raise SystemExit(f"{PROG}: could not write {path}: {exc}") + plt.close(fig) + return f"wrote {path} ({len(list(steps))} frames)" + + panels = (1 if field is None else 1) + (1 if series else 0) + fig, axes = plt.subplots(1, max(1, panels), figsize=(7 * max(1, panels), 5.5)) + axes = np.atleast_1d(axes) + at = 0 + if field is not None: + timestep = state.clamp(fields) + frame = field.frame(timestep) + vmin, vmax, _ = state.limits(field, frame=frame) + plot_slice( + field, + timestep, + ax=axes[at], + vmin=vmin, + vmax=vmax, + cmap=state.colormap(field), + render=state.render, + n_levels=state.n_levels, + frame=frame, + ) + at += 1 + if series: + plot_series(series, ax=axes[at], cursor_time=state.time(fields) if field is not None else None) + fig.tight_layout() + fig.savefig(str(path), dpi=150) + plt.close(fig) + return f"wrote {path}" + + +def as_json(sim, fields, state, series): + """The same information as the text output, for a script to consume.""" + devices = list_devices(sim) + all_series = build_series(sim) + payload = { + "chid": sim.chid, + "path": sim.root_path, + "meshes": len(sim.meshes), + "devices": len(devices), + "hrr_quantities": sum(1 for e in all_series if e["source"] == "hrr"), + "slices": [ + { + "index": i, + "label": f.label, + "quantity": f.quantity, + "unit": f.unit, + "shape": list(f.shape), + "axes": list(f.axes), + "n_timesteps": len(f.times), + "time": [float(f.times[0]), float(f.times[-1])], + "range": [float(f.value_range()[0]), float(f.value_range()[1])], + } + for i, f in enumerate(fields) + ], + "curves": [ + { + "name": e["name"], + "label": e["label"], + "quantity": e["quantity"], + "unit": e["unit"], + "source": e["source"], + "n_samples": int(len(e["values"])), + "time": [float(e["times"][0]), float(e["times"][-1])], + "min": float(np.nanmin(e["values"])), + "max": float(np.nanmax(e["values"])), + "final": float(e["values"][-1]), + } + for e in all_series + ], + } + if series: + payload["selected"] = [ + {"name": e["name"], "times": [float(v) for v in e["times"]], "values": [float(v) for v in e["values"]]} + for e in series + ] + return payload + + +def resolve_curves(sim, names): + """Look up the requested curves by name.""" + by_name = {} + for entry in build_series(sim): + by_name.setdefault(entry["name"], entry) + chosen = [] + for name in names: + if name not in by_name: + raise SystemExit(f"{PROG}: no curve called {name!r} — try --list") + chosen.append(by_name[name]) + return chosen + + +def build_parser(): + parser = argparse.ArgumentParser( + prog=PROG, + description="Look at FDS output in a terminal.", + epilog="With no options it prints what the simulation contains.", + ) + parser.add_argument("path", help="simulation directory, or the .smv file") + parser.add_argument("--list", action="store_true", help="list every slice and curve with the name to pass back in") + parser.add_argument("--slice", type=int, metavar="N", help="draw slice N") + parser.add_argument( + "--time", type=float, metavar="SECONDS", help="time step to draw, the nearest one is used (default: first)" + ) + parser.add_argument( + "--scale", default="step", metavar="MODE", help="colour limits: 'global', 'step' (default) or 'LOW,HIGH'" + ) + parser.add_argument( + "--curve", + action="append", + default=[], + metavar="NAME", + help="draw this device or HRR quantity; repeat for several", + ) + parser.add_argument( + "--height", type=int, default=26, metavar="ROWS", help="character rows for the slice (default: 26)" + ) + parser.add_argument( + "--width", type=int, default=None, metavar="COLS", help="character columns (default: fit the terminal)" + ) + parser.add_argument("--ramp", default=None, metavar="CHARS", help="characters from empty to full, e.g. ' .oO@'") + parser.add_argument("--json", action="store_true", help="print the same information as JSON, for scripting") + parser.add_argument( + "--save", + metavar="FILE", + help="write the current view to an image (.png/.pdf/.svg) or an " + "animation (.gif/.mp4) instead of drawing it as text; " + "needs matplotlib", + ) + parser.add_argument( + "--render", + default="image", + choices=("image", "filled", "lines"), + help="how --save draws a slice (default: image)", + ) + parser.add_argument( + "--caching", + action="store_true", + help="let fdsreader cache the parsed simulation as a .pickle " + "next to the data; off by default so that read-only " + "directories work", + ) + return parser + + +def main(argv=None): + """Entry point. Returns a process exit status.""" + args = build_parser().parse_args(argv) + + try: + sim = _load(args.path, caching=args.caching) + except Exception as exc: + print(f"{PROG}: could not load {args.path}: {exc}", file=sys.stderr) + return 1 + + fields = fields_of(sim) + if args.slice is not None and not 0 <= args.slice < len(fields): + print( + f"{PROG}: --slice {args.slice} out of range (0 … {len(fields) - 1})" + if fields + else f"{PROG}: this simulation has no slice output", + file=sys.stderr, + ) + return 1 + + state = _state_from(args, fields) + state.render = args.render + series = resolve_curves(sim, args.curve) + + if args.json: + json.dump(as_json(sim, fields, state, series), sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 + + if args.save: + from pathlib import Path + + print(save_figure(fields, state, series, Path(args.save), args.height, args.width)) + return 0 + + lines = [] + if args.list: + lines += list_contents(sim) + elif args.slice is None and not args.curve: + lines += overview(sim) + + if args.slice is not None: + lines += render_slice(fields, state, args.height, args.width, args.ramp or RAMP) + if series: + if lines: + lines.append("") + lines += series_lines(series, height=max(6, args.height // 2), width=args.width, cursor_time=args.time) + + print("\n".join(lines)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/fdsreader/explorer/data.py b/fdsreader/explorer/data.py new file mode 100644 index 0000000..1686311 --- /dev/null +++ b/fdsreader/explorer/data.py @@ -0,0 +1,230 @@ +"""Reading the parts of a simulation the explorer shows. + +Only depends on numpy, so it can be used without the plotting and widget extras. +""" + +import csv +from pathlib import Path + +import numpy as np + +TIME_DEVICE_ID = "Time" + + +def list_devices(sim): + """All real devices of a simulation as a flat list. + + The ``Time`` pseudo-device that fdsreader adds for the device time column is dropped. + Devices sharing an ID are stored as a list, and are flattened here. + """ + devices = [] + for entry in sim.devices: + group = entry if isinstance(entry, list) else [entry] + for device in group: + if device.id != TIME_DEVICE_ID: + devices.append(device) + return devices + + +def load_device_data(devices): + """Force the device data to be read. + + fdsreader reads the ``_devc.csv`` lazily and only fills in ``device.unit`` at that + point, so units would be missing from labels otherwise. Reading one device loads the + whole file, hence the single access. + """ + for device in devices: + try: + device.data + except Exception: + pass + break + + +def device_time(sim, device): + """Time axis in seconds matching ``device.data``.""" + if TIME_DEVICE_ID in sim.devices: + times = np.asarray(sim.devices[TIME_DEVICE_ID].data) + if len(times) == len(device.data): + return times + # Fall back to the sample index if the simulation has no device time column. + return np.arange(len(device.data)) + + +def unit_of(device): + """Unit string, cleaned up: fdsreader keeps the trailing newline of the CSV header.""" + return (device.unit or "").strip().strip('"') + + +def quantity_of(device): + quantity = device.quantity + return (quantity.name if quantity is not None else "").strip() + + +def device_label(device, duplicate_ids=()): + """Human-readable entry, e.g. ``TC_1 - TEMPERATURE [C]``. + + Devices whose ID is not unique are additionally identified by their position. + """ + parts = [device.id] + quantity = quantity_of(device) + if quantity: + parts.append(f"— {quantity}") + unit = unit_of(device) + if unit: + parts.append(f"[{unit}]") + if device.id in duplicate_ids: + x, y, z = device.position + parts.append(f"@ ({x:g}, {y:g}, {z:g})") + return " ".join(parts) + + +def slice_label(slc, index=None): + """Entry for a slice, e.g. ``TEMPERATURE [C] - x = 0 m``. + + Slices usually have an empty ``id`` -- FDS only names them when ``SLCF ID`` is given -- + so they are identified by quantity and by the plane they cut. + """ + quantity = slc.quantity.name.strip() + unit = (slc.quantity.unit or "").strip() + head = f"{quantity} [{unit}]" if unit else quantity + if slc.orientation in (1, 2, 3): + dim = "xyz"[slc.orientation - 1] + head += f" — {dim} = {slc.extent[dim][0]:g} m" + else: + head += " — 3D" + if slc.id.strip(): + head += f" ({slc.id.strip()})" + elif index is not None: + head += f" #{index}" + return head + + +def color_limits(values, diverging=None): + """Colour scale for an array: its real minimum and maximum. + + ``diverging`` is decided once for a whole slice and handed back in for single frames, + so the scale cannot flip between centred and one-sided while stepping through time. + """ + low, high = float(np.nanmin(values)), float(np.nanmax(values)) + if not np.isfinite(low) or not np.isfinite(high): + low, high = 0.0, 1.0 + if diverging is None: + diverging = low < 0 < high + if diverging: # velocities read better on a scale centred at zero + bound = max(abs(low), abs(high)) or 1.0 + low, high = -bound, bound + if low == high: # a frame of constant value still needs a finite range + low, high = low - 0.5, high + 0.5 + return low, high, diverging + + +def prepare_slice(slc): + """Read a slice into one global array and work out how to draw it. + + ``to_global`` stitches the per-mesh subslices together and returns the data with the + slice-normal axis squeezed out, so the array is ``(n_t, dim1, dim2)`` with dim1/dim2 in + x, y, z order. Returns ``None`` for anything that is not a drawable 2D slice. + """ + data, coords = slc.to_global(return_coordinates=True) + + # A cell-centered slice sitting exactly on a mesh border has a valid representation on + # either side, and fdsreader hands back both. The first one is used here. + if isinstance(data, tuple): + data, coords = data[0], coords[0] + + spanned = [d for d in ("x", "y", "z") if len(coords[d]) > 1] + if data.size == 0 or len(spanned) != 2: + return None + horizontal, vertical = spanned # x, y, z order puts the upright axis second + + low, high, diverging = color_limits(data) + return { + "slice": slc, + "data": data, + "coords": coords, + "axes": (horizontal, vertical), + "vmin": low, + "vmax": high, + "diverging": diverging, + "cmap": "RdBu_r" if diverging else "inferno", + } + + +def hrr_units(sim): + """Units of the HRR columns. + + ``sim.hrr`` is a plain dict of name -> values; fdsreader skips the unit line of the + ``_hrr.csv`` header when reading it, so it is read again here. + """ + for path in sorted(Path(sim.root_path).glob("*_hrr.csv")): + try: + with open(path) as infile: + units = next(csv.reader([infile.readline()])) + names = next(csv.reader([infile.readline()])) + except (OSError, StopIteration): + return {} + return {n.strip(): u.strip() for n, u in zip(names, units)} + return {} + + +def build_series(sim): + """Every plottable time series of a simulation: the devices and the HRR quantities. + + Both are reduced to the same shape so they can be picked from one list and drawn on one + figure. Each series keeps its own time axis, because the HRR file and the device file + are not necessarily written at the same interval. + """ + series = [] + + devices = list_devices(sim) + load_device_data(devices) # so that units are known when building the labels + seen, duplicate_ids = set(), set() + for device in devices: + (duplicate_ids if device.id in seen else seen).add(device.id) + + for device in devices: + series.append( + { + "label": "DEVC " + device_label(device, duplicate_ids), + "name": device.id, + "times": device_time(sim, device), + "values": np.asarray(device.data), + "quantity": quantity_of(device), + "unit": unit_of(device), + "source": "device", + "device": device, + } + ) + + # `sim.hrr` only exists when the simulation wrote an HRR file. + hrr = getattr(sim, "hrr", None) + if hrr and "Time" in hrr: + units = hrr_units(sim) + times = np.asarray(hrr["Time"]) + for name, values in hrr.items(): + if name == "Time": + continue + unit = units.get(name, "").strip() + series.append( + { + "label": f"HRR {name}" + (f" [{unit}]" if unit else ""), + "name": name, + "times": times, + "values": np.asarray(values), + "quantity": name, + "unit": unit, + "source": "hrr", + "device": None, + } + ) + + return series + + +def axis_label(group): + """Y-axis label for a group of series sharing an axis.""" + quantities = {s["quantity"] for s in group} + units = sorted({s["unit"] for s in group if s["unit"]}) + head = quantities.pop() if len(quantities) == 1 else "Value" + return f"{head} [{', '.join(units)}]" if units else head diff --git a/fdsreader/explorer/fields.py b/fdsreader/explorer/fields.py new file mode 100644 index 0000000..e565ead --- /dev/null +++ b/fdsreader/explorer/fields.py @@ -0,0 +1,197 @@ +"""A 2D field that can be drawn at one instant in time. + +Every front end asks a field the same three things -- what are your coordinates, what did +you look like at time step *t*, and what is your value range -- so slices, and later +boundary data or a plane cut from 3D output, can all be drawn by the same code. + +Frames are read one at a time. A slice file is a sequence of fixed-size records, so the +frame wanted can be seeked to directly instead of materialising the whole time series, +which for a production run would be gigabytes. +""" + +import os + +import numpy as np + +import fdsreader.utils.fortran_data as fdtype + +from .data import color_limits, slice_label + + +def _read_subslice_frame(subslice, timestep): + """One time step of one sub-slice, without reading the rest of the file. + + Mirrors what fdsreader does when it loads the whole array, but seeks to a single + record. Belongs in :mod:`fdsreader.slcf` as ``SubSlice.frame(t)`` eventually. + """ + count = subslice.dimension.size(cell_centered=False) + record = fdtype.combine(fdtype.FLOAT, fdtype.new((("f", count),))) + path = os.path.join(subslice._parent_slice._root_path, subslice.filename) + + with open(path, "rb") as infile: + infile.seek(subslice._offset + timestep * record.itemsize) + data = fdtype.read(infile, record, 1)[0] + + frame = data[1].reshape(subslice.dimension.shape(cell_centered=False), order="F") + if subslice.cell_centered: + # Ignore ghost points on every axis, as the eager path does. + frame = frame[(slice(1, None),) * frame.ndim] + return frame + + +def _bnd_range(slc): + """Value range of a slice from the ``.bnd`` files FDS writes beside the slice files. + + Saves reading any data at all for the global colour scale. The range covers every + sub-slice, so where a cell-centered slice straddles a mesh border it can be slightly + wider than the half that ends up on screen. Returns ``None`` when a file is missing. + """ + low, high = np.inf, -np.inf + for subslice in slc.subslices: + path = os.path.join(slc._root_path, subslice.filename + ".bnd") + if not os.path.exists(path): + return None + try: + with open(path) as infile: + for line in infile: + parts = line.split() + if len(parts) >= 3: + low = min(low, float(parts[1])) + high = max(high, float(parts[2])) + except (OSError, ValueError): + return None + return (float(low), float(high)) if np.isfinite(low) and np.isfinite(high) else None + + +def _dedup(values): + """Sort and drop near-duplicates: FDS writes coordinates with limited precision.""" + values = np.sort(np.asarray(values, dtype=float)) + if values.size: + values = values[np.concatenate(([True], np.diff(values) > 2e-6))] + return values + + +def _sides(slc, normal): + """Sub-slices grouped by where they sit along the slice normal. + + A cell-centered slice lying exactly on a mesh border has no cells *at* the border, so + FDS writes the cells on either side of it. Each side is a complete, equally valid + picture of the slice; the eager path returns both and this keeps them apart. + """ + groups = {} + for subslice in slc.subslices: + coord = subslice.get_coordinates()[normal] + if not len(coord): + continue + groups.setdefault(round(float(coord[0]), 6), []).append(subslice) + return [groups[key] for key in sorted(groups)] + + +class SliceField: + """A 2D ``SLCF`` slice, stitched across meshes one time step at a time.""" + + kind = "slice" + + def __init__(self, slc, index=None, side=0): + self.source = slc + self.index = index + self.label = slice_label(slc, index) + self.quantity = slc.quantity.name.strip() + self.unit = (slc.quantity.unit or "").strip() + self.times = np.asarray(slc.times, dtype=float) + + self.drawable = slc.orientation in (1, 2, 3) + if not self.drawable: # a 3D slice has no single plane to draw + self.axes, self.coords, self._placement, self.shape = (), {}, [], (0, 0) + return + + normal = "xyz"[slc.orientation - 1] + horizontal, vertical = [d for d in ("x", "y", "z") if d != normal] + self.axes = (horizontal, vertical) # x, y, z order puts the upright axis second + + sides = _sides(slc, normal) + subslices = sides[min(side, len(sides) - 1)] if sides else [] + + # Cell coordinates decide where each sub-slice goes in the stitched grid ... + cell = ( + {d: _dedup(np.concatenate([sub.get_coordinates()[d] for sub in subslices])) for d in (horizontal, vertical)} + if subslices + else {horizontal: np.array([]), vertical: np.array([])} + ) + self.shape = (len(cell[horizontal]), len(cell[vertical])) + + # ... while the drawn axes span the extent the slice actually covers, so that a + # cell-centered field is not drawn half a cell adrift. + self.coords = { + normal: np.array([float(slc.extent[normal][0])]), + horizontal: np.linspace(*slc.extent[horizontal], self.shape[0]), + vertical: np.linspace(*slc.extent[vertical], self.shape[1]), + } + + self._placement = [] + for subslice in subslices: + sub_coords = subslice.get_coordinates() + sub_h, sub_v = sub_coords[horizontal], sub_coords[vertical] + if not len(sub_h) or not len(sub_v): + continue + i0 = int(np.argmin(np.abs(cell[horizontal] - sub_h[0]))) + j0 = int(np.argmin(np.abs(cell[vertical] - sub_v[0]))) + self._placement.append((subslice, i0, len(sub_h), j0, len(sub_v))) + + self.drawable = self.shape[0] > 1 and self.shape[1] > 1 + self._range = None + + # -- data ------------------------------------------------------------ + def frame(self, timestep): + """The stitched field at one time step, as ``(n_horizontal, n_vertical)``.""" + out = np.full(self.shape, np.nan, dtype=np.float32) + written = np.zeros(self.shape, dtype=bool) + + for subslice, i0, ni, j0, nj in self._placement: + data = _read_subslice_frame(subslice, timestep) + i1, j1 = min(i0 + ni, self.shape[0]), min(j0 + nj, self.shape[1]) + patch = data[: i1 - i0, : j1 - j0] + region = (slice(i0, i1), slice(j0, j1)) + # Where a cell-centered slice straddles a mesh border two sub-slices claim the + # same cells; the first one wins, the way the eager path keeps one of them. + out[region] = np.where(written[region], out[region], patch) + written[region] = True + return out + + def value_range(self): + """``(vmin, vmax, diverging)`` over the whole time series. + + Read from the ``.bnd`` files when FDS wrote them, so no field data is touched. + """ + if self._range is None: + bounds = _bnd_range(self.source) + if bounds is None: # no .bnd files: fall back to scanning the frames + low, high = np.inf, -np.inf + for t in range(len(self.times)): + frame = self.frame(t) + low = min(low, float(np.nanmin(frame))) + high = max(high, float(np.nanmax(frame))) + bounds = (low, high) + self._range = color_limits(np.asarray(bounds, dtype=float)) + return self._range + + @property + def cmap(self): + return "RdBu_r" if self.value_range()[2] else "inferno" + + def clear_cache(self): + self.source.clear_cache() + + +def fields_of(sim): + """Every drawable 2D field of a simulation. + + Slices today. Boundary data and planes cut from 3D output would be appended here, and + every front end would pick them up without changing. + """ + fields = [] + for i, slc in enumerate(sim.slices): + field = SliceField(slc, i) + if field.drawable: + fields.append(field) + return fields diff --git a/fdsreader/explorer/notebook.py b/fdsreader/explorer/notebook.py new file mode 100644 index 0000000..88dfc7b --- /dev/null +++ b/fdsreader/explorer/notebook.py @@ -0,0 +1,731 @@ +"""Jupyter front end: a directory browser, a time bar, a slice and a curve plot. + +Importing this module requires the notebook extras:: + + pip install "fdsreader[notebook]" +""" + +try: + import ipywidgets as widgets +except ModuleNotFoundError as exc: # pragma: no cover - depends on the installation + raise ModuleNotFoundError( + "fdsreader.explorer.notebook needs ipywidgets and matplotlib. Install them with:\n" + ' pip install "fdsreader[notebook]"\n' + "ipympl is optional and adds zooming and panning." + ) from exc + +from contextlib import contextmanager +from importlib.util import find_spec +from pathlib import Path + +import matplotlib.pyplot as plt +from IPython.display import display + +import fdsreader +from fdsreader import settings + +from .data import build_series, list_devices, quantity_of, unit_of +from .fields import fields_of +from .plots import DIVERGING_CMAPS, SEQUENTIAL_CMAPS, plot_series, plot_slice +from .state import ExplorerState + + +def _select_backend(preference=None): + """Switch matplotlib to ipympl when it is usable, otherwise to inline images. + + ``preference`` is ``None`` to auto-detect, ``True`` to insist on ipympl and ``False`` + to force inline. Auto-detection cannot see whether the *browser* has the matching + front-end extension, so pass ``False`` when the plots come up as + "Failed to load model class 'MPLCanvasModel'". + """ + try: + from IPython import get_ipython + + shell = get_ipython() + except Exception: + shell = None + + if preference is not False: + # Probe without importing: importing ipympl switches the matplotlib backend as a + # side effect, and that would stick even when the inline fall-back is taken. + available = find_spec("ipympl") is not None + if available and shell is not None: + try: + shell.run_line_magic("matplotlib", "widget") + return True + except Exception: + if preference is True: + raise + elif preference is True: + raise RuntimeError( + "interactive_canvas=True, but " + + ("ipympl is not installed" if not available else "no IPython shell is running") + ) + + if shell is not None: + try: + shell.run_line_magic("matplotlib", "inline") + except Exception: + pass + return False + + +@contextmanager +def _figure_surface(fig, view, figsize): + """Hand out a blank figure to draw on, whichever backend is in use. + + With ipympl the one long-lived figure is cleared and redrawn in place, because its + canvas is a widget that already sits in the layout. Without it, a throw-away figure is + rendered into an ``Output`` widget and closed again -- inline figures made inside a + widget callback are not cleaned up on their own and would otherwise pile up. + """ + if fig is not None: + fig.clear() + yield fig + fig.canvas.draw_idle() + else: + view.clear_output(wait=True) + with view: + throwaway = plt.figure(figsize=figsize) + try: + yield throwaway + plt.show() + finally: + plt.close(throwaway) + + +def _clear_surface(fig, view): + """Blank the plot area without drawing anything into it.""" + if fig is not None: + fig.clear() + fig.canvas.draw_idle() + else: + view.clear_output() + + +class SimulationBrowser: + """Directory browser that loads the FDS simulation found in the selected folder.""" + + def __init__(self, start_dir=None, on_load=None): + self.on_load = on_load + self.sim = None + self.current = None + + self.path_box = widgets.Text( + description="Path:", + continuous_update=False, + layout=widgets.Layout(flex="1 1 auto", width="auto", min_width="240px"), + style={"description_width": "50px"}, + ) + self.up_button = widgets.Button(description="⬆ Up", layout=widgets.Layout(width="90px")) + self.folders = widgets.Select( + options=[], + rows=10, + layout=widgets.Layout(width="auto", min_width="220px"), + ) + self.smv_picker = widgets.Dropdown( + description=".smv:", + layout=widgets.Layout(width="340px", display="none"), + style={"description_width": "50px"}, + ) + self.load_button = widgets.Button( + description="Load simulation", + button_style="primary", + disabled=True, + layout=widgets.Layout(width="160px"), + ) + self.status = widgets.HTML() + + self.up_button.on_click(lambda _: self.go_to(self.current.parent)) + self.load_button.on_click(lambda _: self.load()) + self.folders.observe(self._on_folder_selected, names="value") + self.path_box.observe(self._on_path_typed, names="value") + + # The folder list and the status share a row so the browser stays shallow and + # leaves the vertical space to the plots. + self.widget = widgets.VBox( + [ + widgets.HBox([self.path_box], layout=widgets.Layout(width="100%")), + widgets.HBox( + [self.up_button, self.load_button, self.smv_picker], + layout=widgets.Layout(flex_flow="row wrap", align_items="center"), + ), + widgets.HBox( + [ + widgets.Box([self.folders], layout=widgets.Layout(flex="1 1 280px", min_width="240px")), + widgets.Box( + [self.status], + layout=widgets.Layout(flex="1 1 300px", min_width="240px", padding="4px 0 0 14px"), + ), + ], + layout=widgets.Layout(width="100%", flex_flow="row wrap", align_items="flex-start"), + ), + ], + layout=widgets.Layout(width="100%"), + ) + + self.go_to(Path(start_dir) if start_dir is not None else Path.cwd()) + + # -- navigation ------------------------------------------------------ + def _on_path_typed(self, change): + if self.current is not None and change["new"] != str(self.current): + self.go_to(change["new"]) + + def _on_folder_selected(self, change): + if change["new"]: + self.go_to(Path(change["new"])) + + def go_to(self, directory): + directory = Path(directory).expanduser() + # A typed relative path should be relative to the folder on screen, not to the + # directory the Jupyter kernel happens to have been started in. + if not directory.is_absolute() and self.current is not None: + directory = self.current / directory + if not directory.is_dir(): + self.status.value = f"Not a directory: {directory}" + return + + self.current = directory.resolve() + self.path_box.value = str(self.current) + + try: + subdirs = sorted( + (p for p in self.current.iterdir() if p.is_dir() and not p.name.startswith(".")), + key=lambda p: p.name.lower(), + ) + except PermissionError: + subdirs = [] + self.status.value = "Permission denied." + + # Rebuild the list without re-triggering navigation. + self.folders.unobserve(self._on_folder_selected, names="value") + self.folders.options = [(p.name + "/", str(p)) for p in subdirs] + self.folders.value = None + self.folders.observe(self._on_folder_selected, names="value") + + self._scan_for_smv() + + def _scan_for_smv(self): + smv_files = sorted(self.current.glob("*.smv")) + + if not smv_files: + self.smv_picker.layout.display = "none" + self.load_button.disabled = True + self.status.value = ( + f"No .smv file in " + f"{self.current.name or self.current} — keep browsing." + ) + return + + self.load_button.disabled = False + self.smv_picker.options = [(p.name, str(p)) for p in smv_files] + # Assigning `options` does not select anything while the index is None. + self.smv_picker.index = 0 + if len(smv_files) > 1: + self.smv_picker.layout.display = "flex" + self.status.value = ( + f"{len(smv_files)} simulations found — choose one, then click Load simulation." + ) + else: + self.smv_picker.layout.display = "none" + self.status.value = f"Found {smv_files[0].name} — click Load simulation." + + # -- loading --------------------------------------------------------- + def load(self): + smv_path = self.smv_picker.value + if smv_path is None or Path(smv_path).parent != self.current: + self.status.value = "No simulation selected in this folder." + return + + self.status.value = f"Loading {Path(smv_path).name} …" + try: + self.sim = fdsreader.Simulation(smv_path) + except Exception as exc: + self.sim = None + self.status.value = f"Could not load: {exc}" + return + + devices = list_devices(self.sim) + self.status.value = ( + f"Loaded {self.sim.chid} — " + f"{len(devices)} device(s), {len(self.sim.meshes)} mesh(es)." + ) + if self.on_load is not None: + self.on_load(self.sim) + + +class NotebookExplorer: + """Interactive view of one FDS simulation at a time. + + Shows a directory browser, a time bar, a 2D slice and any number of device or HRR + curves. The slice and the curves sit side by side on a wide window and wrap onto + their own lines on a narrow one. + + :param path: Simulation to load straight away; if omitted, browse to one. + :param start_dir: Where the directory browser opens. Defaults to the working directory. + :param interactive_canvas: ``None`` to use ipympl when available, ``True`` to insist, + ``False`` to force static inline images. Pass ``False`` if the plots come up as + "Failed to load model class 'MPLCanvasModel'", which means the browser-side + extension is missing. + :param caching: Sets :data:`fdsreader.settings.ENABLE_CACHING`. Off by default: + fdsreader caches a parsed simulation as a ``.pickle`` next to the data, which + fails on a read-only directory such as a shared course folder. Note that such a + directory must also be free of leftover ``.pickle`` files, since switching the + cache off makes fdsreader try to delete them. + """ + + def __init__( + self, + path=None, + start_dir=None, + interactive_canvas=None, + caching=False, + slice_figsize=(6.5, 5.5), + curve_figsize=(7.0, 5.5), + panel_basis="560px", + panel_min="360px", + ): + settings.ENABLE_CACHING = caching + self.interactive_canvas = _select_backend(interactive_canvas) + self.slice_figsize = slice_figsize + self.curve_figsize = curve_figsize + + self.sim = None + self.series = [] + self.devices = [] + self.fields = [] + self.state = ExplorerState() + self._updating = False + + self._build_widgets(panel_basis, panel_min) + if start_dir is None and path is not None: + start_dir = Path(path).expanduser().parent if Path(path).is_file() else path + self.browser = SimulationBrowser(start_dir=start_dir, on_load=self._on_simulation_loaded) + + self.widget = widgets.VBox( + [ + self.browser.widget, + self._rule(), + self.summary, + self._rule(), + self.time_bar, + self._rule(), + widgets.HBox( + [self.slice_panel, self.curve_panel], + layout=widgets.Layout(width="100%", flex_flow="row wrap", align_items="flex-start"), + ), + ], + layout=widgets.Layout(width="100%"), + ) + + if path is not None: + self.load(path) + + # -- construction ---------------------------------------------------- + @staticmethod + def _rule(): + return widgets.HTML("
") + + def _build_widgets(self, panel_basis, panel_min): + self.summary = widgets.HTML() + + if self.interactive_canvas: + # With ipympl a figure is a live widget: create it once, put it in the layout + # and redraw it in place. A new one per update would stack up canvases. + with plt.ioff(): + self.slice_fig = plt.figure(figsize=self.slice_figsize) + self.curve_fig = plt.figure(figsize=self.curve_figsize) + for fig in (self.slice_fig, self.curve_fig): + canvas = fig.canvas + canvas.header_visible = False + canvas.footer_visible = False + canvas.layout.width = "100%" + canvas.layout.height = "auto" + self.slice_view, self.curve_view = self.slice_fig.canvas, self.curve_fig.canvas + else: + self.slice_fig = self.curve_fig = None + self.slice_view = widgets.Output() + self.curve_view = widgets.Output() + + self.series_select = widgets.SelectMultiple( + description="Curves:", + options=[], + disabled=True, + rows=9, + layout=widgets.Layout(width="auto"), + style={"description_width": "60px"}, + ) + self.series_hint = widgets.HTML( + "Devices (DEVC) and the quantities of " + "the HRR file. Hold ⌘/Ctrl to add single entries, ⇧ for a run of them." + ) + + self.slice_dropdown = widgets.Dropdown( + description="Slice:", + options=[], + disabled=True, + layout=widgets.Layout(width="auto"), + style={"description_width": "60px"}, + ) + self.scale_mode = widgets.ToggleButtons( + options=[("Global", "global"), ("Per step", "step"), ("Manual", "manual")], + value="global", + tooltips=[ + "One scale for the whole time series", + "Rescale to the time step on screen", + "Type the limits yourself", + ], + layout=widgets.Layout(width="auto"), + style={"button_width": "auto"}, + ) + self.scale_min = widgets.FloatText( + description="min", layout=widgets.Layout(width="140px"), style={"description_width": "30px"} + ) + self.scale_max = widgets.FloatText( + description="max", layout=widgets.Layout(width="140px"), style={"description_width": "30px"} + ) + self.scale_manual = widgets.HBox( + [self.scale_min, self.scale_max], layout=widgets.Layout(display="none", flex_flow="row wrap") + ) + self.scale_note = widgets.HTML() + self.slice_note = widgets.HTML() + self.cmap_select = widgets.Dropdown( + description="Map:", + value="auto", + options=[("Auto", "auto")] + [(n, n) for n in SEQUENTIAL_CMAPS + DIVERGING_CMAPS], + layout=widgets.Layout(width="190px"), + style={"description_width": "36px"}, + ) + self.render_mode = widgets.ToggleButtons( + options=[("Image", "image"), ("Filled", "filled"), ("Lines", "lines")], + value="image", + tooltips=["One pixel per cell, as computed", "Filled contours", "Labelled contour lines"], + layout=widgets.Layout(width="auto"), + style={"button_width": "auto"}, + ) + self.level_slider = widgets.IntSlider( + description="Levels:", + min=2, + max=30, + value=12, + continuous_update=False, + layout=widgets.Layout(width="230px", display="none"), + style={"description_width": "50px"}, + ) + + self.time_slider = widgets.IntSlider( + description="Time:", + min=0, + max=0, + value=0, + disabled=True, + continuous_update=False, + readout=False, + style={"description_width": "60px"}, + layout=widgets.Layout(flex="1 1 auto", width="auto", min_width="200px"), + ) + self.time_play = widgets.Play(min=0, max=0, value=0, interval=120, disabled=True) + self.time_readout = widgets.HTML(layout=widgets.Layout(padding="0 0 0 14px")) + # The Play button drives the slider in the browser; the slider's observer draws. + widgets.jslink((self.time_play, "value"), (self.time_slider, "value")) + + self.time_bar = widgets.HBox( + [self.time_play, self.time_slider, self.time_readout], + layout=widgets.Layout(width="100%", flex_flow="row wrap", align_items="center"), + ) + + def row(children): + """A line of controls that wraps when the window is narrow.""" + return widgets.HBox( + children, + layout=widgets.Layout(width="100%", flex_flow="row wrap", align_items="center"), + ) + + def panel(children): + """A column that grows into a wide window and wraps on a narrow one.""" + return widgets.VBox( + children, + layout=widgets.Layout(flex=f"1 1 {panel_basis}", min_width=panel_min, padding="0 8px 0 0"), + ) + + self.slice_panel = panel( + [ + self.slice_dropdown, + row( + [ + widgets.HTML("Colour"), + self.scale_mode, + self.scale_manual, + self.cmap_select, + ] + ), + row( + [ + widgets.HTML("Draw"), + self.render_mode, + self.level_slider, + ] + ), + self.scale_note, + self.slice_note, + self.slice_view, + ] + ) + self.curve_panel = panel([self.series_select, self.series_hint, self.curve_view]) + + self.series_select.observe(self._refresh_curves, names="value") + self.slice_dropdown.observe(self._on_slice_change, names="value") + self.time_slider.observe(self._on_time_change, names="value") + self.scale_mode.observe(self._on_scale_mode_change, names="value") + self.scale_min.observe(self._refresh_slice, names="value") + self.scale_max.observe(self._refresh_slice, names="value") + self.cmap_select.observe(self._refresh_slice, names="value") + self.render_mode.observe(self._on_render_change, names="value") + self.level_slider.observe(self._refresh_slice, names="value") + + # -- public ---------------------------------------------------------- + def load(self, path): + """Load a simulation directly, as clicking *Load simulation* would.""" + self.browser.go_to(Path(path).parent if Path(path).is_file() else path) + self.browser.load() + return self.sim + + @property + def field(self): + """The 2D field currently selected, or ``None``.""" + return self.state.field(self.fields) + + @property + def current_time(self): + """Simulation time currently shown by the time bar, or ``None`` without a field.""" + return self.state.time(self.fields) + + @property + def selection(self): + """The series currently ticked in the curve list.""" + return self.state.selected_series(self.series) + + def _ipython_display_(self): + display(self.widget) + + # -- drawing --------------------------------------------------------- + def _sync_state(self): + """Copy the widget values into the shared state.""" + self.state.timestep = self.time_slider.value + self.state.scale_mode = self.scale_mode.value + self.state.manual_limits = (float(self.scale_min.value), float(self.scale_max.value)) + self.state.cmap = self.cmap_select.value + self.state.render = self.render_mode.value + self.state.n_levels = self.level_slider.value + self.state.curves = tuple(self.series_select.value) + + def _refresh_slice(self, _change=None): + if self._updating: + return + self._sync_state() + field = self.field + if field is None: + self.time_readout.value = "" + _clear_surface(self.slice_fig, self.slice_view) + return + + timestep = self.state.clamp(self.fields) + self.time_readout.value = ( + f"t = {field.times[timestep]:.2f} s " + f"(step {timestep + 1} of {len(field.times)})" + ) + + frame = field.frame(timestep) # read once, drawn and possibly scaled from + vmin, vmax, warning = self.state.limits(field, frame=frame) + self.scale_note.value = f"{warning}" if warning else "" + + with _figure_surface(self.slice_fig, self.slice_view, self.slice_figsize) as fig: + ax = fig.add_subplot(111) + plot_slice( + field, + timestep, + ax=ax, + vmin=vmin, + vmax=vmax, + cmap=self.state.colormap(field), + render=self.state.render, + n_levels=self.state.n_levels, + frame=frame, + ) + fig.tight_layout() + + def _refresh_curves(self, _change=None): + if self._updating: + return + self._sync_state() + selected = self.selection + if not selected: + _clear_surface(self.curve_fig, self.curve_view) + return + with _figure_surface(self.curve_fig, self.curve_view, self.curve_figsize) as fig: + ax = fig.add_subplot(111) + plot_series(selected, ax=ax, cursor_time=self.current_time) + fig.tight_layout() + + # -- callbacks ------------------------------------------------------- + def _on_time_change(self, _change=None): + """The time bar moves the slice and the marker on the curve plot together.""" + if self._updating: + return + self._refresh_slice() + self._refresh_curves() + + def _on_render_change(self, _change=None): + """The level count only matters for the contour modes.""" + self.level_slider.layout.display = "none" if self.render_mode.value == "image" else "flex" + self._refresh_slice() + + def _on_scale_mode_change(self, _change=None): + manual = self.scale_mode.value == "manual" + self.scale_manual.layout.display = "flex" if manual else "none" + field = self.field + if manual and field is not None and not self._updating: + # Start from what is on screen, so the fields are a useful starting point. + self._sync_state() + self.state.scale_mode = "global" + low, high, _ = self.state.limits(field) + self._updating = True + self.scale_min.value, self.scale_max.value = round(low, 4), round(high, 4) + self._updating = False + self._refresh_slice() + + def _on_slice_change(self, _change=None): + if self._updating: + return + index = self.slice_dropdown.value + if index is None: + return + + previous = self.field + if previous is not None: + previous.clear_cache() # keep only one field's data in memory + + self.state.field_index = index + field = self.field + if field is None: + self.slice_note.value = "Nothing to show." + _clear_surface(self.slice_fig, self.slice_view) + return + self.slice_note.value = "" + + self._updating = True + last = len(field.times) - 1 + self.time_slider.max = self.time_play.max = last + self.time_slider.value = self.time_play.value = min(self.time_slider.value, last) + self.time_slider.disabled = self.time_play.disabled = False + self._updating = False + + self._on_time_change() + + # -- loading --------------------------------------------------------- + def _setup_fields(self, sim): + """Fill the field dropdown and the time bar for a freshly loaded simulation.""" + self.fields = fields_of(sim) + self.state.field_index = 0 + self.state.timestep = 0 + + self._updating = True + self.slice_dropdown.options = [(f.label, i) for i, f in enumerate(self.fields)] + self.slice_dropdown.disabled = not self.fields + self.time_slider.disabled = self.time_play.disabled = not self.fields + if not self.fields: + self.time_slider.max = self.time_play.max = 0 + self._updating = False + + _clear_surface(self.slice_fig, self.slice_view) + if not self.fields: + self.slice_note.value = "" + self.time_readout.value = "This simulation has no slice (SLCF) output." + return + + self._updating = True + self.slice_dropdown.index = 0 + self._updating = False + self._on_slice_change() + + def _on_simulation_loaded(self, sim): + """Populate the curve list, the slices and the overview once a simulation is read.""" + self.sim = sim + self.devices = list_devices(sim) + self.series = build_series(sim) + self.state.curves = (0,) if self.series else () + + self._updating = True + self.series_select.options = [(e["label"], i) for i, e in enumerate(self.series)] + self.series_select.disabled = not self.series + self.series_select.value = (0,) if self.series else () + self._updating = False + + if not self.series: + self.summary.value = "This simulation has no device (DEVC) or HRR output." + _clear_surface(self.curve_fig, self.curve_view) + self._setup_fields(sim) + return + + self.summary.value = self._overview() + self._setup_fields(sim) # also positions the time bar + self._refresh_curves() + + def _overview(self): + """Short HTML summary: what is in this simulation, at a glance.""" + n_hrr = sum(1 for e in self.series if e["source"] == "hrr") + counts = [] + if self.devices: + counts.append(f"{len(self.devices)} devices") + if n_hrr: + counts.append(f"{n_hrr} HRR quantities") + if self.fields: + counts.append(f"{len(self.fields)} slices") + + # Devices and the HRR file are not necessarily written at the same interval, so + # report the span covered by everything rather than one series' sample count. + starts = [e["times"][0] for e in self.series if len(e["times"])] + ends = [e["times"][-1] for e in self.series if len(e["times"])] + if starts: + counts.append(f"t = {min(starts):g} … {max(ends):g} s") + + groups = {} + for device in self.devices: + groups.setdefault(quantity_of(device) or "—", []).append(unit_of(device)) + if n_hrr: + groups["HRR file"] = [e["unit"] for e in self.series if e["source"] == "hrr"] + + rows = "".join( + f"{name}" + f"{len(units)}" + f"" + f"{', '.join(sorted(set(units) - {''})) or '—'}" + for name, units in sorted(groups.items()) + ) + table = ( + f""" + + + + + {rows} +
Quantity#Unit
""" + if rows + else "" + ) + + return f""" +
+ {self.sim.chid}  ·  {"  ·  ".join(counts)}{table} +
""" + + +def explore(path=None, **kwargs): + """Show the explorer and return it. + + ``explore()`` opens a directory browser; ``explore("path/to/case")`` loads that + simulation straight away. Keyword arguments are passed to :class:`NotebookExplorer`. + """ + explorer = NotebookExplorer(path=path, **kwargs) + display(explorer.widget) + return explorer diff --git a/fdsreader/explorer/plots.py b/fdsreader/explorer/plots.py new file mode 100644 index 0000000..dc6be5b --- /dev/null +++ b/fdsreader/explorer/plots.py @@ -0,0 +1,191 @@ +"""Matplotlib drawing for the explorer. + +These are plain functions taking an axes, so they can be used on their own to build +figures for a report without any of the widget machinery. +""" + +import matplotlib.pyplot as plt +import numpy as np + +from .data import axis_label, device_label, device_time, quantity_of, unit_of + +#: Colour maps offered for slices. "Auto" picks a one-sided map for quantities such as +#: temperature and a zero-centred one for signed quantities such as velocities. +SEQUENTIAL_CMAPS = [ + "inferno", + "magma", + "plasma", + "viridis", + "cividis", + "turbo", + "hot", + "afmhot", + "gist_heat", + "Greys_r", +] +DIVERGING_CMAPS = ["RdBu_r", "coolwarm", "bwr", "seismic", "PuOr_r", "PRGn_r", "Spectral_r"] + + +def plot_series(series, ax=None, cursor_time=None): + """Plot one or more time series against simulation time. + + Series whose unit differs from the first one are drawn against a second y-axis on the + right, so a heat release rate in kW and a temperature in C can share a figure. + ``cursor_time`` draws a marker at one instant, used to tie the curves to a slice. + """ + if ax is None: + _, ax = plt.subplots() + + if not series: + ax.set_title("Nothing selected") + ax.set_xlabel("Time [s]") + return ax + + units = [] + for entry in series: + if entry["unit"] not in units: + units.append(entry["unit"]) + + # One extra axis is readable; beyond that the remaining units share the right-hand one + # and the axis label names them all. + right = ax.twinx() if len(units) > 1 else None + if right is not None: + right.grid(False) + + colors = plt.rcParams["axes.prop_cycle"].by_key()["color"] + left_group, right_group, handles = [], [], [] + + for i, entry in enumerate(series): + on_left = right is None or entry["unit"] == units[0] + target = ax if on_left else right + (left_group if on_left else right_group).append(entry) + (line,) = target.plot( + entry["times"], + entry["values"], + lw=1.5, + color=colors[i % len(colors)], + label=entry["label"], + ) + handles.append(line) + + ax.set_ylabel(axis_label(left_group)) + if right_group: + right.set_ylabel(axis_label(right_group)) + + single = series[0] if len(series) == 1 else None + + if single is not None and single["device"] is not None: + device, times = single["device"], single["times"] + # FDS also writes the initial state of every device at t=0; that is not an event, + # so it would otherwise put a spurious marker on every single plot. + for act_time, state in getattr(device, "activation_times", []): + if not len(times) or act_time <= times[0]: + continue + ax.axvline(act_time, color="tab:red", ls="--", lw=1, alpha=0.8) + ax.annotate( + "activated" if state else "deactivated", + xy=(act_time, 1.0), + xycoords=("data", "axes fraction"), + xytext=(4, -12), + textcoords="offset points", + fontsize=8, + color="tab:red", + ) + + if cursor_time is not None: + ax.axvline(cursor_time, color="0.35", lw=1.2, alpha=0.7) + + ax.set_xlabel("Time [s]") + ax.margins(x=0.01) + + if single is not None: + title = single["label"] + if single["device"] is not None: + x, y, z = single["device"].position + title += f" · position ({x:g}, {y:g}, {z:g}) m" + # The range goes on a second title line rather than floating inside the axes, + # where it would sit on top of either the title or the data. + finite = single["values"][np.isfinite(single["values"])] + if finite.size: + title += f"\nmin {finite.min():.4g} · max {finite.max():.4g} · final {finite[-1]:.4g}" + ax.set_title(title, fontsize=10) + else: + ax.set_title(f"{len(series)} series", fontsize=10) + ax.legend(handles, [h.get_label() for h in handles], fontsize=8, loc="best", framealpha=0.85) + + return ax + + +def plot_device(sim, device, ax=None, cursor_time=None): + """Plot a single device against simulation time.""" + return plot_series( + [ + { + "label": device_label(device), + "name": device.id, + "times": device_time(sim, device), + "values": np.asarray(device.data), + "quantity": quantity_of(device), + "unit": unit_of(device), + "source": "device", + "device": device, + } + ], + ax=ax, + cursor_time=cursor_time, + ) + + +def plot_slice(field, timestep, ax=None, vmin=None, vmax=None, cmap=None, render="image", n_levels=12, frame=None): + """Draw one time step of a 2D field. + + ``vmin``/``vmax`` override the field's own range, which is what the per-step and + manual scaling modes use. ``render`` is ``"image"`` (one pixel per cell), ``"filled"`` + (filled contours) or ``"lines"`` (labelled contour lines). Pass ``frame`` to reuse an + array that has already been read. + """ + if ax is None: + _, ax = plt.subplots() + low, high, _ = field.value_range() + vmin = low if vmin is None else vmin + vmax = high if vmax is None else vmax + cmap = field.cmap if cmap is None else cmap + if frame is None: + frame = field.frame(timestep) + + horizontal, vertical = field.axes + x, y = field.coords[horizontal], field.coords[vertical] + + if render == "image": + mappable = ax.imshow( + frame.T, # (dim1, dim2) -> rows must run along the upright axis + origin="lower", + extent=[x[0], x[-1], y[0], y[-1]], + aspect="equal", + cmap=cmap, + vmin=vmin, + vmax=vmax, + interpolation="nearest", # show the cells as computed, do not smooth them away + ) + else: + # Contours are drawn on fixed levels spanning the colour limits, so that changing + # the scaling mode moves the contours with it. "both" keeps values outside the + # range coloured rather than blank. + levels = np.linspace(vmin, vmax, max(2, n_levels) + 1) + if render == "filled": + mappable = ax.contourf(x, y, frame.T, levels=levels, cmap=cmap, extend="both") + else: + mappable = ax.contour(x, y, frame.T, levels=levels, cmap=cmap, linewidths=1.0, vmin=vmin, vmax=vmax) + ax.clabel(mappable, inline=True, fontsize=7, fmt="%.4g") + ax.set_aspect("equal") + ax.set_xlim(x[0], x[-1]) + ax.set_ylim(y[0], y[-1]) + + bar = ax.figure.colorbar(mappable, ax=ax, fraction=0.046, pad=0.04) + bar.set_label(f"{field.quantity} [{field.unit}]" if field.unit else field.quantity) + + ax.set_xlabel(f"{horizontal} [m]") + ax.set_ylabel(f"{vertical} [m]") + ax.set_title(f"{field.label} · t = {field.times[timestep]:.2f} s", fontsize=10) + ax.grid(False) + return ax diff --git a/fdsreader/explorer/render.py b/fdsreader/explorer/render.py new file mode 100644 index 0000000..4f5216c --- /dev/null +++ b/fdsreader/explorer/render.py @@ -0,0 +1,164 @@ +"""Terminal rendering backend: drawing slices and time series with characters. + +Needs nothing but numpy, so the command line front end stays usable in an installation +that has neither matplotlib nor ipywidgets. +""" + +import shutil + +import numpy as np + +#: Characters from empty to full. The default ramp is deliberately short: a terminal +#: without colour can only distinguish a handful of levels anyway. +RAMP = " .:-=+*#%@" + +#: A character cell is about twice as tall as it is wide. Horizontal counts are scaled by +#: this so that a slice keeps the aspect ratio it has in metres. +CELL_ASPECT = 2.0 + + +def terminal_size(fallback=(100, 30)): + """Width and height of the terminal, falling back when it is not a tty.""" + size = shutil.get_terminal_size(fallback) + return size.columns, size.lines + + +def block_mean(values, rows, cols): + """Area-average a 2D array down to a character grid. + + Averaging rather than sampling keeps thin hot structures visible when a slice is much + finer than the terminal. + """ + row_edges = np.linspace(0, values.shape[0], rows + 1).astype(int) + col_edges = np.linspace(0, values.shape[1], cols + 1).astype(int) + out = np.empty((rows, cols)) + for i in range(rows): + r0, r1 = row_edges[i], max(row_edges[i] + 1, row_edges[i + 1]) + for j in range(cols): + c0, c1 = col_edges[j], max(col_edges[j] + 1, col_edges[j + 1]) + out[i, j] = np.nanmean(values[r0:r1, c0:c1]) + return out + + +def shape_for(field, height, width): + """Character grid for a field that keeps its physical aspect ratio. + + Returns ``(rows, cols)`` fitting inside ``height`` x ``width`` characters. + """ + horizontal, vertical = field.axes + xs, ys = field.coords[horizontal], field.coords[vertical] + span_h = float(xs[-1] - xs[0]) or 1.0 + span_v = float(ys[-1] - ys[0]) or 1.0 + + rows = max(4, height) + cols = int(round(span_h / span_v * rows * CELL_ASPECT)) + if cols > width: # too wide for the terminal, so scale the other way round + cols = max(4, width) + rows = max(4, int(round(span_v / span_h * cols / CELL_ASPECT))) + return rows, max(4, cols) + + +def slice_lines(field, timestep, vmin, vmax, height=26, width=None, ramp=RAMP, frame=None): + """Render one time step of a 2D field as text lines, axes included.""" + if width is None: + width = max(20, terminal_size()[0] - 10) + + horizontal, vertical = field.axes + xs, ys = field.coords[horizontal], field.coords[vertical] + rows, cols = shape_for(field, height, width) + + if frame is None: + frame = field.frame(timestep) + # data is (dim1, dim2); transpose so that rows run along the upright axis + grid = block_mean(frame.T, rows, cols) + + span = (vmax - vmin) or 1.0 + index = np.clip(((grid - vmin) / span * (len(ramp) - 1)).round().astype(int), 0, len(ramp) - 1) + + lines = [] + for i in range(rows - 1, -1, -1): # origin at the bottom + if i % 5 == 0 or i == rows - 1: + coord = ys[0] + (ys[-1] - ys[0]) * i / max(1, rows - 1) + label = f"{coord:7.2f} " + else: + label = " " * 8 + lines.append(label + "|" + "".join(ramp[j] for j in index[i])) + + lines.append(" " * 8 + "+" + "-" * cols) + left, right = f"{xs[0]:g}", f"{xs[-1]:g}" + pad = max(1, cols - len(left) - len(right)) + lines.append(" " * 9 + left + " " * pad + right + f" {horizontal} [m]") + lines.append(" " * 9 + f"scale {vmin:.4g} … {vmax:.4g} ramp '{ramp}' grid {cols}x{rows}") + return lines + + +def series_lines(series, height=16, width=None, cursor_time=None): + """Render one or more time series as text lines. + + Several series share the plot; each gets its own marker, and the y-axis is only + labelled with real numbers when they share a unit. + """ + if not series: + return ["(nothing selected)"] + if width is None: + width = max(30, terminal_size()[0] - 12) + + markers = "*o+x#~" + t_start = min(float(s["times"][0]) for s in series) + t_end = max(float(s["times"][-1]) for s in series) + t_span = (t_end - t_start) or 1.0 + + units = {s["unit"] for s in series} + shared_scale = len(units) == 1 + if shared_scale: + low = min(float(np.nanmin(s["values"])) for s in series) + high = max(float(np.nanmax(s["values"])) for s in series) + else: # mixed units: each series is scaled to its own range, so only shapes compare + low, high = 0.0, 1.0 + + canvas = [[" "] * width for _ in range(height)] + for n, entry in enumerate(series): + values = np.asarray(entry["values"], dtype=float) + if shared_scale: + lo, hi = low, high + else: + lo = float(np.nanmin(values)) + hi = float(np.nanmax(values)) + span = (hi - lo) or 1.0 + + xs = np.clip( + ((np.asarray(entry["times"], dtype=float) - t_start) / t_span * (width - 1)).round().astype(int), + 0, + width - 1, + ) + ys = np.clip(((values - lo) / span * (height - 1)).round().astype(int), 0, height - 1) + mark = markers[n % len(markers)] + for x, y in zip(xs, ys): + canvas[y][x] = mark + + if cursor_time is not None: + column = int(round((float(cursor_time) - t_start) / t_span * (width - 1))) + if 0 <= column < width: + for row in canvas: + if row[column] == " ": + row[column] = "|" + + lines = [] + for r in range(height - 1, -1, -1): + if shared_scale: + label = f"{low + (high - low) * r / max(1, height - 1):10.4g} |" + else: + label = f"{r / max(1, height - 1):10.2f} |" + lines.append(label + "".join(canvas[r])) + + lines.append(" " * 10 + "+" + "-" * width) + left, right = f"{t_start:g}", f"{t_end:g}" + pad = max(1, width - len(left) - len(right)) + lines.append(" " * 11 + left + " " * pad + right + " t [s]") + + for n, entry in enumerate(series): + note = "" if shared_scale else " (scaled to its own range)" + lines.append(f" {markers[n % len(markers)]} {entry['label']}{note}") + if not shared_scale: + lines.append(" y-axis is relative: the selected series do not share a unit") + return lines diff --git a/fdsreader/explorer/state.py b/fdsreader/explorer/state.py new file mode 100644 index 0000000..18144a7 --- /dev/null +++ b/fdsreader/explorer/state.py @@ -0,0 +1,96 @@ +"""What the user is currently looking at, independent of which front end shows it. + +Keeping the selection *and the rules that act on it* here is what stops the front ends +drifting apart: the notebook and the command line ask the same object for the colour +limits, so "per step" means the same thing in both, and a third front end inherits the +behaviour rather than reimplementing it. +""" + +from dataclasses import dataclass +from dataclasses import field as _dataclass_field +from typing import Tuple + +import numpy as np + +from .data import color_limits + +SCALE_MODES = ("global", "step", "manual") +RENDER_MODES = ("image", "filled", "lines") + + +@dataclass +class ExplorerState: + """The current selection. Front ends read and write it; nothing here draws.""" + + field_index: int = 0 + timestep: int = 0 + scale_mode: str = "global" + manual_limits: Tuple[float, float] | None = None + cmap: str = "auto" + render: str = "image" + n_levels: int = 12 + curves: Tuple[int, ...] = _dataclass_field(default_factory=tuple) + + # -- selection ------------------------------------------------------- + def field(self, fields): + """The selected field, or ``None`` when there is nothing to show.""" + if not fields: + return None + self.field_index = max(0, min(self.field_index, len(fields) - 1)) + return fields[self.field_index] + + def clamp(self, fields): + """Keep the time step inside the selected field.""" + current = self.field(fields) + if current is None: + self.timestep = 0 + else: + self.timestep = max(0, min(self.timestep, len(current.times) - 1)) + return self.timestep + + def time(self, fields): + """Simulation time currently selected, in seconds, or ``None``.""" + current = self.field(fields) + if current is None or not len(current.times): + return None + return float(current.times[self.clamp(fields)]) + + def seek(self, fields, seconds): + """Move to the time step nearest to ``seconds``.""" + current = self.field(fields) + if current is None or not len(current.times): + return 0 + self.timestep = int(np.argmin(np.abs(current.times - float(seconds)))) + return self.timestep + + # -- drawing rules --------------------------------------------------- + def limits(self, field, frame=None): + """Colour limits for the frame on screen. + + Returns ``(vmin, vmax, warning)``; *warning* is a message when the request could + not be honoured and the global scale was used instead. + """ + low, high, diverging = field.value_range() + + if self.scale_mode == "manual": + if self.manual_limits is not None: + lo, hi = (float(v) for v in self.manual_limits) + if np.isfinite(lo) and np.isfinite(hi) and lo < hi: + return lo, hi, None + return low, high, "min must be below max — showing the global scale" + + if self.scale_mode == "step": + if frame is None: + frame = field.frame(self.timestep) + lo, hi, _ = color_limits(frame, diverging) + return lo, hi, None + + return low, high, None + + def colormap(self, field): + """The colour map to draw with, resolving ``"auto"`` against the field.""" + return field.cmap if self.cmap == "auto" else self.cmap + + def selected_series(self, series): + """The series ticked in the curve list, dropping any stale indices.""" + return [series[i] for i in self.curves if 0 <= i < len(series)] diff --git a/pyproject.toml b/pyproject.toml index a38f361..521c754 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,9 @@ dependencies = [ [project.urls] Repository = "https://github.com/FireDynamics/fdsreader" +[project.scripts] +fdsreader-explorer-cli = "fdsreader.explorer.cli:main" + [project.optional-dependencies] dev = [ "pytest>=7.0", @@ -41,6 +44,10 @@ docs = [ "furo", "autodocsumm", ] +notebook = [ + "matplotlib>=3.5", + "ipywidgets>=8", +] [tool.setuptools_scm] version_scheme = "post-release" @@ -52,6 +59,7 @@ packages = [ "fdsreader.bndf", "fdsreader.devc", "fdsreader.evac", + "fdsreader.explorer", "fdsreader.export", "fdsreader.fds_classes", "fdsreader.geom", diff --git a/tests/explorer/conftest.py b/tests/explorer/conftest.py new file mode 100644 index 0000000..e745fd4 --- /dev/null +++ b/tests/explorer/conftest.py @@ -0,0 +1,99 @@ +"""Fixtures for the explorer tests. + +The cases are written from scratch rather than committed as binary fixtures, so the tests +stay small and readable and do not depend on an FDS build. +""" + +import csv + +import numpy as np +import pytest + + +@pytest.fixture +def tiny_case(tmp_path): + """A simulation with devices and an HRR file, written out on the fly.""" + root = tmp_path / "tiny" + root.mkdir() + + (root / "tiny.smv").write_text( + "CHID\n tiny\n\n" + "CSVF\n devc\n tiny_devc.csv\n\n" + "CSVF\n hrr\n tiny_hrr.csv\n\n" + "DEVICE\n TC_1 % TEMPERATURE\n 1.0 0.0 1.0 0.0 0.0 1.0\n\n" + "DEVICE\n TC_2 % TEMPERATURE\n 2.0 0.0 2.0 0.0 0.0 1.0\n\n" + "DEVICE_ACT\n 2 5.00 1\n" + ) + + times = np.arange(0.0, 10.5, 0.5) + with open(root / "tiny_devc.csv", "w", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(["s", "C", "C"]) + writer.writerow(["Time", "TC_1", "TC_2"]) + for t, a, b in zip(times, 20 + 8 * times, 20 + 3 * times): + writer.writerow([f"{t:.4f}", f"{a:.4f}", f"{b:.4f}"]) + + with open(root / "tiny_hrr.csv", "w", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(["s", "kW"]) + writer.writerow(["Time", "HRR"]) + for t in times: + writer.writerow([f"{t:.4f}", f"{100 * t:.4f}"]) + + return root + + +class FakeField: + """A 2D field standing in for a slice, so rendering can be tested without FDS files.""" + + kind = "slice" + drawable = True + + def __init__(self, nx=8, ny=16, n_t=3, quantity="TEMPERATURE", unit="C"): + self.label = f"{quantity} [{unit}] — x = 0 m" + self.quantity = quantity + self.unit = unit + self.times = np.linspace(0.0, 2.0, n_t) + self.axes = ("y", "z") + self.coords = {"x": np.array([0.0]), + "y": np.linspace(-2.0, 2.0, nx), + "z": np.linspace(0.0, 8.0, ny)} + self.shape = (nx, ny) + # a blob that rises and grows, so frames differ both in where the heat is and + # in how hot it gets -- which is what makes a per-step colour scale visible + yy, zz = np.meshgrid(self.coords["y"], self.coords["z"], indexing="ij") + self._frames = [20 + (150 + 180 * k) * np.exp(-(yy ** 2) - (zz - 1 - 2 * k) ** 2) + for k in range(n_t)] + + def frame(self, timestep): + return self._frames[timestep] + + def value_range(self): + low = min(float(f.min()) for f in self._frames) + high = max(float(f.max()) for f in self._frames) + return low, high, False + + @property + def cmap(self): + return "inferno" + + def clear_cache(self): + pass + + +@pytest.fixture +def fake_field(): + return FakeField() + + +@pytest.fixture +def fake_series(): + times = np.linspace(0.0, 4.0, 21) + return [ + {"label": "DEVC A — TEMPERATURE [C]", "name": "A", "times": times, + "values": 20 + 10 * times, "quantity": "TEMPERATURE", "unit": "C", + "source": "device", "device": None}, + {"label": "DEVC B — TEMPERATURE [C]", "name": "B", "times": times, + "values": 20 + 4 * times ** 2, "quantity": "TEMPERATURE", "unit": "C", + "source": "device", "device": None}, + ] diff --git a/tests/explorer/test_cli.py b/tests/explorer/test_cli.py new file mode 100644 index 0000000..fdaa493 --- /dev/null +++ b/tests/explorer/test_cli.py @@ -0,0 +1,116 @@ +"""The command line front end, driven the way a user drives it.""" + +import json + +import pytest + +from fdsreader.explorer.cli import main + + +def run(capsys, *argv): + """Run the CLI and return (exit status, stdout, stderr).""" + status = main([str(a) for a in argv]) + captured = capsys.readouterr() + return status, captured.out, captured.err + + +def test_overview(tiny_case, capsys): + status, out, _ = run(capsys, tiny_case) + assert status == 0 + assert "tiny" in out + assert "devices (DEVC) 2" in out + assert "HRR quantities 1" in out + assert "slices (SLCF) 0" in out + assert "time 0 … 10 s" in out + assert "TEMPERATURE 2 C" in out + + +def test_list_names_what_can_be_asked_for(tiny_case, capsys): + _, out, _ = run(capsys, tiny_case, "--list") + assert "slices:\n (none)" in out + assert "--curve TC_1" in out + assert "--curve HRR" in out + + +def test_curve_is_drawn_with_axes_and_a_legend(tiny_case, capsys): + _, out, _ = run(capsys, tiny_case, "--curve", "TC_1", "--width", "20", "--height", "12") + lines = out.splitlines() + # TC_1 rises linearly from 20 to 100 C, so the top row is labelled with the maximum + # and carries ink only at the right-hand end + assert lines[0].startswith(" 100 |") + assert lines[0].rstrip().endswith("*") + assert lines[0].split("|", 1)[1].startswith(" ") + assert lines[-4].startswith(" 20 |*") # and the bottom row starts at t = 0 + assert lines[-3].startswith(" +---") # then the axis rule + assert "t [s]" in lines[-2] + assert lines[-1] == " * DEVC TC_1 — TEMPERATURE [C]" + + +def test_two_curves_share_one_scale_when_units_match(tiny_case, capsys): + _, out, _ = run(capsys, tiny_case, "--curve", "TC_1", "--curve", "TC_2") + assert "scaled to its own range" not in out + assert out.rstrip().endswith("o DEVC TC_2 — TEMPERATURE [C]") + + +def test_mixed_units_are_flagged(tiny_case, capsys): + _, out, _ = run(capsys, tiny_case, "--curve", "TC_1", "--curve", "HRR") + assert "y-axis is relative" in out + + +def test_cursor_is_drawn_at_the_requested_time(tiny_case, capsys): + _, without, _ = run(capsys, tiny_case, "--curve", "TC_1", "--width", "20") + _, with_cursor, _ = run(capsys, tiny_case, "--curve", "TC_1", "--width", "20", + "--time", "5") + assert with_cursor.count("|") > without.count("|") + + +def test_json_is_machine_readable(tiny_case, capsys): + status, out, _ = run(capsys, tiny_case, "--json") + assert status == 0 + payload = json.loads(out) + assert payload["chid"] == "tiny" + assert payload["devices"] == 2 + assert payload["slices"] == [] + names = {curve["name"] for curve in payload["curves"]} + assert names == {"TC_1", "TC_2", "HRR"} + tc1 = next(c for c in payload["curves"] if c["name"] == "TC_1") + assert tc1["min"] == pytest.approx(20.0) + assert tc1["max"] == pytest.approx(100.0) + assert tc1["unit"] == "C" + + +def test_json_includes_the_selected_curves(tiny_case, capsys): + _, out, _ = run(capsys, tiny_case, "--json", "--curve", "HRR") + payload = json.loads(out) + assert [s["name"] for s in payload["selected"]] == ["HRR"] + assert len(payload["selected"][0]["values"]) == 21 + + +def test_unknown_curve_is_reported(tiny_case, capsys): + with pytest.raises(SystemExit) as raised: + run(capsys, tiny_case, "--curve", "NOPE") + assert "no curve called 'NOPE'" in str(raised.value) + + +def test_asking_for_a_slice_when_there_are_none(tiny_case, capsys): + status, _, err = run(capsys, tiny_case, "--slice", "0") + assert status == 1 + assert "no slice output" in err + + +def test_bad_path_is_reported_on_stderr(tmp_path, capsys): + status, out, err = run(capsys, tmp_path / "nowhere") + assert status == 1 + assert out == "" + assert "could not load" in err + + +@pytest.mark.parametrize("scale", ["sideways", "5", "10,5", "a,b"]) +def test_bad_scale_is_rejected(tiny_case, capsys, scale): + with pytest.raises(SystemExit): + run(capsys, tiny_case, "--curve", "TC_1", "--scale", scale) + + +def test_caching_is_off_unless_asked_for(tiny_case, capsys): + run(capsys, tiny_case) + assert not list(tiny_case.glob("*.pickle")) diff --git a/tests/explorer/test_render.py b/tests/explorer/test_render.py new file mode 100644 index 0000000..f22101c --- /dev/null +++ b/tests/explorer/test_render.py @@ -0,0 +1,106 @@ +"""The terminal rendering is pinned to its exact output. + +Text is easy to compare, so these read as pictures of what the CLI prints; a change in +layout shows up as a readable diff rather than a number that no longer matches. +""" + +import numpy as np + +from fdsreader.explorer.render import block_mean, series_lines, shape_for, slice_lines + + +def ink(lines): + """How many characters of a drawing are not blank.""" + return sum(ch not in " |" for line in lines for ch in line) + + +def test_block_mean_averages_rather_than_samples(): + values = np.arange(16, dtype=float).reshape(4, 4) + assert block_mean(values, 2, 2).tolist() == [[2.5, 4.5], [10.5, 12.5]] + + +def test_block_mean_handles_upsampling_without_empty_blocks(): + out = block_mean(np.array([[1.0, 2.0]]), 3, 4) + assert out.shape == (3, 4) + assert np.isfinite(out).all() + + +def test_shape_for_keeps_the_physical_aspect_ratio(fake_field): + # the field is 4 m wide and 8 m tall, and a character cell is twice as tall as wide + assert shape_for(fake_field, 20, 200) == (20, 20) + + +def test_shape_for_fits_a_narrow_terminal(fake_field): + rows, cols = shape_for(fake_field, 40, 12) + assert cols <= 12 + assert rows >= 4 + + +def test_slice_lines(fake_field): + low, high, _ = fake_field.value_range() + assert slice_lines(fake_field, 1, low, high, height=10, width=40) == [ + " 8.00 | ", + " | ", + " | ", + " | ", + " 4.44 | ..... ", + " | .:+++:. ", + " | .:===:. ", + " | ... ", + " | ", + " 0.00 | ", + " +----------", + " -2 2 y [m]", + " scale 20 … 471.6 ramp ' .:-=+*#%@' grid 10x10", + ] + + +def test_slice_lines_moves_with_time(fake_field): + low, high, _ = fake_field.value_range() + first = slice_lines(fake_field, 0, low, high, height=10, width=40)[:10] + last = slice_lines(fake_field, 2, low, high, height=10, width=40)[:10] + assert first != last + # the blob rises, so its ink sits lower in the first frame than in the last + weight = lambda rows: sum(i for i, row in enumerate(rows) if row.strip("| ")) + assert weight(first) > weight(last) # row 0 is the top of the picture + + +def test_slice_lines_follows_the_colour_limits(fake_field): + wide = slice_lines(fake_field, 1, 0.0, 5000.0, height=6, width=20)[:6] + tight = slice_lines(fake_field, 1, 20.0, 60.0, height=6, width=20)[:6] + assert ink(wide) < ink(tight) + assert "@" in "".join(tight) + assert "@" not in "".join(wide) + + +def test_series_lines_shared_unit(fake_series): + assert series_lines(fake_series, height=6, width=24, cursor_time=2.0) == [ + " 84 | | oo", + " 71.2 | | oo ", + " 58.4 | | oo ****", + " 45.6 | *oooo** ", + " 32.8 | ***ooo o ", + " 20 |oooo ooo | ", + " +------------------------", + " 0 4 t [s]", + " * DEVC A — TEMPERATURE [C]", + " o DEVC B — TEMPERATURE [C]", + ] + + +def test_series_lines_without_a_cursor_has_no_rule(fake_series): + assert "|" not in "".join( + line.split("|", 1)[1] for line in + series_lines(fake_series, height=6, width=24)[:6] + ) + + +def test_series_lines_mixed_units_say_so(fake_series): + mixed = [fake_series[0], dict(fake_series[1], unit="kW", quantity="HRR")] + lines = series_lines(mixed, height=5, width=20) + assert "scaled to its own range" in lines[-3] + assert lines[-1].startswith(" y-axis is relative") + + +def test_series_lines_with_nothing_selected(): + assert series_lines([]) == ["(nothing selected)"] diff --git a/tests/explorer/test_state.py b/tests/explorer/test_state.py new file mode 100644 index 0000000..d08fed6 --- /dev/null +++ b/tests/explorer/test_state.py @@ -0,0 +1,70 @@ +"""The rules every front end shares: which frame, and on what colour scale.""" + +import numpy as np +import pytest + +from fdsreader.explorer.state import ExplorerState + + +def test_field_is_clamped_to_what_exists(fake_field): + state = ExplorerState(field_index=7) + assert state.field([fake_field]) is fake_field + assert state.field_index == 0 + assert state.field([]) is None + + +def test_timestep_is_clamped(fake_field): + state = ExplorerState(timestep=99) + assert state.clamp([fake_field]) == len(fake_field.times) - 1 + + +def test_seek_picks_the_nearest_time(fake_field): + state = ExplorerState() + assert state.seek([fake_field], 0.9) == 1 # times are 0.0, 1.0, 2.0 + assert state.time([fake_field]) == pytest.approx(1.0) + assert state.seek([fake_field], 99.0) == 2 # past the end lands on the last step + + +def test_global_scale_is_the_same_for_every_step(fake_field): + state = ExplorerState(scale_mode="global") + first = state.limits(fake_field) + state.timestep = 2 + assert state.limits(fake_field) == first + assert first[2] is None + + +def test_step_scale_follows_the_frame(fake_field): + state = ExplorerState(scale_mode="step", timestep=0) + low, high, warning = state.limits(fake_field) + assert warning is None + assert high < fake_field.value_range()[1] # one frame is cooler than the whole run + + +def test_step_scale_reuses_a_frame_that_was_already_read(fake_field): + state = ExplorerState(scale_mode="step") + frame = np.full(fake_field.shape, 100.0) + low, high, _ = state.limits(fake_field, frame=frame) + assert (low, high) == (99.5, 100.5) # a flat frame still gets a finite range + + +def test_manual_scale_is_used_when_it_is_valid(fake_field): + state = ExplorerState(scale_mode="manual", manual_limits=(10.0, 30.0)) + assert state.limits(fake_field) == (10.0, 30.0, None) + + +@pytest.mark.parametrize("limits", [(30.0, 10.0), (5.0, 5.0), None, (np.nan, 1.0)]) +def test_manual_scale_falls_back_and_says_why(fake_field, limits): + state = ExplorerState(scale_mode="manual", manual_limits=limits) + low, high, warning = state.limits(fake_field) + assert (low, high) == fake_field.value_range()[:2] + assert "min must be below max" in warning + + +def test_auto_colormap_defers_to_the_field(fake_field): + assert ExplorerState(cmap="auto").colormap(fake_field) == fake_field.cmap + assert ExplorerState(cmap="viridis").colormap(fake_field) == "viridis" + + +def test_selected_series_ignores_stale_indices(fake_series): + state = ExplorerState(curves=(1, 99, -1)) + assert state.selected_series(fake_series) == [fake_series[1]] From 36a2039e05284249fc8ce764096feb73c396c6f9 Mon Sep 17 00:00:00 2001 From: Lukas Date: Mon, 21 Sep 2026 10:49:41 +0200 Subject: [PATCH 2/4] docs(readme): document the explorer Adds an Explorer section covering both front ends -- the Jupyter one and the command line one -- and mentions the optional "notebook" extra in the installation instructions. --- README.md | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/README.md b/README.md index 2517af6..3537544 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,12 @@ The package is available on PyPI and can be installed using pip: ```sh pip install fdsreader ``` + +The interactive [explorer](#explorer) needs a plotting stack, which is an optional extra: +```sh +pip install "fdsreader[notebook]" +``` + _FDS Version 6.7.5 and above are fully supported. Versions below 6.7.5 might work, but are not guaranteed to work._ ## Usage example @@ -60,6 +66,89 @@ fds.settings.DEBUG = True Beware that not all attributes and methods are covered in this diagram. For a complete documentation of all classes check the API Documentation below. +## Explorer + +`fdsreader.explorer` looks at a simulation interactively instead of writing plotting code +for it: a time bar driving a 2D slice, and any number of device or HRR quantities plotted +against time. + +### In Jupyter + +```python +from fdsreader.explorer import explore + +explorer = explore() +``` + +That renders the whole interface — a directory browser, the time bar, the slice on the +left and the curves on the right. Browse to a case and press *Load simulation*. + +```python +explore("./sample_data") # skip the browser and load this case +explore(start_dir="/data/cases") # open the browser somewhere specific +explore(interactive_canvas=False) # static images instead of ipympl canvases +explore(caching=True) # allow the .pickle cache (writable data only) +``` + +`explore()` returns the explorer, so the simulation and the current selection stay +available in later cells: + +```python +sim = explorer.sim +explorer.current_time # where the time bar is, in seconds +explorer.selection # the curves currently ticked +explorer.field.frame(400) # the slice on screen, at time step 400 +``` + +`ipympl` is optional; with it each plot gains a toolbar to zoom, pan and save. If the +plots come up as `Failed to load model class 'MPLCanvasModel'`, its browser-side +extension is not loading — pass `interactive_canvas=False`. + +### In a terminal + +The command line front end needs nothing but numpy, so it works over SSH on a machine +with no plotting stack and no browser: + +```sh +fdsreader-explorer-cli ./sample_data # what is in it +fdsreader-explorer-cli ./sample_data --list # slices and curves by name +fdsreader-explorer-cli ./sample_data --slice 0 --time 4.4 # draw a slice +fdsreader-explorer-cli ./sample_data --curve TC_1 --curve HRR +fdsreader-explorer-cli ./sample_data --json # for scripting +``` + +``` +TEMPERATURE [C] — x = 0 m #0 · t = 4.401 s (step 630 of 749) + + 10.00 | + | + 8.82 | + | + | + | ... + | ..... + 5.88 | ..... + | ...:... + | ..... + | .... + | ... + 2.94 | .:.. + | -. + | -:-. + | .-*: + | +%+. + 0.00 | +#: + +------------------ + -2.5 2.5 y [m] + scale 20 … 1412 ramp ' .:-=+*#%@' grid 18x18 +``` + +`--save FILE` writes the current view as an image or an animation instead +(`.png`/`.pdf`/`.svg`/`.gif`/`.mp4`, needs matplotlib). + +Slices are read one time step at a time, so stepping through a long run costs a few +hundred kilobytes rather than loading the whole series into memory. + ## API Documentation [https://fdsreader.readthedocs.io](https://fdsreader.readthedocs.io) From ae65c6a626fa5cb5da76372022864bad8fe9635c Mon Sep 17 00:00:00 2001 From: Lukas Date: Mon, 21 Sep 2026 10:53:03 +0200 Subject: [PATCH 3/4] docs: say plainly that the command line front end is not interactive It draws one view and exits. The README and the docs page described both front ends as interactive, which is only true of the Jupyter one. --- README.md | 13 +++++++------ docs/explorer.rst | 5 +++-- fdsreader/explorer/__init__.py | 3 ++- 3 files changed, 12 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 3537544..f4b871e 100644 --- a/README.md +++ b/README.md @@ -22,10 +22,11 @@ The package is available on PyPI and can be installed using pip: pip install fdsreader ``` -The interactive [explorer](#explorer) needs a plotting stack, which is an optional extra: +The Jupyter [explorer](#explorer) needs a plotting stack, which is an optional extra: ```sh pip install "fdsreader[notebook]" ``` +Its command line counterpart needs nothing beyond numpy and is always installed. _FDS Version 6.7.5 and above are fully supported. Versions below 6.7.5 might work, but are not guaranteed to work._ @@ -68,9 +69,9 @@ documentation of all classes check the API Documentation below. ## Explorer -`fdsreader.explorer` looks at a simulation interactively instead of writing plotting code -for it: a time bar driving a 2D slice, and any number of device or HRR quantities plotted -against time. +`fdsreader.explorer` looks at a simulation without writing plotting code for it: a time +bar driving a 2D slice, and any number of device or HRR quantities plotted against time. +In Jupyter that is an interactive widget; in a terminal it draws one view and exits. ### In Jupyter @@ -106,8 +107,8 @@ extension is not loading — pass `interactive_canvas=False`. ### In a terminal -The command line front end needs nothing but numpy, so it works over SSH on a machine -with no plotting stack and no browser: +The command line front end draws one view and exits. It needs nothing but numpy, so it +works over SSH on a machine with no plotting stack and no browser: ```sh fdsreader-explorer-cli ./sample_data # what is in it diff --git a/docs/explorer.rst b/docs/explorer.rst index f68c77c..4b30819 100644 --- a/docs/explorer.rst +++ b/docs/explorer.rst @@ -1,8 +1,9 @@ Explorer ________ -Front ends for looking at a simulation interactively. The core is importable with nothing -but numpy; the Jupyter front end additionally needs ``pip install "fdsreader[notebook]"``. +Front ends for looking at a simulation: an interactive widget for Jupyter, and a command +line renderer that draws one view and exits. The core is importable with nothing but +numpy; only the Jupyter front end needs ``pip install "fdsreader[notebook]"``. .. automodule:: fdsreader.explorer :autosummary: diff --git a/fdsreader/explorer/__init__.py b/fdsreader/explorer/__init__.py index ca75ced..0e53080 100644 --- a/fdsreader/explorer/__init__.py +++ b/fdsreader/explorer/__init__.py @@ -3,7 +3,8 @@ Three of them share one core: * :mod:`~fdsreader.explorer.notebook` -- widgets for Jupyter, ``explore()`` -* :mod:`~fdsreader.explorer.cli` -- a terminal application, ``fdsreader-explorer-cli`` +* :mod:`~fdsreader.explorer.cli` -- ``fdsreader-explorer-cli``, which draws one view + in a terminal and exits * a desktop front end may follow; it would sit beside these two The shared parts are :mod:`~fdsreader.explorer.data` (reading devices, HRR and slices, From 2a9846ee1f27f5b04cc16a337c19fe027daec7e1 Mon Sep 17 00:00:00 2001 From: Lukas Date: Mon, 21 Sep 2026 11:13:44 +0200 Subject: [PATCH 4/4] feat(explorer): interactive terminal front end fdsreader-explorer-cli -i opens a full-screen view where the slice and the curves are chosen from lists inside the application, instead of being named on the command line. Built on curses from the standard library, so it adds no dependency. The module is absent on Windows, where the error names windows-curses. Key handling and layout are plain functions -- Interactive.handle_key and layout() -- so the behaviour is tested without a terminal. The drawing, the modal list pickers and the colour pairs are the only part that needs curses. The colour scale now defaults to the whole run interactively, where frames should stay comparable while stepping, and to the step on screen for a single picture, which is what a lone frame wants. --scale still overrides both. 23 new tests. --- README.md | 33 ++- docs/explorer.rst | 13 +- fdsreader/explorer/__init__.py | 4 +- fdsreader/explorer/cli.py | 35 ++- fdsreader/explorer/tui.py | 448 +++++++++++++++++++++++++++++++++ tests/explorer/test_tui.py | 173 +++++++++++++ 6 files changed, 696 insertions(+), 10 deletions(-) create mode 100644 fdsreader/explorer/tui.py create mode 100644 tests/explorer/test_tui.py diff --git a/README.md b/README.md index f4b871e..91daaa9 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,8 @@ documentation of all classes check the API Documentation below. `fdsreader.explorer` looks at a simulation without writing plotting code for it: a time bar driving a 2D slice, and any number of device or HRR quantities plotted against time. -In Jupyter that is an interactive widget; in a terminal it draws one view and exits. +In Jupyter that is an interactive widget; in a terminal you can either explore it +interactively or draw a single view and exit. ### In Jupyter @@ -107,8 +108,31 @@ extension is not loading — pass `interactive_canvas=False`. ### In a terminal -The command line front end draws one view and exits. It needs nothing but numpy, so it -works over SSH on a machine with no plotting stack and no browser: +The command line front end needs nothing but numpy, so it works over SSH on a machine +with no plotting stack and no browser. + +`-i` opens a full-screen interactive view, where the slice and the curves are chosen from +lists rather than named on the command line: + +```sh +fdsreader-explorer-cli ./sample_data -i +``` + +| key | | +|---|---| +| `←` `→` or `h` `l` | step through time | +| `page up` / `page down` | ten steps | +| `home` / `end` | first / last step | +| `space` | play, pause | +| `+` `-` | play speed | +| `f` | pick the slice from a list | +| `c` | pick the curves from a list (`space` ticks, `enter` accepts) | +| `[` `]` | previous / next slice | +| `g` / `s` | colour scale over the whole run / this step | +| `?` | help | +| `q` | quit | + +Without `-i` it draws one view and exits, which is what you want in a script: ```sh fdsreader-explorer-cli ./sample_data # what is in it @@ -147,6 +171,9 @@ TEMPERATURE [C] — x = 0 m #0 · t = 4.401 s (step 630 of 749) `--save FILE` writes the current view as an image or an animation instead (`.png`/`.pdf`/`.svg`/`.gif`/`.mp4`, needs matplotlib). +The interactive view uses `curses`, which is in the standard library everywhere except +Windows; there, `pip install windows-curses` provides it. + Slices are read one time step at a time, so stepping through a long run costs a few hundred kilobytes rather than loading the whole series into memory. diff --git a/docs/explorer.rst b/docs/explorer.rst index 4b30819..88eaef0 100644 --- a/docs/explorer.rst +++ b/docs/explorer.rst @@ -1,9 +1,10 @@ Explorer ________ -Front ends for looking at a simulation: an interactive widget for Jupyter, and a command -line renderer that draws one view and exits. The core is importable with nothing but -numpy; only the Jupyter front end needs ``pip install "fdsreader[notebook]"``. +Front ends for looking at a simulation: an interactive widget for Jupyter, and a terminal +application that either explores it interactively (``-i``) or draws one view and exits. +The core is importable with nothing but numpy; only the Jupyter front end needs +``pip install "fdsreader[notebook]"``. .. automodule:: fdsreader.explorer :autosummary: @@ -47,6 +48,12 @@ numpy; only the Jupyter front end needs ``pip install "fdsreader[notebook]"``. :undoc-members: :noindex: +.. automodule:: fdsreader.explorer.tui + :autosummary: + :members: + :undoc-members: + :noindex: + .. automodule:: fdsreader.explorer.cli :autosummary: :members: diff --git a/fdsreader/explorer/__init__.py b/fdsreader/explorer/__init__.py index 0e53080..34fe076 100644 --- a/fdsreader/explorer/__init__.py +++ b/fdsreader/explorer/__init__.py @@ -4,7 +4,8 @@ * :mod:`~fdsreader.explorer.notebook` -- widgets for Jupyter, ``explore()`` * :mod:`~fdsreader.explorer.cli` -- ``fdsreader-explorer-cli``, which draws one view - in a terminal and exits + in a terminal and exits, or with ``-i`` hands over to + :mod:`~fdsreader.explorer.tui` for a full-screen interactive view * a desktop front end may follow; it would sit beside these two The shared parts are :mod:`~fdsreader.explorer.data` (reading devices, HRR and slices, @@ -65,6 +66,7 @@ "plot_device": "plots", "plot_series": "plots", "plot_slice": "plots", + "Interactive": "tui", "RAMP": "render", "series_lines": "render", "slice_lines": "render", diff --git a/fdsreader/explorer/cli.py b/fdsreader/explorer/cli.py index 64aadd7..5d06214 100644 --- a/fdsreader/explorer/cli.py +++ b/fdsreader/explorer/cli.py @@ -4,6 +4,7 @@ or a set of curves, as text. Needs nothing but numpy, so it works over SSH on a machine that has no plotting stack and no browser. + fdsreader-explorer-cli CASE -i explore it interactively fdsreader-explorer-cli CASE what is in it fdsreader-explorer-cli CASE --slice 0 --time 4.4 draw a slice fdsreader-explorer-cli CASE --curve T_1.0 --curve HRR @@ -33,11 +34,14 @@ def _load(path, caching=False): def _state_from(args, fields): """Build the shared state from the command line arguments.""" + # One picture is best scaled to itself; while stepping through time a fixed scale + # keeps the frames comparable. + scale = args.scale or ("global" if args.interactive else "step") state = ExplorerState(scale_mode="step") if args.slice is not None: state.field_index = args.slice - if args.scale in ("global", "step"): - state.scale_mode = args.scale + if scale in ("global", "step"): + state.scale_mode = scale else: try: low, high = (float(part) for part in args.scale.split(",")) @@ -259,13 +263,26 @@ def build_parser(): epilog="With no options it prints what the simulation contains.", ) parser.add_argument("path", help="simulation directory, or the .smv file") + parser.add_argument( + "-i", + "--interactive", + action="store_true", + help="explore the simulation in a full-screen terminal view, " + "choosing the slice and the curves from lists rather than " + "naming them here", + ) parser.add_argument("--list", action="store_true", help="list every slice and curve with the name to pass back in") parser.add_argument("--slice", type=int, metavar="N", help="draw slice N") parser.add_argument( "--time", type=float, metavar="SECONDS", help="time step to draw, the nearest one is used (default: first)" ) parser.add_argument( - "--scale", default="step", metavar="MODE", help="colour limits: 'global', 'step' (default) or 'LOW,HIGH'" + "--scale", + default=None, + metavar="MODE", + help="colour limits: 'global', 'step' or 'LOW,HIGH'. Defaults to 'step' for a " + "single picture and 'global' interactively, where frames should stay comparable " + "while stepping through time", ) parser.add_argument( "--curve", @@ -329,6 +346,18 @@ def main(argv=None): state.render = args.render series = resolve_curves(sim, args.curve) + if args.interactive: + from .tui import run as run_interactive + + if series: + state.curves = tuple(range(len(series))) + try: + run_interactive(sim, state=state) + except ModuleNotFoundError as exc: + print(f"{PROG}: {exc}", file=sys.stderr) + return 1 + return 0 + if args.json: json.dump(as_json(sim, fields, state, series), sys.stdout, indent=2) sys.stdout.write("\n") diff --git a/fdsreader/explorer/tui.py b/fdsreader/explorer/tui.py new file mode 100644 index 0000000..6f51d28 --- /dev/null +++ b/fdsreader/explorer/tui.py @@ -0,0 +1,448 @@ +"""Interactive terminal front end. + +A keyboard-driven view of one simulation: a slice on one side, curves on the other, and +a time bar driving both. What is shown is chosen inside the application rather than on +the command line. + +Built on :mod:`curses` from the standard library, so it adds no dependency. That module +is not part of the standard library on Windows, where ``pip install windows-curses`` +provides it. + +The drawing and the key handling are kept apart from curses itself -- :func:`layout` and +:meth:`Interactive.handle_key` are ordinary functions -- so the behaviour can be tested +without a terminal. +""" + +from .data import build_series +from .fields import fields_of +from .render import RAMP, series_lines, slice_lines +from .state import ExplorerState + +#: Width at which the slice and the curves stop fitting side by side and stack instead. +SIDE_BY_SIDE_COLUMNS = 100 + +#: How long a frame is held while playing, in milliseconds, slowest to fastest. +PLAY_INTERVALS = (500, 250, 120, 60, 30) + +HELP = [ + "", + " Time", + " left / right one step h / l one step", + " page up / page down ten steps < / > one per cent", + " home / end first / last space play / pause", + " + / - play speed", + "", + " Choosing what to show", + " f pick the slice from a list", + " c pick the curves from a list", + " [ / ] previous / next slice", + "", + " Colour scale", + " g the whole run", + " s the step on screen", + "", + " ? this help q quit", + "", +] + + +def layout(rows, cols): + """Where everything goes, for a terminal of ``rows`` by ``cols`` characters. + + Returns a dict of ``(top, left, height, width)`` boxes. The slice and the curves sit + side by side when there is room and stack when there is not, which mirrors what the + notebook front end does with its panels. + """ + rows, cols = max(rows, 6), max(cols, 20) + header, footer = 2, 2 + body_top = header + body_height = max(1, rows - header - footer) + + if cols >= SIDE_BY_SIDE_COLUMNS: + left_width = max(20, cols // 2) + boxes = { + "slice": (body_top, 0, body_height, left_width), + "curves": (body_top, left_width + 1, body_height, cols - left_width - 1), + } + else: + slice_height = max(1, (body_height * 3) // 5) + boxes = { + "slice": (body_top, 0, slice_height, cols), + "curves": (body_top + slice_height, 0, body_height - slice_height, cols), + } + + boxes["header"] = (0, 0, header, cols) + boxes["footer"] = (rows - footer, 0, footer, cols) + return boxes + + +class Interactive: + """The application: what is selected, and what each key does to it. + + Holds no curses objects, so it can be driven directly in a test. + """ + + def __init__(self, sim, state=None): + self.sim = sim + self.fields = fields_of(sim) + self.series = build_series(sim) + self.state = state or ExplorerState() + self.state.curves = (0,) if self.series else () + self.playing = False + self.speed = 2 # index into PLAY_INTERVALS + self.message = "" + self.showing_help = False + + # -- what is on screen ----------------------------------------------- + @property + def field(self): + return self.state.field(self.fields) + + @property + def interval(self): + return PLAY_INTERVALS[self.speed] + + def title(self): + field = self.field + if field is None: + return f"{self.sim.chid} (no slice output)" + step = self.state.clamp(self.fields) + return f"{self.sim.chid} {field.label} t = {field.times[step]:.3f} s step {step + 1}/{len(field.times)}" + + def status(self): + if self.message: + return self.message + scale = {"global": "whole run", "step": "this step", "manual": "manual"} + bits = [f"scale: {scale.get(self.state.scale_mode, self.state.scale_mode)}"] + bits.append(f"curves: {len(self.state.curves)}") + if self.playing: + bits.append(f"playing ({self.interval} ms)") + bits.append("? for help, q to quit") + return " ".join(bits) + + # -- drawing ---------------------------------------------------------- + def slice_text(self, height, width): + field = self.field + if field is None: + return ["", " This simulation has no slice (SLCF) output."] + step = self.state.clamp(self.fields) + frame = field.frame(step) + vmin, vmax, warning = self.state.limits(field, frame=frame) + lines = slice_lines(field, step, vmin, vmax, height=max(4, height - 3), width=max(10, width - 2), frame=frame) + if warning: + lines.append(f" note: {warning}") + return lines + + def curves_text(self, height, width): + chosen = self.state.selected_series(self.series) + if not chosen: + return ["", " No curves selected — press c to choose some."] + return series_lines( + chosen, + height=max(4, height - 4), + width=max(10, width - 13), + cursor_time=self.state.time(self.fields), + ) + + # -- keys -------------------------------------------------------------- + def step_by(self, delta): + field = self.field + if field is None: + return + last = len(field.times) - 1 + self.state.timestep = max(0, min(self.state.timestep + delta, last)) + + def handle_key(self, key): + """Act on one key. Returns ``"quit"``, ``"fields"``, ``"curves"`` or ``None``. + + ``key`` is a character, or one of the ``"left"``, ``"right"``, ``"pgup"``, + ``"pgdn"``, ``"home"``, ``"end"`` names, so tests need no curses constants. + """ + self.message = "" + if self.showing_help and key not in ("?",): + self.showing_help = False + return None + + field = self.field + per_cent = max(1, len(field.times) // 100) if field is not None else 1 + + if key in ("q", "Q"): + return "quit" + if key == "?": + self.showing_help = not self.showing_help + return None + if key in ("left", "h"): + self.step_by(-1) + elif key in ("right", "l"): + self.step_by(1) + elif key == "pgup": + self.step_by(-10) + elif key == "pgdn": + self.step_by(10) + elif key == "<": + self.step_by(-per_cent) + elif key == ">": + self.step_by(per_cent) + elif key == "home": + self.state.timestep = 0 + elif key == "end": + self.state.timestep = (len(field.times) - 1) if field is not None else 0 + elif key == " ": + self.playing = not self.playing + elif key == "+": + self.speed = min(self.speed + 1, len(PLAY_INTERVALS) - 1) + elif key == "-": + self.speed = max(self.speed - 1, 0) + elif key == "g": + self.state.scale_mode = "global" + elif key == "s": + self.state.scale_mode = "step" + elif key == "[": + self.select_field(self.state.field_index - 1) + elif key == "]": + self.select_field(self.state.field_index + 1) + elif key == "f": + return "fields" + elif key == "c": + return "curves" + return None + + def advance(self): + """One frame of playback; stops at the end rather than looping.""" + field = self.field + if field is None or not self.playing: + return + if self.state.timestep >= len(field.times) - 1: + self.playing = False + else: + self.state.timestep += 1 + + def select_field(self, index): + if not self.fields: + return + previous = self.field + index = max(0, min(index, len(self.fields) - 1)) + if previous is not None and index != self.state.field_index: + previous.clear_cache() # keep one field's data in memory at a time + self.state.field_index = index + self.state.clamp(self.fields) + + def set_curves(self, indices): + self.state.curves = tuple(sorted(set(indices))) + + # -- list contents for the pickers -------------------------------------- + def field_choices(self): + return [f.label for f in self.fields] + + def curve_choices(self): + return [entry["label"] for entry in self.series] + + +# ---------------------------------------------------------------- curses layer + +#: 256-colour approximation of the sequential map the other front ends use, from the +#: coldest ramp character to the hottest. +_HEAT_256 = (233, 17, 54, 90, 126, 160, 196, 202, 208, 220, 227) +#: The same idea for a terminal with only the eight basic colours. +_HEAT_8 = (0, 4, 4, 5, 5, 1, 1, 3, 3, 7, 7) + + +def _key_name(key): + """Turn a curses key code into the names :meth:`Interactive.handle_key` expects.""" + import curses + + names = { + curses.KEY_LEFT: "left", + curses.KEY_RIGHT: "right", + curses.KEY_UP: "up", + curses.KEY_DOWN: "down", + curses.KEY_PPAGE: "pgup", + curses.KEY_NPAGE: "pgdn", + curses.KEY_HOME: "home", + curses.KEY_END: "end", + 10: "enter", + 13: "enter", + 27: "escape", + } + if key in names: + return names[key] + if 0 <= key < 256: + return chr(key) + return "" + + +def _init_colors(): + """Colour pairs for the ramp, or ``None`` when the terminal has no colour.""" + import curses + + if not curses.has_colors(): + return None + curses.start_color() + try: + curses.use_default_colors() + except curses.error: + pass + palette = _HEAT_256 if curses.COLORS >= 256 else _HEAT_8 + attrs = [] + for i, colour in enumerate(palette[: len(RAMP)], start=1): + try: + curses.init_pair(i, colour, -1) + attrs.append(curses.color_pair(i)) + except curses.error: + attrs.append(0) + return attrs + + +def _put(window, row, col, text, attr=0, width=None): + """Write text, clipped to the window; curses errors on the last cell are ignored.""" + import curses + + height, cols = window.getmaxyx() + if not (0 <= row < height) or col >= cols: + return + room = (cols - col) if width is None else min(width, cols - col) + if room <= 0: + return + try: + window.addnstr(row, col, text, room, attr) + except curses.error: + pass # writing the bottom-right cell always raises + + +def _draw_ramped(window, row, col, text, ramp_attrs, width): + """Draw a line of slice art, colouring each cell by how far up the ramp it is.""" + if ramp_attrs is None: + _put(window, row, col, text, 0, width) + return + for offset, char in enumerate(text[:width]): + index = RAMP.find(char) + attr = ramp_attrs[min(index, len(ramp_attrs) - 1)] if index > 0 else 0 + _put(window, row, col + offset, char, attr, 1) + + +def _pick(stdscr, title, choices, selected, multiple): + """A modal list. Returns the chosen indices, or ``None`` when cancelled.""" + import curses + + if not choices: + return None + chosen = set(selected) + cursor = min(selected) if selected else 0 + + while True: + stdscr.erase() + rows, cols = stdscr.getmaxyx() + _put(stdscr, 0, 0, title, curses.A_BOLD) + hint = "space to tick, enter to accept, esc to cancel" if multiple else "enter to choose, esc to cancel" + _put(stdscr, 1, 0, hint, curses.A_DIM) + + room = max(1, rows - 4) + first = max(0, min(cursor - room // 2, len(choices) - room)) + for offset in range(min(room, len(choices) - first)): + index = first + offset + mark = "[x] " if index in chosen else "[ ] " if multiple else " " + attr = curses.A_REVERSE if index == cursor else 0 + _put(stdscr, 2 + offset, 0, f" {mark}{choices[index]}", attr, cols - 1) + + if len(choices) > room: + _put(stdscr, rows - 1, 0, f"{cursor + 1}/{len(choices)}", curses.A_DIM) + stdscr.refresh() + + key = _key_name(stdscr.getch()) + if key == "escape" or key in ("q", "Q"): + return None + if key in ("up", "k"): + cursor = max(0, cursor - 1) + elif key in ("down", "j"): + cursor = min(len(choices) - 1, cursor + 1) + elif key == "pgup": + cursor = max(0, cursor - room) + elif key == "pgdn": + cursor = min(len(choices) - 1, cursor + room) + elif key == "home": + cursor = 0 + elif key == "end": + cursor = len(choices) - 1 + elif key == " " and multiple: + chosen.symmetric_difference_update({cursor}) + elif key == "enter": + if multiple: + return sorted(chosen) + return [cursor] + + +def _draw(stdscr, app, ramp_attrs): + import curses + + stdscr.erase() + rows, cols = stdscr.getmaxyx() + + if app.showing_help: + for row, line in enumerate(HELP[: rows - 1]): + heading = line.strip() and not line.startswith(" ") + _put(stdscr, row, 0, line, curses.A_BOLD if heading else 0) + _put(stdscr, rows - 1, 0, " any key to go back", curses.A_DIM) + stdscr.refresh() + return + + boxes = layout(rows, cols) + top, left, _, width = boxes["header"] + _put(stdscr, top, left, app.title(), curses.A_BOLD, width) + _put(stdscr, top + 1, left, "─" * width, curses.A_DIM, width) + + top, left, height, width = boxes["slice"] + for offset, line in enumerate(app.slice_text(height, width)[:height]): + _draw_ramped(stdscr, top + offset, left, line, ramp_attrs, width) + + top, left, height, width = boxes["curves"] + for offset, line in enumerate(app.curves_text(height, width)[:height]): + _put(stdscr, top + offset, left, line, 0, width) + + top, left, _, width = boxes["footer"] + _put(stdscr, top, left, "─" * width, curses.A_DIM, width) + _put(stdscr, top + 1, left, " " + app.status(), curses.A_DIM, width) + stdscr.refresh() + + +def run(sim, state=None): + """Show the interactive explorer for ``sim`` until the user quits.""" + try: + import curses + except ModuleNotFoundError as exc: # pragma: no cover - Windows without the shim + raise ModuleNotFoundError( + "the interactive explorer needs the curses module, which is not part of the " + "standard library on Windows. Install it with: pip install windows-curses" + ) from exc + + app = Interactive(sim, state=state) + + def main(stdscr): + curses.curs_set(0) + stdscr.keypad(True) + ramp_attrs = _init_colors() + + while True: + _draw(stdscr, app, ramp_attrs) + stdscr.timeout(app.interval if app.playing else -1) + key = stdscr.getch() + + if key == -1: # the wait ran out, so play the next frame + app.advance() + continue + if key == curses.KEY_RESIZE: + continue + + action = app.handle_key(_key_name(key)) + if action == "quit": + return + if action == "fields": + picked = _pick(stdscr, " Slice", app.field_choices(), [app.state.field_index], False) + if picked: + app.select_field(picked[0]) + elif action == "curves": + picked = _pick(stdscr, " Curves", app.curve_choices(), list(app.state.curves), True) + if picked is not None: + app.set_curves(picked) + + curses.wrapper(main) + return app diff --git a/tests/explorer/test_tui.py b/tests/explorer/test_tui.py new file mode 100644 index 0000000..ffb20bf --- /dev/null +++ b/tests/explorer/test_tui.py @@ -0,0 +1,173 @@ +"""The interactive front end's behaviour, without a terminal. + +Key handling and layout are ordinary functions, so what the application *does* can be +tested directly; only the drawing needs curses. +""" + +import pytest + +from fdsreader.explorer import tui +from fdsreader.explorer.tui import SIDE_BY_SIDE_COLUMNS, Interactive, layout + + +class FakeSim: + chid = "fake" + root_path = "." + + +@pytest.fixture +def app(monkeypatch, fake_field, fake_series): + """An application over one field and two curves.""" + monkeypatch.setattr(tui, "fields_of", lambda sim: [fake_field, fake_field]) + monkeypatch.setattr(tui, "build_series", lambda sim: fake_series) + return Interactive(FakeSim()) + + +# -- layout --------------------------------------------------------------- +def test_wide_terminal_puts_the_panes_side_by_side(): + boxes = layout(40, SIDE_BY_SIDE_COLUMNS) + assert boxes["slice"][0] == boxes["curves"][0] # same top row + assert boxes["curves"][1] > 0 # curves start further right + + +def test_narrow_terminal_stacks_the_panes(): + boxes = layout(40, SIDE_BY_SIDE_COLUMNS - 1) + assert boxes["curves"][1] == 0 # both start at the left + assert boxes["curves"][0] > boxes["slice"][0] # curves sit below + + +@pytest.mark.parametrize("rows,cols", [(6, 20), (24, 80), (60, 200), (1, 1)]) +def test_layout_stays_inside_the_terminal(rows, cols): + boxes = layout(rows, cols) + for top, left, height, width in boxes.values(): + assert top >= 0 and left >= 0 + assert height >= 1 and width >= 1 + assert top + height <= max(rows, 6) + assert left + width <= max(cols, 20) + + +# -- time ----------------------------------------------------------------- +def test_stepping_moves_one_frame(app): + app.handle_key("right") + assert app.state.timestep == 1 + app.handle_key("h") + assert app.state.timestep == 0 + + +def test_stepping_is_clamped_at_both_ends(app): + app.handle_key("left") + assert app.state.timestep == 0 + app.handle_key("end") + last = len(app.field.times) - 1 + assert app.state.timestep == last + app.handle_key("right") + assert app.state.timestep == last + + +def test_home_and_end(app): + app.handle_key("end") + app.handle_key("home") + assert app.state.timestep == 0 + + +def test_play_toggles_and_advance_stops_at_the_end(app): + app.handle_key(" ") + assert app.playing + for _ in range(len(app.field.times) + 5): + app.advance() + assert app.state.timestep == len(app.field.times) - 1 + assert not app.playing # it stops rather than looping + + +def test_advance_does_nothing_while_paused(app): + app.advance() + assert app.state.timestep == 0 + + +def test_speed_is_clamped(app): + for _ in range(20): + app.handle_key("+") + fastest = app.interval + for _ in range(20): + app.handle_key("-") + assert app.interval > fastest + + +# -- what is shown --------------------------------------------------------- +def test_scale_keys(app): + app.handle_key("s") + assert app.state.scale_mode == "step" + app.handle_key("g") + assert app.state.scale_mode == "global" + + +def test_bracket_keys_change_field_and_clamp(app): + app.handle_key("]") + assert app.state.field_index == 1 + app.handle_key("]") + assert app.state.field_index == 1 # only two fields + app.handle_key("[") + app.handle_key("[") + assert app.state.field_index == 0 + + +def test_changing_field_keeps_the_time_step_valid(app): + app.handle_key("end") + app.select_field(1) + assert app.state.timestep <= len(app.field.times) - 1 + + +def test_picker_keys_ask_for_a_list(app): + assert app.handle_key("f") == "fields" + assert app.handle_key("c") == "curves" + assert len(app.field_choices()) == 2 + assert len(app.curve_choices()) == 2 + + +def test_set_curves_sorts_and_deduplicates(app): + app.set_curves([1, 0, 1]) + assert app.state.curves == (0, 1) + + +def test_quit_and_help(app): + assert app.handle_key("q") == "quit" + assert app.handle_key("?") is None + assert app.showing_help + app.handle_key("l") # any other key closes it again + assert not app.showing_help + + +# -- text it produces ------------------------------------------------------- +def test_title_names_the_field_and_the_time(app): + app.handle_key("right") + title = app.title() + assert "fake" in title + assert "step 2/" in title + + +def test_status_reports_the_scale_and_the_curves(app): + app.handle_key("g") + status = app.status() + assert "whole run" in status + assert "curves: 1" in status + + +def test_slice_and_curve_text_are_drawable(app): + assert any(line.strip() for line in app.slice_text(20, 60)) + assert any(line.strip() for line in app.curves_text(20, 60)) + + +def test_no_curves_selected_says_so(app): + app.set_curves([]) + assert "press c" in " ".join(app.curves_text(20, 60)) + + +def test_a_simulation_without_slices(monkeypatch, fake_series): + monkeypatch.setattr(tui, "fields_of", lambda sim: []) + monkeypatch.setattr(tui, "build_series", lambda sim: fake_series) + app = Interactive(FakeSim()) + assert app.field is None + assert "no slice output" in app.title() + assert "no slice" in " ".join(app.slice_text(20, 60)).lower() + app.handle_key("right") # must not raise + app.advance()