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, animate, plot_scalars, savefig
from ._vis import view, control, show, action

__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
def animate(fields: phi.field._field.SampledField, dim='frames', repeat=True, interval=200, title=False, size=(8, 6), show_color_bar=False, same_scale=True, **plt_args) ‑> matplotlib.animation.Animation

Creates a Matplotlib animation from fields. fields may be a sequence of frames or a single SampledField instances with a frames dimension.

Args

fields
SampledField with frames dimension or tuple or list of SampledField.
dim
Time dimension to animate (default='frames').
repeat
Whether the video should loop.
interval
Frame time in milliseconds.
title
Figure/sub-figure title. If str or tuple/list of str. True to generate a title automatically.
size
Figure size
show_color_bar
Whether to show a color bar
same_scale
Whether to use the same scale, both temporally and for all sub-figures.
**plt_args
Further plotting arguments, see plot().

Returns

Matplotlib Animation

Expand source code
def animate(fields: SampledField,
            dim='frames',
            repeat=True,
            interval=200,
            title=False,
            size=(8, 6),
            show_color_bar=False,
            same_scale=True,
            **plt_args) -> animation.Animation:
    """
    Creates a Matplotlib animation from `fields`.
    `fields` may be a sequence of frames or a single `SampledField` instances with a `frames` dimension.

    Args:
        fields: `SampledField` with `frames` dimension or `tuple` or `list` of `SampledField`.
        dim: Time dimension to animate (default=`'frames'`).
        repeat: Whether the video should loop.
        interval: Frame time in milliseconds.
        title: Figure/sub-figure title. If `str` or `tuple`/`list` of `str`. `True` to generate a title automatically.
        size: Figure size
        show_color_bar: Whether to show a color bar
        same_scale: Whether to use the same scale, both temporally and for all sub-figures.
        **plt_args: Further plotting arguments, see `plot()`.

    Returns:
        Matplotlib `Animation`
    """
    assert isinstance(fields, SampledField)
    assert dim in fields.shape, f"Animation dimension {dim} not present in data."
    fields = list(fields.unstack(dim))
    fig, _ = _subplots(fields[0], size, None)

    def func(frame: int):
        field = fields[frame]
        for axis in fig.axes:
            axis.clear()
        plot(field, existing_figure=fig, title=title, show_color_bar=show_color_bar, same_scale=same_scale, **plt_args)

    ani = animation.FuncAnimation(fig, func, init_func=lambda: fig.axes, repeat=repeat, frames=len(fields), interval=interval)
    plt.close(fig)
    return ani
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")

The value o

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 and float values.
description
Description of what the control does.
**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:
    ```python
    dt = control(1.0, (0.1, 10), name="Time increment")
    ```

    The value o

    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` and `float` values.
        description: Description of what the control does.
        **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 plot(field: phi.field._field.SampledField, title=False, size=(12, 5), show_color_bar=True, same_scale=True, existing_figure: matplotlib.figure.Figure = None, **plt_args)

Creates a Matplotlib figure to display a single field or batch of fields.

Use matplotlib.pyplot.show() or matplotlib.pyplot.savefig() to view the figure.

Args

field
SampledField, may contain batch dimensions which will create sub-figures.
title
Figure/sub-figure title. If str or tuple/list of str. True to generate a title automatically.
show_color_bar
Whether to show a colorbar for heatmap plots.
size
Figure (width, height) in inches.
same_scale
Whether to use the same value scale for all subplots.
existing_figure
Existing Matplotlib figure to add this plot to. Figure will not be cleared before plotting.
**plt_args
Additional plotting arguments passed to Matplotlib.

Returns

Matplotlib figure.

Expand source code
def plot(field: SampledField or tuple or list,
         title=False,
         size=(12, 5),
         show_color_bar=True,
         same_scale=True,
         existing_figure: plt.Figure or None = None,
         **plt_args):
    """
    Creates a Matplotlib figure to display a single field or batch of fields.

    Use [`matplotlib.pyplot.show()`](https://matplotlib.org/3.1.1/api/_as_gen/matplotlib.pyplot.show.html) or
    [`matplotlib.pyplot.savefig()`](https://matplotlib.org/stable/api/_as_gen/matplotlib.pyplot.savefig.html) to view the figure.

    Args:
        field: `SampledField`, may contain batch dimensions which will create sub-figures.
        title: Figure/sub-figure title. If `str` or `tuple`/`list` of `str`. `True` to generate a title automatically.
        show_color_bar: Whether to show a colorbar for heatmap plots.
        size: Figure (width, height) in inches.
        same_scale: Whether to use the same value scale for all subplots.
        existing_figure: Existing Matplotlib figure to add this plot to. Figure will not be cleared before plotting.
        **plt_args: Additional plotting arguments passed to Matplotlib.

    Returns:
        [Matplotlib figure](https://matplotlib.org/stable/api/figure_api.html#matplotlib.figure.Figure).
    """
    fig, fields = _subplots(field, size, existing_figure=existing_figure)
    if title:
        for b in range(len(fig.axes)):
            if isinstance(title, str):
                sub_title = title
            elif title is True:
                sub_title = f"{b} of {field.shape.batch}"
            elif isinstance(title, (tuple, list)):
                sub_title = title[b]
            else:
                sub_title = None
            if sub_title is not None:
                fig.axes[b].set_title(sub_title)
    if same_scale and any(isinstance(f, Grid) for f in fields):
        min_val = min([float(f.values.min) for f in fields if isinstance(f, Grid)])
        max_val = min([float(f.values.max) for f in fields if isinstance(f, Grid)])
    else:
        min_val, max_val = None, None
    for axis, field in zip(fig.axes, fields):
        _plot(axis, field, show_color_bar=show_color_bar, vmin=min_val, vmax=max_val, **plt_args)
    plt.tight_layout()
    return fig
def plot_scalars(scene: str, names: str = None, reduce: str = 'names', down='', smooth=1, smooth_alpha=0.2, smooth_linewidth=2.0, size=(8, 6), transform: Callable = None, tight_layout=True, grid='y', log_scale='', legend='upper right', x='steps', xlim=None, ylim=None, titles=True, labels: phi.math._tensors.Tensor = None, xlabel: str = None, ylabel: str = None, colors: phi.math._tensors.Tensor = 'default')
Expand source code
def plot_scalars(scene: str or tuple or list or Scene or math.Tensor,
                 names: str or tuple or list or math.Tensor = None,
                 reduce: str or tuple or list or math.Shape = 'names',
                 down='',
                 smooth=1,
                 smooth_alpha=0.2,
                 smooth_linewidth=2.,
                 size=(8, 6),
                 transform: Callable = None,
                 tight_layout=True,
                 grid='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'):
    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)}"
    if not isinstance(colors, math.Tensor):
        colors = math.wrap(colors)
    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)
    axes = axes if isinstance(axes, numpy.ndarray) else [axes]

    for b, axis in zip(batches.meshgrid(), axes):
        assert isinstance(axis, plt.Axes)
        names_equal = names[b].rank == 0
        paths_equal = scene.paths[b].rank == 0
        if titles:
            if isinstance(titles, str):
                axis.set_title(titles)
            elif names_equal:
                axis.set_title(display_name(str(names[b])))
            elif paths_equal:
                axis.set_title(os.path.basename(str(scene.paths[b])))
        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):
            logging.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}"
                logging.debug(f"Reading {os.path.join(path, 'log_step_time.txt')}")
                _, x_values, *_ = numpy.loadtxt(os.path.join(path, "log_step_time.txt")).T
                x_values = np.cumsum(x_values[:len(values)])
            if transform:
                x_values, values = transform(np.stack([x_values, values]))
            if color == 'default':
                color = cycle[i]
            elif isinstance(color, Number):
                color = cycle[int(color)]
            logging.debug(f"Plotting curve {label}")
            axis.plot(x_values, values, color=color, alpha=smooth_alpha, linewidth=1)
            axis.plot(*smooth_uniform_curve(x_values, values, n=smooth), color=color, linewidth=smooth_linewidth, label=label)
            if grid:
                grid_axis = 'both' if 'x' in grid and 'y' in grid else grid
                axis.grid(which='both', axis=grid_axis, linestyle='--')
            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)
        if legend:
            axis.legend(loc=legend)
    # Final touches
    if tight_layout:
        plt.tight_layout()
    return fig
def savefig(filename: str, transparent=True)
Expand source code
def savefig(filename: str, transparent=True):
    plt.savefig(filename, transparent=transparent)
def show(model: phi.vis._vis_base.VisModel = None, play=True, gui: phi.vis._vis_base.Gui = None, keep_alive=True, **config)

Launch the registered user interface (web interface by default).

This method may block until the GUI is closed.

This method prepares the vis before showing it. No more fields should be added to the vis after this method is invoked.

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

Args

model
(Optional) VisModel, the application to display. If unspecified, searches the calling script for a subclass of App and instantiates it.
play
If true, invokes App.play(). The default value is False unless "autorun" is passed as a command line argument.
gui
(optional) class of GUI to use
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: VisModel or None = None, play=True, gui: Gui or str = None, keep_alive=True, **config):
    """
    Launch the registered user interface (web interface by default).

    This method may block until the GUI is closed.

    This method prepares the vis before showing it. No more fields should be added to the vis after this method is invoked.

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

    Args:
      model: (Optional) `VisModel`, the application to display. If unspecified, searches the calling script for a subclass of App and instantiates it.
      play: If true, invokes `App.play()`. The default value is False unless "autorun" is passed as a command line argument.
      gui: (optional) class of GUI to use
      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
    """
    if model is None:
        import pylab
        pylab.show()
        return
    assert isinstance(model, VisModel), f"show() first argument must be an App instance but got {model}"
    model.prepare()
    # --- Setup Gui ---
    gui = default_gui() if gui is None else get_gui(gui)
    gui.configure(config)
    gui.setup(model)
    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
def view(*fields: str, play: bool = True, gui=None, name: str = None, description: str = None, scene: bool = False, keep_alive=True, select: str = '', framerate=None, namespace=None, **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' and 'widgets'. 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.
**config
Additional GUI configuration arguments.

Returns

Viewer

Expand source code
def view(*fields: str or SampledField,
         play: bool = True,
         gui=None,
         name: str = None,
         description: str = None,
         scene: bool or Scene = False,
         keep_alive=True,
         select: str or tuple or list = '',
         framerate=None,
         namespace=None,
         **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'` and `'widgets'`.
            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.
        **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=True)
    show(viewer, play=play, gui=gui, keep_alive=keep_alive, framerate=framerate, select=select, **config)
    return viewer

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, **values):
        self._log.log_scalars(self.steps, **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 isinstance(value, SampledField):
            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.

        ```python
        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, 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.

    ```python
    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, step_time=self._elapsed)
            finally:
                self._post_step()
    finally:
        self._in_loop = False
        self._call(self.progress_unavailable)