r"""Module for generating a Python state for ParaView.
This module uses paraview.smtrace to generate a trace for a selected set of
proxies by mimicking the creation of various pipeline components in sequence.
Typical usage of this module is as follows::
from paraview import smstate
state = smstate.get_state()
print (state)
Note, this cannot be called when Python tracing is active.
"""
from paraview import servermanager as sm
from paraview import smtrace
from paraview import simple
from paraview.util import proxy as proxy_util
RECORD_MODIFIED_PROPERTIES = sm.vtkSMTrace.RECORD_MODIFIED_PROPERTIES
RECORD_ALL_PROPERTIES = sm.vtkSMTrace.RECORD_ALL_PROPERTIES
RECORD_ACTIVE_MODIFIED_PROPERTIES = sm.vtkSMTrace.RECORD_ACTIVE_MODIFIED_PROPERTIES
[docs]
class visible_representations(object):
"""filter object to skip hidden representations from being saved in state file"""
def __call__(self, proxy):
if not proxy_util.supported_proxies()(proxy): return False
try:
return proxy.Visibility
except AttributeError:
pass
return True
[docs]
def get_consumers(proxy, filter, consumer_set, recursive=True):
"""Returns the consumers for a proxy iteratively. If filter is non-None,
filter is used to cull consumers."""
return proxy_util.get_consumers(proxy, filter, consumer_set, recursive)
[docs]
def get_producers(proxy, filter, producer_set):
"""Returns the producers for a proxy iteratively. If filter is non-None,
filter is used to cull producers."""
return proxy_util.get_producers(proxy, filter, producer_set)
def _get_layout_for_saving_all_views(proxy):
"""Returns the layout whose views are all saved in a single image by the
proxy, if it is an image extractor with 'SaveAllViews' enabled. Otherwise,
returns None."""
if proxy.GetXMLGroup() != "extractors":
return None
try:
if not proxy.Writer.SaveAllViews:
return None
view = proxy.Producer
except AttributeError:
return None
if not view:
return None
return sm._getPyProxy(sm.vtkSMViewLayoutProxy.FindLayout(view.SMProxy))
[docs]
def get_state(options=None, source_set=[], filter=None, raw=False,
preamble=None, postamble=None):
"""Returns the state string"""
if options:
options = sm._getPyProxy(options)
propertiesToTraceOnCreate = options.PropertiesToTraceOnCreate.ConvertValue(options.PropertiesToTraceOnCreate.GetElement(0))
skipHiddenRepresentations = options.SkipHiddenDisplayProperties
skipRenderingComponents = options.SkipRenderingComponents
skipActiveComponents = options.SkipActiveComponents
skipLayoutComponents = options.SkipLayoutComponents
skipAnimationComponents = options.SkipAnimationComponents
else:
propertiesToTraceOnCreate = RECORD_MODIFIED_PROPERTIES
skipHiddenRepresentations = True
skipRenderingComponents = False
skipActiveComponents = False
skipLayoutComponents = False
skipAnimationComponents = False
# essential to ensure any obsolete accessors don't linger - can cause havoc
# when saving state following a Python trace session
# (paraview/paraview#18994)
import gc
gc.collect()
if sm.vtkSMTrace.GetActiveTracer():
raise RuntimeError("Cannot generate Python state when tracing is active.")
if filter is None:
filter = visible_representations() if skipHiddenRepresentations else proxy_util.supported_proxies()
# build a set of proxies of interest
if source_set:
start_set = source_set
else:
# if nothing is specified, we save all views, sources, selections and animation scene
start_set = [x for x in simple.GetSources().values()]
start_set += [x for x in simple.GetSelections().values()]
start_set += simple.GetViews()
start_set += [simple.GetAnimationScene()]
start_set = [x for x in start_set if filter(x)]
# now, locate dependencies for the start_set, pruning irrelevant branches
consumers = set(start_set)
for proxy in start_set:
get_consumers(proxy, filter, consumers)
producers = set()
for proxy in consumers:
get_producers(proxy, filter, producers)
# image extractors that save all views need the layout and all the views in
# it, which are not dependencies tracked by the proxy manager.
for proxy in list(consumers.union(producers)):
layout = _get_layout_for_saving_all_views(proxy)
if layout is None or layout in producers or not filter(layout):
continue
producers.add(layout)
for view in simple.GetViewsInLayout(layout):
if view not in producers and filter(view):
producers.add(view)
get_producers(view, filter, producers)
# proxies_of_interest is set of all proxies that we should trace.
proxies_of_interest = producers.union(consumers)
previous_target_type = smtrace.Trace.output_target_type
smtrace.Trace.output_target_type = smtrace.TraceTargetType.STATEFILE
tracer = smtrace.ScopedTracer(preamble="")
trace_result = ""
with tracer:
# this ensures that lookup tables/scalar bars etc. are fully traced.
tracer.config.SetFullyTraceSupplementalProxies(True)
tracer.config.SetSkipRenderingComponents(skipRenderingComponents)
tracer.config.SetPropertiesToTraceOnCreate(propertiesToTraceOnCreate)
trace = smtrace.TraceOutput()
if preamble is None:
trace.append("# state file generated using %s" % simple.GetParaViewSourceVersion())
trace.append("import paraview")
trace.append("paraview.compatibility.major = %d" % sm.vtkSMProxyManager.GetVersionMajor())
trace.append("paraview.compatibility.minor = %d" % sm.vtkSMProxyManager.GetVersionMinor())
elif preamble:
trace.append(preamble)
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# --------------------------------------------------------------------------
# We trace the views and layouts, if any.
if skipRenderingComponents:
views = []
else:
views = [x for x in proxies_of_interest if smtrace.Trace.get_registered_name(x, "views")]
if views:
# sort views by their names, so the state has some structure to it.
views = sorted(views, key=lambda x: \
smtrace.Trace.get_registered_name(x, "views"))
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup views used in the visualization",
"# ----------------------------------------------------------------"])
for view in views:
# FIXME: save view camera positions and size.
traceitem = smtrace.RegisterViewProxy(view)
traceitem.finalize()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
if not skipActiveComponents:
trace.append_separated(["SetActiveView(None)"])
# from views, build the list of layouts of interest.
layouts = set()
for aview in views:
l = simple.GetLayout(aview)
if l:
layouts.add(simple.GetLayout(aview))
if skipLayoutComponents:
# still trace layouts needed by image extractors that save all views.
layouts = set([x for x in layouts if x in proxies_of_interest])
# trace create of layouts
if layouts:
layouts = sorted(layouts, key=lambda x: \
smtrace.Trace.get_registered_name(x, "layouts"))
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup view layouts",
"# ----------------------------------------------------------------"])
for layout in layouts:
traceitem = smtrace.RegisterLayoutProxy(layout)
traceitem.finalize(filter=lambda x: x in views)
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
if views and not skipActiveComponents:
# restore the active view after the layouts have been created.
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# restore active view",
"SetActiveView(%s)" % smtrace.Trace.get_accessor(simple.GetActiveView()),
"# ----------------------------------------------------------------"])
# --------------------------------------------------------------------------
# Next, trace selection sources.
sorted_proxies_of_interest = proxy_util.toposort(proxies_of_interest)
sorted_selection_sources = [x for x in sorted_proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "selection_sources")]
if sorted_selection_sources:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup the selection sources",
"# ----------------------------------------------------------------"])
for selection in sorted_selection_sources:
traceitem = smtrace.RegisterSelectionProxy(selection)
traceitem.finalize("selection_sources")
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# --------------------------------------------------------------------------
# Next, trace selections.
sorted_selections = [x for x in sorted_proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "selections")]
selected_source_id = None
if sorted_selections and len(sorted_selections) > 1:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup the selections",
"# ----------------------------------------------------------------"])
for selection in sorted_selections:
traceitem = smtrace.RegisterSelectionProxy(selection)
traceitem.finalize("selections")
if selection.GetXMLName() == "AppendSelections":
selected_source_id = selection.GetSelectionId()
selected_source_port = selection.GetSelectionPort()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# --------------------------------------------------------------------------
# Next, trace data processing pipelines.
sorted_sources = [x for x in sorted_proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "sources")]
if sorted_sources:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup the data processing pipelines",
"# ----------------------------------------------------------------"])
for source in sorted_sources:
traceitem = smtrace.RegisterPipelineProxy(source, saving_state=True)
traceitem.finalize()
if source.GetGlobalID() == selected_source_id:
selected_source_varname = traceitem.varname
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# --------------------------------------------------------------------------
# Link selections to selected source
if sorted_selections and len(sorted_selections) > 1:
trace.append_separated([ \
f"appendSelections.SetSelectionId({selected_source_varname}.GetGlobalID())",
f"appendSelections.SetSelectionPort({selected_source_port})"])
# --------------------------------------------------------------------------
# Can't decide if the representations should be saved with the pipeline
# objects or afterwards, opting for afterwards for now since the topological
# sort doesn't guarantee that the representations will follow their sources
# anyways.
sorted_representations = [x for x in sorted_proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "representations")]
scalarbar_representations = [x for x in sorted_proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "scalar_bars")]
# print ("sorted_representations", sorted_representations)
# print ("scalarbar_representations", scalarbar_representations)
if not skipRenderingComponents and (sorted_representations or scalarbar_representations):
for view in views:
view_representations = [x for x in view.Representations if x in sorted_representations]
view_scalarbars = [x for x in view.Representations if x in scalarbar_representations]
if view_representations or view_scalarbars:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup the visualization in view '%s'" % smtrace.Trace.get_accessor(view),
"# ----------------------------------------------------------------"])
for rep in view_representations:
try:
producer = rep.Input
port = rep.Input.Port
traceitem = smtrace.Show(producer, port, view, rep,
comment="show data from %s" % smtrace.Trace.get_accessor(producer))
traceitem.finalize()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
if rep.UseSeparateColorMap:
trace.append_separated([ \
"# set separate color map",
"%s.UseSeparateColorMap = True" % ( \
smtrace.Trace.get_accessor(rep))])
if rep.BlockUseSeparateColorMaps:
selectorsAndUseSeparateColorMap = rep.BlockUseSeparateColorMaps.GetData()
if len(selectorsAndUseSeparateColorMap) % 2 == 0 and len(selectorsAndUseSeparateColorMap) > 0:
trace.append_separated([ \
"# set separate color map for blocks",
"%s.BlockUseSeparateColorMaps = %s" % ( \
smtrace.Trace.get_accessor(rep), ", ".join(selectorsAndUseSeparateColorMap))])
except AttributeError:
pass
# save the scalar bar properties themselves.
if view_scalarbars:
trace.append_separated("# setup the color legend parameters for each legend in this view")
for rep in view_scalarbars:
smtrace.Trace.get_accessor(rep)
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
trace.append_separated([ \
"# set color bar visibility", "%s.Visibility = %s" % ( \
smtrace.Trace.get_accessor(rep), rep.Visibility)])
for rep in view_representations:
try:
producer = rep.Input
port = rep.Input.Port
if rep.IsScalarBarVisible(view):
# FIXME: this will save this multiple times, right now,
# if two representations use the same LUT.
# If someone wants to fix it, also fix it below for blocks.
trace.append_separated([ \
"# show color legend",
"%s.SetScalarBarVisibility(%s, True)" % ( \
smtrace.Trace.get_accessor(rep),
smtrace.Trace.get_accessor(view))])
if rep.BlockColorArrayNames:
selectors = rep.BlockColorArrayNames.GetData()[::3]
isScalarVisibles = rep.IsBlocksScalarBarVisible(view, selectors)
if all(isScalarVisible == 1 for isScalarVisible in isScalarVisibles):
trace.append_separated([ \
"# set color map for blocks",
"%s.SetBlocksScalarBarVisibility(%s, %s, True)" % ( \
smtrace.Trace.get_accessor(rep),
smtrace.Trace.get_accessor(view),
str(selectors))])
else:
for i in range(len(selectors)):
if isScalarVisibles[i]:
trace.append_separated([ \
"# set color map for block %s" % selectors[i],
"%s.SetBlockScalarBarVisibility(%s, %s, True)" % ( \
smtrace.Trace.get_accessor(rep),
smtrace.Trace.get_accessor(view),
selectors[i])])
if not rep.Visibility:
traceitem = smtrace.Hide(producer, port, view)
traceitem.finalize()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
except AttributeError:
pass
# --------------------------------------------------------------------------
# Now, trace the transfer functions (color maps and opacity maps) used.
ctfs = set([x for x in proxies_of_interest if smtrace.Trace.get_registered_name(x, "lookup_tables")])
if not skipRenderingComponents and ctfs:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup color maps and opacity maps used in the visualization",
"# note: the Get..() functions create a new object, if needed",
"# ----------------------------------------------------------------"])
for ctf in ctfs:
smtrace.Trace.get_accessor(ctf)
if ctf.ScalarOpacityFunction in proxies_of_interest:
smtrace.Trace.get_accessor(ctf.ScalarOpacityFunction)
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# --------------------------------------------------------------------------
# Now, trace the animation scenes
# There should be only one but keep the code generic
anims = set([x for x in proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "animation")])
if not skipRenderingComponents and anims and not skipAnimationComponents:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup animation scene, tracks and keyframes",
"# note: the Get..() functions create a new object, if needed",
"# ----------------------------------------------------------------"])
for anim in anims:
traceitem = smtrace.TraceAnimationProxy(anim)
traceitem.finalize()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# Trace extractors.
exgens = set([x for x in proxies_of_interest \
if smtrace.Trace.get_registered_name(x, "extractors")])
if exgens:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# setup extractors",
"# ----------------------------------------------------------------"])
for exgen in exgens:
# FIXME: this currently doesn't handle multiple output ports
# correctly.
traceitem = smtrace.CreateExtractor( \
xmlname=exgen.Writer.GetXMLName(),
producer=exgen.Producer,
extractor=exgen,
registrationName=smtrace.Trace.get_registered_name(exgen, "extractors"))
traceitem.finalize()
del traceitem
trace.append_separated(smtrace.get_current_trace_output_and_reset(raw=True))
# restore the active source since the order in which the pipeline is created
# in the state file can end up changing the active source to be different
# than what it was when the state is being saved.
if not skipActiveComponents:
trace.append_separated([ \
"# ----------------------------------------------------------------",
"# restore active source",
"SetActiveSource(%s)" % smtrace.Trace.get_accessor(simple.GetActiveSource()),
"# ----------------------------------------------------------------"])
# Setup camera links
trace.append_separated(["# ----------------------------------------------------------------",
"# setup camera links"])
number_of_links = sm.ProxyManager().GetNumberOfLinks()
for i in range(number_of_links):
link_name = sm.ProxyManager().GetLinkName(i)
link_proxy = sm.ProxyManager().GetRegisteredLink(link_name)
if (not isinstance(link_proxy, sm.vtkSMCameraLink)
or link_proxy.GetNumberOfLinkedObjects() != 4):
continue
# Bidirectional link means that the linked proxies have the following pattern:
# index 0: Proxy_0
# index 1: Proxy_1
# index 2: Proxy_1
# index 3: Proxy_0
is_bidirectional_link = (link_proxy.GetLinkedProxy(0) == link_proxy.GetLinkedProxy(3) and
link_proxy.GetLinkedProxy(1) == link_proxy.GetLinkedProxy(2))
if not is_bidirectional_link:
continue
# Everything is valid so we can add this to the linked proxies
first_proxy = link_proxy.GetLinkedProxy(0)
second_proxy = link_proxy.GetLinkedProxy(1)
linkitem = smtrace.CallFunction("AddCameraLink",
first_proxy,
second_proxy,
link_name)
linkitem.finalize()
del linkitem
trace.append(smtrace.get_current_trace_output_and_reset(raw=True))
trace.append("# ----------------------------------------------------------------")
if postamble is None:
trace.append_separated(smtrace._get_standard_postamble_comment())
elif postamble:
trace.append_separated(postamble)
trace_result = str(trace) if not raw else trace.raw_data()
smtrace.Trace.output_target_type = previous_target_type
return trace_result
if __name__ == "__main__":
print("Running test")
simple.Mandelbrot()
simple.Show()
simple.Hide()
simple.Shrink().ShrinkFactor = 0.4
simple.UpdatePipeline()
simple.Clip().ClipType.Normal[1] = 1
rep = simple.Show()
view = simple.Render()
view.ViewSize = [500, 500]
rep.SetScalarBarVisibility(view, True)
simple.Render()
# rep.SetScalarBarVisibility(view, False)
print("====================================================================")
print(get_state())