r"""
Module that creates a source to match the data generated by
`paraview.demos.wavelet_miniapp`
This module supports the ability to be loaded in as a plugin in ParaView.
When loaded as a plugin, this adds a new data source named "Wavelet Mini-App"
(WaveletMiniApp) that can be used to produce data similar to when the
`wavelet_miniapp` is executed with properties matching the command line arguments
accepted by that module.
This is useful for generating tests or states that can be used with the miniapp.
"""
from paraview.demos.wavelet_miniapp import parser, create_dataset
# ----------------------------------------------------------------
# If this module is loaded as a Plugin in ParaView, we create a source
# that acts similar to the miniapp.This is useful for testing.
from paraview.util.vtkAlgorithm import *
[docs]
@smproxy.source(name="WaveletMiniApp",
label="Wavelet Mini-App")
class WaveletMiniApp(VTKPythonAlgorithmBase):
def __init__(self):
VTKPythonAlgorithmBase.__init__(self, nInputPorts=0, nOutputPorts=1, outputType='vtkImageData')
global parser
self._args = parser.parse_args(["-s", "not-used"])
def _get_update_time(self, outInfo):
executive = self.GetExecutive()
timesteps = self.GetTimestepValues()
if timesteps is None or len(timesteps) == 0:
return None
elif outInfo.Has(executive.UPDATE_TIME_STEP()) and len(timesteps) > 0:
import bisect
utime = outInfo.Get(executive.UPDATE_TIME_STEP())
i = bisect.bisect_right(timesteps, utime)
if i:
return timesteps[i - 1]
return timesteps[0]
else:
assert (len(timesteps) > 0)
return timesteps[0]
def _get_update_piece(self, outInfo):
executive = self.GetExecutive()
return (executive.GetUpdatePiece(outInfo), executive.GetUpdateNumberOfPieces(outInfo))
@smproperty.intvector(name="NumberOfTimeSteps", default_values=100)
@smdomain.intrange(min="1")
def SetTimeSteps(self, count):
"""Specify the number of timesteps (>=1)"""
if self._args.timesteps != count:
self._args.timesteps = count
self.Modified()
@smproperty.intvector(name="Size", default_values=101)
@smdomain.intrange(min="1")
def SetSize(self, size):
"""Specify the size (>=1) for the output datasets"""
if self._args.size != size:
self._args.size = size
self.Modified()
@smproperty.doublevector(name="TimestepValues", information_only="1", si_class="vtkSITimeStepsProperty")
def GetTimestepValues(self):
return [step / float(self._args.timesteps) for step in range(self._args.timesteps)]
def RequestInformation(self, request, inInfoVec, outInfoVec):
executive = self.GetExecutive()
outInfo = outInfoVec.GetInformationObject(0)
outInfo.Remove(executive.TIME_STEPS())
outInfo.Remove(executive.TIME_RANGE())
timesteps = self.GetTimestepValues()
if timesteps is not None:
for t in timesteps:
outInfo.Append(executive.TIME_STEPS(), t)
outInfo.Append(executive.TIME_RANGE(), timesteps[0])
outInfo.Append(executive.TIME_RANGE(), timesteps[-1])
ext = (self._args.size - 1) // 2
outInfo.Set(executive.WHOLE_EXTENT(), -ext, ext, -ext, ext, -ext, ext)
from vtkmodules.vtkCommonDataModel import vtkDataObject
outInfo.Set(vtkDataObject.SPACING(), 1, 1, 1)
return 1
def RequestData(self, request, inInfoVec, outInfoVec):
from vtkmodules.vtkCommonDataModel import vtkDataObject, vtkStreamingDemandDrivenPipeline
data_time = self._get_update_time(outInfoVec.GetInformationObject(0))
timestep = self.GetTimestepValues().index(data_time)
piece, num_pieces = self._get_update_piece(outInfoVec.GetInformationObject(0))
dataset, wholeExtent = create_dataset(timestep, self._args, piece, num_pieces)
output = vtkDataObject.GetData(outInfoVec, 0)
output.ShallowCopy(dataset)
output.GetInformation().Set(output.DATA_TIME_STEP(), data_time)
return 1