Module `phi.vis`

Visualization: plotting, interactive user interfaces.

Use view() to show fields or field variables in an interactive user interface.

Use plot() to plot fields using Matplotlib.

See the user interface documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html

Expand source code

"""
Visualization: plotting, interactive user interfaces.

Use `view()` to show fields or field variables in an interactive user interface.

Use `plot()` to plot fields using Matplotlib.

See the user interface documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html
"""
from ._viewer import Viewer
from ._matplotlib import plot_scalars
from ._io import load_scalars
from ._plot_util import smooth
from ._vis import view, control, show, close, action, plot, overlay, write_image, write_image as savefig

__all__ = [key for key in globals().keys() if not key.startswith('_')]

__pdoc__ = {
    'Viewer.actions': False,
    'Viewer.can_progress': False,
    'Viewer.control_names': False,
    'Viewer.curve_names': False,
    'Viewer.field_names': False,
    'Viewer.get_control': False,
    'Viewer.get_curve': False,
    'Viewer.get_field': False,
    'Viewer.run_action': False,
    'Viewer.set_control_value': False,
    'Viewer.log_scalars': False,
    'Viewer.controls': False,
    'Viewer.get_control_value': False,
    'Viewer.info': False,
    'Viewer.reset': False,
    'Viewer.progress': False,
    'Viewer.__init__': False,
}

Functions

def action(fun)

Expand source code

def action(fun):
    doc = inspect.getdoc(fun)
    ACTIONS[Action(fun.__name__, doc)] = fun
    return fun

def close(figure=None)

Close and destroy a figure.

Args

figure: (Optional) A figure that was created using plot(). If not specified, closes the figure created most recently.

Expand source code

def close(figure=None):
    """
    Close and destroy a figure.

    Args:
        figure: (Optional) A figure that was created using `plot()`.
            If not specified, closes the figure created most recently.
    """
    if figure is None:
        figure = LAST_FIGURE[0]
    if isinstance(figure, Tensor):
        for fig in figure:
            close(fig)
    else:
        plots = get_plots_by_figure(figure)
        plots.close(figure)

def control(value, range: tuple = None, description='', **kwargs)

Mark a variable as controllable by any GUI created via view().

Example:

>>> dt = control(1.0, (0.1, 10), name="Time increment (dt)")

This will cause a control component (slider, checkbox, text field, drop-down, etc.) to be generated in the user interface. Changes to that component will immediately be reflected in the Python variable assigned to the control. The Python variable will always hold a primitive type, such as int, float´,bool or str`.

Args

value: Initial value. Must be either int, float, bool or str.
range: (Optional) Specify range of possible values as (min, max). Only for int, float and str values.
description: Human-readable description.
**kwargs: Additional arguments to determine the appearance of the GUI component, e.g. rows for text fields or log=False for float sliders.

Returns

value

Expand source code

def control(value, range: tuple = None, description="", **kwargs):
    """
    Mark a variable as controllable by any GUI created via `view()`.

    Example:
    >>> dt = control(1.0, (0.1, 10), name="Time increment (dt)")

    This will cause a control component (slider, checkbox, text field, drop-down, etc.) to be generated in the user interface.
    Changes to that component will immediately be reflected in the Python variable assigned to the control.
    The Python variable will always hold a primitive type, such as `int`, `float´, `bool` or `str`.

    Args:
        value: Initial value. Must be either `int`, `float`, `bool` or `str`.
        range: (Optional) Specify range of possible values as `(min, max)`. Only for `int`, `float` and `str` values.
        description: Human-readable description.
        **kwargs: Additional arguments to determine the appearance of the GUI component,
            e.g. `rows` for text fields or `log=False` for float sliders.

    Returns:
        `value`
    """
    assert type(value) in (int, float, bool, str), f"Value must be one of (int, float, bool, str) but {type(value)}"
    calling_code = inspect.stack()[1].code_context[0]
    assert 'control' in calling_code and '=' in calling_code, f"control() must be used in a variable assignment statement but context is: {calling_code}"
    calling_code = calling_code[:calling_code.index('control')]
    var_names = [var.strip() for var in calling_code.split('=')[:-1]]
    var_names = [n for n in var_names if n]
    for var_name in var_names:
        ctrl = Control(var_name, type(value), value, range, description, kwargs)
        value_range(ctrl)  # checks if valid
        CONTROL_VARS[var_name] = ctrl
    return value

def load_scalars(scene: phi.field._scene.Scene, name: str, prefix='log_', suffix='.txt', x: Optional[str] = 'steps', entries_dim=(iterationˢ=None), batch_dim=(batchᵇ=None))

Read one or a Tensor of scalar logs as curves.

Args

scene: Scene or str. Directory containing the log files.
name: Log file base name.
prefix: Log file prefix.
suffix: Log file suffix.
x: 'steps' or 'time'
entries_dim: Curve dimension.

Returns

Tensor containing entries_dim and vector.

Expand source code

@math.broadcast
def load_scalars(scene: Scene or str,
                 name: str,
                 prefix='log_',
                 suffix='.txt',
                 x: Optional[str]='steps',
                 entries_dim=spatial('iteration'),
                 batch_dim=batch('batch')):
    """
    Read one or a `Tensor` of scalar logs as curves.

    Args:
        scene: `Scene` or `str`. Directory containing the log files.
        name: Log file base name.
        prefix: Log file prefix.
        suffix: Log file suffix.
        x: 'steps'  or 'time'
        entries_dim: Curve dimension.

    Returns:
        `Tensor` containing `entries_dim` and `vector`.
    """
    assert x in (None, 'steps', 'time')
    if isinstance(scene, str):
        scene = Scene.at(scene)
    assert isinstance(scene, Scene), f"scene must be a Scene or str but got {type(scene)}"
    assert shape(scene).rank == 0, f"Use math.map(load_scalars, ...) to load data from multiple scenes"
    ML_LOGGER.debug(f"Reading {os.path.join(scene.path, f'{prefix}{name}{suffix}')}")
    curve = numpy.loadtxt(os.path.join(scene.path, f"log_{name}.txt"))
    if curve.ndim == 2:
        x_values = curve[:, 0]
        values = curve[:, 1:]
    elif curve.ndim == 1 and numpy.floor(curve[0]) == curve[0]:  # new format but only one entry
        x_values = curve[None, 0]
        values = curve[None, 1:]
    else:
        values = curve[:, None]
        x_values = numpy.arange(len(values))
    if x == 'time':
        assert x == 'time', f"x must be 'steps' or 'time' but got {x}"
        ML_LOGGER.debug(f"Reading {os.path.join(scene.path, 'log_step_time.txt')}")
        _, x_values, *_ = numpy.loadtxt(os.path.join(scene.path, "log_step_time.txt")).T
        values = values[:len(x_values + 1)]
        x_values = numpy.cumsum(x_values[:len(values) - 1])
        x_values = numpy.concatenate([[0.], x_values])
    x_values = wrap(x_values, entries_dim)
    values = wrap(values, entries_dim, batch_dim)
    if x is not None:
        return math.stack([x_values, values], channel(vector=[x, name]))
    return values

def overlay(*fields: Union[phi.field._field.SampledField, phiml.math._tensors.Tensor]) ‑> phiml.math._tensors.Tensor

Specify that multiple fields should be drawn on top of one another in the same figure. The fields will be plotted in the order they are given, i.e. the last field on top.

>>> plot(vis.overlay(heatmap, points, velocity))

Args

*fields: SampledField or Tensor instances

Returns

Plottable object

Expand source code

def overlay(*fields: Union[SampledField, Tensor]) -> Tensor:
    """
    Specify that multiple fields should be drawn on top of one another in the same figure.
    The fields will be plotted in the order they are given, i.e. the last field on top.

    >>> plot(vis.overlay(heatmap, points, velocity))

    Args:
        *fields: `SampledField` or `Tensor` instances

    Returns:
        Plottable object
    """
    return layout(fields, math.channel('overlay'))

def plot(*fields: Union[phi.field._field.SampledField, phiml.math._tensors.Tensor, phi.geom._geom.Geometry, list, tuple, dict], lib: Union[str, phi.vis._vis_base.PlottingLibrary] = None, row_dims: Union[str, tuple, list, set, phiml.math._shape.Shape, Callable] = None, col_dims: Union[str, tuple, list, set, phiml.math._shape.Shape, Callable] = <function batch>, animate: Union[str, tuple, list, set, phiml.math._shape.Shape, Callable] = None, overlay: Union[str, tuple, list, set, phiml.math._shape.Shape, Callable] = 'overlay', title: Union[str, phiml.math._tensors.Tensor, list, tuple] = None, size=(12, 5), same_scale: Union[bool, phiml.math._shape.Shape, tuple, list, str] = True, log_dims: Union[phiml.math._shape.Shape, tuple, list, str] = '', show_color_bar=True, color: Union[str, int, phiml.math._tensors.Tensor, list, tuple] = None, alpha: Union[float, phiml.math._tensors.Tensor, list, tuple] = 1.0, err: Union[float, phiml.math._tensors.Tensor, list, tuple] = 0.0, frame_time=100, repeat=True)

Creates one or multiple figures and sub-figures and plots the given fields.

To show the figures, use show().

The arguments row_dims, col_dims, animate and overlay() control how data is presented. Each accepts dimensions as a str, Shape, tuple, list or type function. In addition to the dimensions present on the data to be plotted, the dimensions args is created if multiple arguments are passed, and tuple, list, dict are generated for corresponding objects to be plotted.

Args

fields: Fields or Tensors to plot.
lib: Plotting library name or reference. Valid names are 'matplotlib', 'plotly' and 'console'.
row_dims: Batch dimensions along which sub-figures should be laid out vertically. Shape or comma-separated names as str, tuple or list.
col_dims: Batch dimensions along which sub-figures should be laid out horizontally. Shape or comma-separated names as str, tuple or list.
title: str for figures with a single subplot. For subplots, pass a string Tensor matching the content dimensions, i.e. row_dims and col_dims. Passing a tuple, list or dict, will create a tensor with these names internally.
size: Figure size in inches, (width, height).
same_scale: Whether to use the same axis limits for all sub-figures.
log_dims: Dimensions for which the plot axes should be scaled logarithmically. Can be given as a comma-separated str, a sequence of dimension names or a Shape. Use '_' to scale unnamed axes logarithmically, e.g. the y-axis of scalar functions.
show_color_bar: Whether to display color bars for heat maps.
color: Tensor of line / marker colors. The color can be specified either as a cycle index (int tensor) or as a hex code (str tensor). The color of different lines and markers can vary.
alpha: Opacity as float or Tensor. This affects all elements, not only line plots. Opacity can vary between lines and markers.
err: Expected deviation from the value given in fields. For supported plots, adds error bars of size 2·err. If the plotted data is the mean of some distribution, a good choice for err is the standard deviation along the mean dims.
animate: Time dimension to animate. If not present in the data, will produce a regular plot instead.
overlay: Dimensions along which elements should be overlaid in the same subplot. The default is only the overlay() dimension which is created by overlay().
frame_time: Interval between frames in the animation.
repeat: Whether the animation should loop.

Returns

Tensor of figure objects. The tensor contains those dimensions of fields that were not reduced by row_dims, col_dims or animate. Currently, only single-figure plots are supported.

In case of an animation, a displayable animation object will be returned instead of a Tensor.

Expand source code

def plot(*fields: Union[SampledField, Tensor, Geometry, list, tuple, dict],
         lib: Union[str, PlottingLibrary] = None,
         row_dims: DimFilter = None,
         col_dims: DimFilter = batch,
         animate: DimFilter = None,
         overlay: DimFilter = 'overlay',
         title: Union[str, Tensor, list, tuple] = None,
         size=(12, 5),
         same_scale: Union[bool, Shape, tuple, list, str] = True,
         log_dims: Union[str, tuple, list, Shape] = '',
         show_color_bar=True,
         color: Union[str, int, Tensor, list, tuple] = None,
         alpha: Union[float, Tensor, list, tuple] = 1.,
         err: Union[Tensor, tuple, list, float] = 0.,
         frame_time=100,
         repeat=True):
    """
    Creates one or multiple figures and sub-figures and plots the given fields.

    To show the figures, use `show()`.

    The arguments `row_dims`, `col_dims`, `animate` and `overlay` control how data is presented.
    Each accepts dimensions as a `str`, `Shape`, tuple, list or type function.
    In addition to the dimensions present on the data to be plotted, the dimensions `args` is created if multiple arguments are passed,
    and `tuple`, `list`, `dict` are generated for corresponding objects to be plotted.

    Args:
        fields: Fields or Tensors to plot.
        lib: Plotting library name or reference. Valid names are `'matplotlib'`, `'plotly'` and `'console'`.
        row_dims: Batch dimensions along which sub-figures should be laid out vertically.
            `Shape` or comma-separated names as `str`, `tuple` or `list`.
        col_dims: Batch dimensions along which sub-figures should be laid out horizontally.
            `Shape` or comma-separated names as `str`, `tuple` or `list`.
        title: `str` for figures with a single subplot.
            For subplots, pass a string `Tensor` matching the content dimensions, i.e. `row_dims` and `col_dims`.
            Passing a `tuple`, `list` or `dict`, will create a tensor with these names internally.
        size: Figure size in inches, `(width, height)`.
        same_scale: Whether to use the same axis limits for all sub-figures.
        log_dims: Dimensions for which the plot axes should be scaled logarithmically.
            Can be given as a comma-separated `str`, a sequence of dimension names or a `Shape`.
            Use `'_'` to scale unnamed axes logarithmically, e.g. the y-axis of scalar functions.
        show_color_bar: Whether to display color bars for heat maps.
        color: `Tensor` of line / marker colors.
            The color can be specified either as a cycle index (int tensor) or as a hex code (str tensor).
            The color of different lines and markers can vary.
        alpha: Opacity as `float` or `Tensor`.
            This affects all elements, not only line plots.
            Opacity can vary between lines and markers.
        err: Expected deviation from the value given in `fields`.
            For supported plots, adds error bars of size *2·err*.
            If the plotted data is the mean of some distribution, a good choice for `err` is the standard deviation along the mean dims.
        animate: Time dimension to animate.
            If not present in the data, will produce a regular plot instead.
        overlay: Dimensions along which elements should be overlaid in the same subplot.
            The default is only the `overlay` dimension which is created by `overlay()`.
        frame_time: Interval between frames in the animation.
        repeat: Whether the animation should loop.

    Returns:
        `Tensor` of figure objects.
        The tensor contains those dimensions of `fields` that were not reduced by `row_dims`, `col_dims` or `animate`.
        Currently, only single-figure plots are supported.

        In case of an animation, a displayable animation object will be returned instead of a `Tensor`.
    """
    positioning = {}
    indices: Dict[Tuple[int, int], List[dict]] = {}
    nrows, ncols, fig_shape, reduced_shape = layout_sub_figures(layout(fields, batch('args')), row_dims, col_dims, animate, overlay, 0, 0, positioning, indices, {})
    animate = fig_shape.only(animate)
    fig_shape = fig_shape.without(animate)
    plots = default_plots() if lib is None else get_plots(lib)
    # --- Process arguments ---
    if title is None:
        title_by_subplot = {pos: title_label(common_index(*i, exclude=reduced_shape.singleton)) for pos, i in indices.items()}
    elif isinstance(title, Tensor) and ('rows' in title.shape or 'cols' in title.shape):
        title_by_subplot = {(row, col): title.rows[row].cols[col].native() for (row, col) in positioning}
    else:
        title = layout_pytree_node(title, wrap_leaf=True)
        title_by_subplot = {pos: _title(title, i[0]) for pos, i in indices.items()}
    log_dims = parse_dim_order(log_dims) or ()
    color = layout_pytree_node(color, wrap_leaf=True)
    alpha = layout_pytree_node(alpha, wrap_leaf=True)
    err = layout_pytree_node(err, wrap_leaf=True)
    if same_scale is True:
        same_scale = '_'
    elif same_scale is False or same_scale is None:
        same_scale = ''
    same_scale = parse_dim_order(same_scale)
    if '_' in same_scale:
        if any([f.values.dtype.kind == complex for l in positioning.values() for f in l]):
            min_val = 0
            max_val = max([float(abs(f.values).finite_max) for l in positioning.values() for f in l])
        else:
            min_val = min([float(f.values.finite_min) for l in positioning.values() for f in l])
            max_val = max([float(f.values.finite_max) for l in positioning.values() for f in l])
    else:
        min_val = max_val = None
    # --- Layout ---
    subplots = {pos: _space(*fields, ignore_dims=animate) for pos, fields in positioning.items()}
    subplots = {pos: _insert_value_dim(space, pos, subplots, min_val, max_val) for pos, space in subplots.items()}
    if same_scale:
        shared_lim: Box = share_axes(*subplots.values(), axes=same_scale)
        subplots = {pos: replace_bounds(lim, shared_lim) for pos, lim in subplots.items()}
    # --- animate or plot ---
    if fig_shape.volume == 1:
        figure, axes = plots.create_figure(size, nrows, ncols, subplots, title_by_subplot, log_dims)
        if animate:
            def plot_frame(frame: int):
                for pos, fields in positioning.items():
                    for i, f in enumerate(fields):
                        idx = indices[pos][i]
                        f = f[{animate.name: frame}]
                        plots.plot(f, figure, axes[pos], subplots[pos], min_val, max_val, show_color_bar, color[idx], alpha[idx], err[idx])
                plots.finalize(figure)
            anim = plots.animate(figure, animate.size, plot_frame, frame_time, repeat)
            if 'google.colab' in sys.modules or 'ipykernel' in sys.modules:
                plots.close(figure)
            LAST_FIGURE[0] = anim
            return anim
        else:
            for pos, fields in positioning.items():
                for i, f in enumerate(fields):
                    idx = indices[pos][i]
                    err_ = err[idx]
                    while isinstance(err_, Layout) and not err_.shape and isinstance(err_.native(), Tensor):
                        err_ = err_.native()[idx]
                    color_ = color[idx]
                    while isinstance(color_, Layout) and not color_.shape and isinstance(color_.native(), Tensor):
                        color_ = color_.native()[idx]
                    plots.plot(f, figure, axes[pos], subplots[pos], min_val, max_val, show_color_bar, color_, alpha[idx], err_)
            plots.finalize(figure)
            LAST_FIGURE[0] = figure
            return layout(figure)
    else:
        raise NotImplementedError(f"Figure batches not yet supported. Use rows and cols to reduce all batch dimensions. Not reduced. {fig_shape}")

def plot_scalars(scene: Union[str, tuple, list, phi.field._scene.Scene, phiml.math._tensors.Tensor], names: Union[str, phiml.math._tensors.Tensor, list, tuple] = None, reduce: Union[phiml.math._shape.Shape, tuple, list, str] = 'names', down='', smooth=1, smooth_alpha=0.2, smooth_linewidth=2.0, size=(8, 6), transform: Callable = None, tight_layout=True, grid: Union[str, dict] = 'y', log_scale='', legend='upper right', x='steps', xlim=None, ylim=None, titles=True, labels: phiml.math._tensors.Tensor = None, xlabel: str = None, ylabel: str = None, colors: phiml.math._tensors.Tensor = 'default', dashed: phiml.math._tensors.Tensor = False)

Args

scene: str or Tensor. Scene paths containing the data to plot.
names: Data files to plot for each scene. The file must be located inside the scene directory and have the name log_<name>.txt.
reduce: Tensor dimensions along which all curves are plotted in the same diagram.
down: Tensor dimensions along which diagrams are ordered top-to-bottom instead of left-to-right.
smooth: int or Tensor. Number of data points to average, -1 for all.
smooth_alpha: Opacity of the non-smoothed curves under the smoothed curves.
smooth_linewidth: Line width of the smoothed curves.
size: Figure size in inches.
transform: Function T(x,y) -> (x,y) transforming the curves.
tight_layout:
grid:
log_scale:
legend:
x:
xlim:
ylim:
titles:
labels:
xlabel:
ylabel:
colors: Line colors as str, int or Tensor. Integers are interpreted as indices of the default color list.

Returns

MatPlotLib figure

Expand source code

def plot_scalars(scene: Union[str, tuple, list, Scene, math.Tensor],
                 names: Union[str, tuple, list, math.Tensor] = None,
                 reduce: Union[str, tuple, list, math.Shape] = 'names',
                 down='',
                 smooth=1,
                 smooth_alpha=0.2,
                 smooth_linewidth=2.,
                 size=(8, 6),
                 transform: Callable = None,
                 tight_layout=True,
                 grid: Union[str, dict] = 'y',
                 log_scale='',
                 legend='upper right',
                 x='steps',
                 xlim=None,
                 ylim=None,
                 titles=True,
                 labels: math.Tensor = None,
                 xlabel: str = None,
                 ylabel: str = None,
                 colors: math.Tensor = 'default',
                 dashed: math.Tensor = False):
    """

    Args:
        scene: `str` or `Tensor`. Scene paths containing the data to plot.
        names: Data files to plot for each scene. The file must be located inside the scene directory and have the name `log_<name>.txt`.
        reduce: Tensor dimensions along which all curves are plotted in the same diagram.
        down: Tensor dimensions along which diagrams are ordered top-to-bottom instead of left-to-right.
        smooth: `int` or `Tensor`. Number of data points to average, -1 for all.
        smooth_alpha: Opacity of the non-smoothed curves under the smoothed curves.
        smooth_linewidth: Line width of the smoothed curves.
        size: Figure size in inches.
        transform: Function `T(x,y) -> (x,y)` transforming the curves.
        tight_layout:
        grid:
        log_scale:
        legend:
        x:
        xlim:
        ylim:
        titles:
        labels:
        xlabel:
        ylabel:
        colors: Line colors as `str`, `int` or `Tensor`. Integers are interpreted as indices of the default color list.

    Returns:
        MatPlotLib [figure](https://matplotlib.org/stable/api/figure_api.html#matplotlib.figure.Figure)
    """
    warnings.warn("plot_scalars is deprecated. Use load_scalars() and plot() instead.", DeprecationWarning, stacklevel=2)
    scene = Scene.at(scene)
    additional_reduce = ()
    if names is None:
        first_path = next(iter(math.flatten(scene.paths)))
        names = [_str(n) for n in os.listdir(first_path)]
        names = [n[4:-4] for n in names if n.endswith('.txt') and n.startswith('log_')]
        names = math.wrap(names, batch('names'))
        additional_reduce = ['names']
    elif isinstance(names, str):
        names = math.wrap(names)
    elif isinstance(names, (tuple, list)):
        names = math.wrap(names, batch('names'))
    else:
        assert isinstance(names, math.Tensor), f"Invalid argument 'names': {type(names)}"
    colors = math.wrap(colors)
    dashed = math.wrap(dashed)
    if xlabel is None:
        xlabel = 'Iterations' if x == 'steps' else 'Time (s)'

    shape = (scene.shape & names.shape)
    batches = shape.without(reduce).without(additional_reduce)

    cycle = list(plt.rcParams['axes.prop_cycle'].by_key()['color'])
    fig, axes = plt.subplots(batches.only(down).volume, batches.without(down).volume, figsize=size)
    MATPLOTLIB.current_figure = fig
    axes = axes if isinstance(axes, numpy.ndarray) else np.array(axes)

    for b, axis in zip(math.concat_shapes(batches.only(down), batches.without(down)).meshgrid(), axes.flatten()):
        assert isinstance(axis, plt.Axes)
        names_equal = names[b].rank == 0
        paths_equal = scene.paths[b].rank == 0
        if titles is not None and titles is not False:
            if isinstance(titles, str):
                axis.set_title(titles)
            elif isinstance(titles, Tensor):
                axis.set_title(titles[b].native())
            elif names_equal:
                axis.set_title(display_name(names[b].native()))
            elif paths_equal:
                axis.set_title(os.path.basename(scene.paths[b].native()))
        if labels is not None:
            curve_labels = labels
        elif names_equal:
            curve_labels = math.map(os.path.basename, scene.paths[b])
        elif paths_equal:
            curve_labels = names[b]
        else:
            curve_labels = math.map(lambda p, n: f"{os.path.basename(p)} - {n}", scene.paths[b], names[b])

        def single_plot(name, path, label, i, color, dashed_, smooth):
            ML_LOGGER.debug(f"Reading {os.path.join(path, f'log_{name}.txt')}")
            curve = numpy.loadtxt(os.path.join(path, f"log_{name}.txt"))
            if curve.ndim == 2:
                x_values, values, *_ = curve.T
            else:
                values = curve
                x_values = np.arange(len(values))
            if x == 'steps':
                pass
            else:
                assert x == 'time', f"x must be 'steps' or 'time' but got {x}"
                ML_LOGGER.debug(f"Reading {os.path.join(path, 'log_step_time.txt')}")
                _, x_values, *_ = numpy.loadtxt(os.path.join(path, "log_step_time.txt")).T
                values = values[:len(x_values+1)]
                x_values = np.cumsum(x_values[:len(values)-1])
                x_values = np.concatenate([[0.], x_values])
            if transform:
                x_values, values = transform(np.stack([x_values, values]))
            if color == 'default':
                color = cycle[i]
            try:
                color = int(color)
            except ValueError:
                pass
            if isinstance(color, Number):
                color = cycle[int(color)]
            ML_LOGGER.debug(f"Plotting curve {label}")
            if smooth > 1:
                axis.plot(x_values, values, color=color, alpha=smooth_alpha, linewidth=1)
                curve = np.stack([x_values, values], -1)
                axis.plot(*smooth_uniform_curve(curve, smooth).T, *(['--'] if dashed_ else []), color=color, linewidth=smooth_linewidth, label=label)
            else:
                axis.plot(x_values, values, *(['--'] if dashed_ else []), color=color, linewidth=1, label=label)
            if grid:
                if isinstance(grid, dict):
                    axis.grid(**grid)
                else:
                    grid_axis = 'both' if 'x' in grid and 'y' in grid else grid
                    axis.grid(which='both', axis=grid_axis, linestyle='--', linewidth=size[1] * 0.3)
            if 'x' in log_scale:
                axis.set_xscale('log')
            if 'y' in log_scale:
                axis.set_yscale('log')
            if xlim:
                axis.set_xlim(xlim)
            if ylim:
                axis.set_ylim(ylim)
            if xlabel:
                axis.set_xlabel(xlabel)
            if ylabel:
                axis.set_ylabel(ylabel)
            return name

        math.map(single_plot, names[b], scene.paths[b], curve_labels, math.range_tensor(shape.after_gather(b)), colors, dashed, smooth)
        if legend:
            axis.legend(loc=legend)
    # Final touches
    if tight_layout:
        plt.tight_layout()
    return fig

def savefig(path: str, figure=None, dpi=120.0, close=False)

Save a figure to an image file.

Args

figure: Matplotlib or Plotly figure or text.
path: File path.
dpi: Pixels per inch.
close: Whether to close the figure after saving it.

Expand source code

def write_image(path: str, figure=None, dpi=120., close=False):
    """
    Save a figure to an image file.

    Args:
        figure: Matplotlib or Plotly figure or text.
        path: File path.
        dpi: Pixels per inch.
        close: Whether to close the figure after saving it.
    """
    figure = figure or LAST_FIGURE[0]
    if figure is None:
        figure = default_plots().current_figure
    assert figure is not None, "No figure to save."
    lib = get_plots_by_figure(figure)
    path = os.path.expanduser(path)
    lib.save(figure, path, dpi)
    if close:
        close_(figure=figure)

def show(*model: Union[phi.vis._vis_base.VisModel, phi.field._field.SampledField, tuple, list, phiml.math._tensors.Tensor, phi.geom._geom.Geometry], play=True, gui: Union[phi.vis._vis_base.Gui, str] = None, lib: Union[phi.vis._vis_base.Gui, str] = None, keep_alive=True, **config)

If model is a user interface model, launches the registered user interface. This will typically be the Dash web interface or the console interface if dash is not available. This method prepares the model before showing it. No more fields should be added to the vis after this method is invoked.

Args

model: (Optional) VisModel, the application or plottable object to display. If unspecified, shows the most recently plotted figure.
play: If true, invokes App.play(). The default value is False unless "autorun" is passed as a command line argument.
gui: Deprecated. Use lib instead. (optional) class of GUI to use
lib: Gui class or plotting library as str, e.g. 'matplotlib' or 'plotly'
keep_alive: Whether the GUI keeps the vis alive. If False, the program will exit when the main script is finished.
**config: additional GUI configuration parameters. For a full list of parameters, see the respective GUI documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html

Expand source code

def show(*model: Union[VisModel, SampledField, tuple, list, Tensor, Geometry],
         play=True,
         gui: Union[Gui, str] = None,
         lib: Union[Gui, str] = None,
         keep_alive=True,
         **config):
    """
    If `model` is a user interface model, launches the registered user interface.
    This will typically be the Dash web interface or the console interface if dash is not available.
    This method prepares the `model` before showing it. No more fields should be added to the vis after this method is invoked.

    See Also:
        `view()`.

    If `model` is plottable, e.g. a `SampledField` or `Tensor`, a figure is created and shown.
    If `model` is a figure, it is simply shown.

    See Also:
        `plot()`.

    This method may block until the GUI or plot window is closed.

    Also see the user interface documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html

    Args:
      model: (Optional) `VisModel`, the application or plottable object to display.
        If unspecified, shows the most recently plotted figure.
      play: If true, invokes `App.play()`. The default value is False unless "autorun" is passed as a command line argument.
      gui: Deprecated. Use `lib` instead. (optional) class of GUI to use
      lib: Gui class or plotting library as `str`, e.g. `'matplotlib'` or `'plotly'`
      keep_alive: Whether the GUI keeps the vis alive. If `False`, the program will exit when the main script is finished.
      **config: additional GUI configuration parameters.
        For a full list of parameters, see the respective GUI documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html
    """
    lib = lib if lib is not None else gui
    if len(model) == 1 and isinstance(model[0], VisModel):
        model[0].prepare()
        # --- Setup Gui ---
        gui = default_gui() if lib is None else get_gui(lib)
        gui.configure(config)
        gui.setup(model[0])
        if play:  # this needs to be done even if model cannot progress right now
            gui.auto_play()
        if gui.asynchronous:
            display_thread = Thread(target=lambda: gui.show(True), name="AsyncGui", daemon=not keep_alive)
            display_thread.start()
        else:
            gui.show(True)  # may be blocking call
    elif len(model) == 0:
        plots = default_plots() if lib is None else get_plots(lib)
        return plots.show(plots.current_figure)
    else:
        plots = default_plots() if lib is None else get_plots(lib)
        fig_tensor = plot(*model, lib=plots, **config)
        if isinstance(fig_tensor, Tensor):
            for fig in fig_tensor:
                plots.show(fig)
        else:
            return plots.show(fig_tensor)

def smooth(curves: phiml.math._tensors.Tensor, n: int) ‑> phiml.math._tensors.Tensor

Applies a smoothing kernel to curves, all channels independently.

Args

curves: Tensor containing at least one spatial dimension
n: Kernel size, i.e. number of values to average.

Returns

Smoothed curves as Tensor

Expand source code

@map_c2b
def smooth(curves: Tensor, n: int) -> Tensor:
    """
    Applies a smoothing kernel to curves, all channels independently.

    Args:
        curves: `Tensor` containing at least one spatial dimension
        n: Kernel size, i.e. number of values to average.

    Returns:
        Smoothed curves as `Tensor`
    """
    assert isinstance(n, int), f"n must be an int but got {n}"
    assert n >= 1, f"n must be at least 1 but got {n}"
    if n == 1:
        return curves
    kernel = ones(spatial(curves).with_sizes(n)) / n ** spatial(curves).rank
    return convolve(curves, kernel, extrapolation=extrapolation.SYMMETRIC_GRADIENT)

def view(*fields: Union[str, phi.field._field.SampledField], play: bool = True, gui=None, name: str = None, description: str = None, scene: Union[bool, phi.field._scene.Scene] = False, keep_alive=True, select: Union[str, tuple, list] = '', framerate=None, namespace=None, log_performance=True, **config) ‑> phi.vis._viewer.Viewer

Show fields in a graphical user interface.

fields may contain instances of Field or variable names of top-level variables (main module or Jupyter notebook). During loops, e.g. view().range(), the variable status is tracked and the GUI is updated.

When called from a Python script, name and description may be specified in the module docstring (string before imports). The first line is interpreted as the name, the rest as the subtitle. If not specified, a generic name and description is chosen.

Args

*fields: (Optional) Contents to be displayed. Either variable names or values. For field instances, all variables referencing the value will be shown. If not provided, the user namespace is searched for Field variables.
play: Whether to immediately start executing loops.
gui: (Optional) Name of GUI as str or GUI class. Built-in GUIs can be selected via 'dash', 'console'. See https://tum-pbs.github.io/PhiFlow/Visualization.html
name: (Optional) Name to display in GUI and use for the output directory if scene=True. Will be generated from the top-level script if not provided.
description: (Optional) Description to be displayed in the GUI. Will be generated from the top-level script if not provided.
scene: Existing Scene to write into or bool. If True, creates a new Scene in ~/phi/<name>
keep_alive: Whether the GUI should keep running even after the main thread finishes.
framerate: Target frame rate in Hz. Play will not step faster than the framerate. None for unlimited frame rate.
select: Dimension names along which one item to show is selected. Dimensions may be passed as tuple of str or as comma-separated names in a single str. For each select dimension, an associated selection slider will be created.
log_performance: Whether to measure and log the time each step takes. If True, will be logged as step_time to log_step_time.txt.
**config: Additional GUI configuration arguments.

Returns

Viewer

Expand source code

def view(*fields: Union[str, SampledField],
         play: bool = True,
         gui=None,
         name: str = None,
         description: str = None,
         scene: Union[bool, Scene] = False,
         keep_alive=True,
         select: Union[str, tuple, list] = '',
         framerate=None,
         namespace=None,
         log_performance=True,
         **config) -> Viewer:
    """
    Show `fields` in a graphical user interface.

    `fields` may contain instances of `Field` or variable names of top-level variables (main module or Jupyter notebook).
    During loops, e.g. `view().range()`, the variable status is tracked and the GUI is updated.

    When called from a Python script, name and description may be specified in the module docstring (string before imports).
    The first line is interpreted as the name, the rest as the subtitle.
    If not specified, a generic name and description is chosen.

    Args:
        *fields: (Optional) Contents to be displayed. Either variable names or values.
            For field instances, all variables referencing the value will be shown.
            If not provided, the user namespace is searched for Field variables.
        play: Whether to immediately start executing loops.
        gui: (Optional) Name of GUI as `str` or GUI class.
            Built-in GUIs can be selected via `'dash'`, `'console'`.
            See https://tum-pbs.github.io/PhiFlow/Visualization.html
        name: (Optional) Name to display in GUI and use for the output directory if `scene=True`.
            Will be generated from the top-level script if not provided.
        description: (Optional) Description to be displayed in the GUI.
            Will be generated from the top-level script if not provided.
        scene: Existing `Scene` to write into or `bool`. If `True`, creates a new Scene in `~/phi/<name>`
        keep_alive: Whether the GUI should keep running even after the main thread finishes.
        framerate: Target frame rate in Hz. Play will not step faster than the framerate. `None` for unlimited frame rate.
        select: Dimension names along which one item to show is selected.
            Dimensions may be passed as `tuple` of `str` or as comma-separated names in a single `str`.
            For each `select` dimension, an associated selection slider will be created.
        log_performance: Whether to measure and log the time each step takes.
            If `True`, will be logged as `step_time` to `log_step_time.txt`.
        **config: Additional GUI configuration arguments.

    Returns:
        `Viewer`
    """
    default_namespace = get_user_namespace(1)
    user_namespace = default_namespace if namespace is None else DictNamespace(namespace,
                                                                               title=default_namespace.get_title(),
                                                                               description=default_namespace.get_description(),
                                                                               reference=default_namespace.get_reference())
    variables = _default_field_variables(user_namespace, fields)
    actions = dict(ACTIONS)
    ACTIONS.clear()
    if scene is False:
        scene = None
    elif scene is True:
        scene = Scene.create(os.path.join("~", "phi", _slugify_filename(name or user_namespace.get_reference())))
        print(f"Created scene at {scene}")
    else:
        assert isinstance(scene, Scene)
    name = name or user_namespace.get_title()
    description = description or user_namespace.get_description()
    gui = default_gui() if gui is None else get_gui(gui)
    controls = tuple(c for c in sorted(CONTROL_VARS.values(), key=lambda c: c.name) if
                     user_namespace.get_variable(c.name) is not None)
    CONTROL_VARS.clear()
    viewer = create_viewer(user_namespace, variables, name, description, scene, asynchronous=gui.asynchronous,
                           controls=controls, actions=actions, log_performance=log_performance)
    show(viewer, play=play, gui=gui, keep_alive=keep_alive, framerate=framerate, select=select, **config)
    return viewer

def write_image(path: str, figure=None, dpi=120.0, close=False)

Save a figure to an image file.

Args

figure: Matplotlib or Plotly figure or text.
path: File path.
dpi: Pixels per inch.
close: Whether to close the figure after saving it.

Expand source code

def write_image(path: str, figure=None, dpi=120., close=False):
    """
    Save a figure to an image file.

    Args:
        figure: Matplotlib or Plotly figure or text.
        path: File path.
        dpi: Pixels per inch.
        close: Whether to close the figure after saving it.
    """
    figure = figure or LAST_FIGURE[0]
    if figure is None:
        figure = default_plots().current_figure
    assert figure is not None, "No figure to save."
    lib = get_plots_by_figure(figure)
    path = os.path.expanduser(path)
    lib.save(figure, path, dpi)
    if close:
        close_(figure=figure)

Classes

class Viewer

Shows variables from the user namespace. To create a Viewer, call view() from the top-level Python script or from a notebook.

Use Viewer.range() to control the loop execution from the user interface.

Also see the user interface documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html

Expand source code

class Viewer(VisModel):
    """
    Shows variables from the user namespace.
    To create a `Viewer`, call `phi.vis.view()` from the top-level Python script or from a notebook.

    Use `Viewer.range()` to control the loop execution from the user interface.

    Also see the user interface documentation at https://tum-pbs.github.io/PhiFlow/Visualization.html
    """

    def __init__(self,
                 namespace: UserNamespace,
                 fields: dict,
                 name: str,
                 description: str,
                 scene: Scene,
                 controls: tuple,
                 actions: dict,
                 log_performance: bool,
                 ):
        VisModel.__init__(self, name, description, scene=scene)
        self.initial_field_values = fields
        self._controls = controls
        self.namespace = namespace
        self.log_performance = log_performance
        self._rec = None
        self._in_loop = False
        self._log = SceneLog(self.scene)
        self.log_file = self._log.log_file
        self._elapsed = None
        self.reset_step = 0
        self._actions = {}
        custom_reset = False
        self.reset_count = 0
        for action, function in actions.items():
            if action.name == 'reset':
                self._actions[action] = partial(self.reset, custom_reset=function)
                custom_reset = True
            else:
                self._actions[action] = function
        if not custom_reset:
            self._actions[Action('reset', Viewer.reset.__doc__)] = self.reset

    def log_scalars(self, reduce=math.mean, **values):
        self._log.log_scalars(self.steps, reduce=reduce, **values)

    def info(self, message: str):  # may be replaced by a different solution later on
        """
        Update the status message.
        The status message is written to the console and the log file.
        Additionally, it may be displayed by the user interface.

        See `debug()`.

        Args:
            message: Message to display
        """
        message = str(message)
        self.message = message
        self._log.log(message)

    def __rrshift__(self, other):
        self.info(other)

    @property
    def field_names(self) -> tuple:
        return tuple(self.initial_field_values.keys())

    def get_field(self, name, dim_selection: dict) -> SampledField:
        if name not in self.initial_field_values:
            raise KeyError(name)
        if self._rec:
            value = self._rec[name]
        else:
            value = self.namespace.get_variable(name)
        if callable(value):
            value = value()
        if isinstance(value, (SampledField, Tensor)):
            value = value[dim_selection]
        return value

    @property
    def curve_names(self) -> tuple:
        return self._log.scalar_curve_names

    def get_curve(self, name: str) -> tuple:
        return self._log.get_scalar_curve(name)

    @property
    def controls(self) -> Tuple[Control]:
        return self._controls

    def get_control_value(self, name):
        return self.namespace.get_variable(name)

    def set_control_value(self, name, value):
        self.namespace.set_variable(name, value)

    @property
    def actions(self) -> tuple:
        return tuple(self._actions.keys())

    def run_action(self, name):
        for action, fun in self._actions.items():
            if action.name == name:
                fun()
                return
        raise KeyError(name)

    def range(self, *args, warmup=0, **rec_dim):
        """
        Similarly to `range()`, returns a generator that can be used in a `for` loop.

        >>> for step in ModuleViewer().range(100):
        >>>     print(f'Running step {step}')

        However, `Viewer.range()` enables controlling the flow via the user interface.
        Each element returned by the generator waits for `progress` to be invoked once.

        Note that `step` is always equal to `Viewer.steps`.

        This method can be invoked multiple times.
        However, do not call this method while one `range` is still active.

        Args:
            *args: Either no arguments for infinite loop or single `int` argument `stop`.
                Must be empty if `rec_dim` is used.
            **rec_dim: Can be used instead of `*args` to record values along a new batch dimension of this name.
                The recorded values can be accessed as `Viewer.rec.<name>` or `Viewer.rec['<name>']`.
            warmup: Number of uncounted loop iterations to perform before `step()` is invoked for the first time.

        Yields:
            Step count of `Viewer`.
        """
        for _ in range(warmup):
            yield self.steps

        self._in_loop = True
        self._call(self.progress_available)

        if rec_dim:
            assert len(rec_dim) == 1, f"Only one rec_dim allowed but got {rec_dim}"
            assert not args, f"No positional arguments are allowed when a rec_dim is specified. {rec_dim}"
            rec_dim_name = next(iter(rec_dim.keys()))
            size = rec_dim[rec_dim_name]
            assert isinstance(size, int)
            self._rec = Record(rec_dim_name)
            self._rec.append(self.initial_field_values, warn_missing=False)
            args = [size]
            self.growing_dims = [rec_dim_name]

        if len(args) == 0:
            def count():
                i = 0
                while True:
                    yield i
                    i += 1

            step_source = count()
        else:
            step_source = range(*args)

        try:
            for step in step_source:
                self.steps = step - self.reset_step
                try:
                    self._pre_step()
                    t = time.perf_counter()
                    yield step - self.reset_step
                    self._elapsed = time.perf_counter() - t
                    self.steps = step - self.reset_step + 1
                    if rec_dim:
                        self._rec.append({name: self.namespace.get_variable(name) for name in self.field_names})
                    if self.log_performance:
                        self._log.log_scalars(self.steps, reduce=None, step_time=self._elapsed)
                finally:
                    self._post_step()
        finally:
            self._in_loop = False
            self._call(self.progress_unavailable)

    def _pre_step(self):
        self._call(self.pre_step)

    def _post_step(self):
        self._call(self.post_step)

    @property
    def rec(self) -> 'Record':
        """
        Read recorded fields as `viewer.rec.<name>`.
        Accessing `rec` without having started a recording using `Viewer.range()` raises an `AssertionError`.
        """
        assert self._rec, "Enable recording by calling range() with a dimension name, e.g. 'range(frames=10)'."
        return self._rec

    def progress(self):
        raise AssertionError("progress() not supported by synchronous Viewer.")

    @property
    def can_progress(self) -> bool:
        return self._in_loop

    def reset(self, custom_reset=None):
        """
        Restores all viewed fields to the states they were in when the viewer was created.
        Changes variable values in the user namespace.
        """
        if custom_reset:
            custom_reset()
        for name, value in self.initial_field_values.items():
            self.namespace.set_variable(name, value)
        self.reset_step += self.steps
        self.steps = 0
        self.reset_count += 1

Ancestors

phi.vis._vis_base.VisModel

Subclasses

phi.vis._viewer.AsyncViewer

Instance variables

var rec : phi.vis._viewer.Record

Read recorded fields as viewer.rec.<name>. Accessing rec without having started a recording using Viewer.range() raises an AssertionError.

Expand source code

@property
def rec(self) -> 'Record':
    """
    Read recorded fields as `viewer.rec.<name>`.
    Accessing `rec` without having started a recording using `Viewer.range()` raises an `AssertionError`.
    """
    assert self._rec, "Enable recording by calling range() with a dimension name, e.g. 'range(frames=10)'."
    return self._rec

Methods

def range(self, *args, warmup=0, **rec_dim)

Similarly to range(), returns a generator that can be used in a for loop.

>>> for step in ModuleViewer().range(100):
>>>     print(f'Running step {step}')

However, Viewer.range() enables controlling the flow via the user interface. Each element returned by the generator waits for progress to be invoked once.

Note that step is always equal to Viewer.steps.

This method can be invoked multiple times. However, do not call this method while one range is still active.

Args

*args: Either no arguments for infinite loop or single int argument stop. Must be empty if rec_dim is used.
**rec_dim: Can be used instead of *args to record values along a new batch dimension of this name. The recorded values can be accessed as Viewer.rec.<name> or Viewer.rec['<name>'].
warmup: Number of uncounted loop iterations to perform before step() is invoked for the first time.

Yields

Step count of Viewer.

Expand source code

def range(self, *args, warmup=0, **rec_dim):
    """
    Similarly to `range()`, returns a generator that can be used in a `for` loop.

    >>> for step in ModuleViewer().range(100):
    >>>     print(f'Running step {step}')

    However, `Viewer.range()` enables controlling the flow via the user interface.
    Each element returned by the generator waits for `progress` to be invoked once.

    Note that `step` is always equal to `Viewer.steps`.

    This method can be invoked multiple times.
    However, do not call this method while one `range` is still active.

    Args:
        *args: Either no arguments for infinite loop or single `int` argument `stop`.
            Must be empty if `rec_dim` is used.
        **rec_dim: Can be used instead of `*args` to record values along a new batch dimension of this name.
            The recorded values can be accessed as `Viewer.rec.<name>` or `Viewer.rec['<name>']`.
        warmup: Number of uncounted loop iterations to perform before `step()` is invoked for the first time.

    Yields:
        Step count of `Viewer`.
    """
    for _ in range(warmup):
        yield self.steps

    self._in_loop = True
    self._call(self.progress_available)

    if rec_dim:
        assert len(rec_dim) == 1, f"Only one rec_dim allowed but got {rec_dim}"
        assert not args, f"No positional arguments are allowed when a rec_dim is specified. {rec_dim}"
        rec_dim_name = next(iter(rec_dim.keys()))
        size = rec_dim[rec_dim_name]
        assert isinstance(size, int)
        self._rec = Record(rec_dim_name)
        self._rec.append(self.initial_field_values, warn_missing=False)
        args = [size]
        self.growing_dims = [rec_dim_name]

    if len(args) == 0:
        def count():
            i = 0
            while True:
                yield i
                i += 1

        step_source = count()
    else:
        step_source = range(*args)

    try:
        for step in step_source:
            self.steps = step - self.reset_step
            try:
                self._pre_step()
                t = time.perf_counter()
                yield step - self.reset_step
                self._elapsed = time.perf_counter() - t
                self.steps = step - self.reset_step + 1
                if rec_dim:
                    self._rec.append({name: self.namespace.get_variable(name) for name in self.field_names})
                if self.log_performance:
                    self._log.log_scalars(self.steps, reduce=None, step_time=self._elapsed)
            finally:
                self._post_step()
    finally:
        self._in_loop = False
        self._call(self.progress_unavailable)