r"""
Internal functions not intended for use by users. This may change without
notice.
"""
from .. import logger
from ..modules.vtkPVInSitu import vtkInSituPipelinePython
from ..modules.vtkPVPythonCatalyst import vtkCPPythonScriptV2Helper
def _get_active_helper():
return vtkCPPythonScriptV2Helper.GetActiveInstance()
def _get_active_data_description():
if not IsInsitu():
return None
helper = _get_active_helper()
return helper.GetDataDescription()
def _transform_registration_name(name):
from . import get_args
import re, argparse
regex = re.compile(r"^\${args\.([\w.-]+)}$")
m = regex.match(name)
if not m:
return name
argname = m.group(1)
parser = argparse.ArgumentParser()
parser.add_argument("--%s" % argname, dest=argname)
parser.add_argument("-%s" % argname, dest=argname)
result = parser.parse_known_args(get_args())[0]
val = getattr(result, argname)
return val if val else name
[docs]def IsInsitu():
"""Returns True if executing in an insitu environment, else false"""
helper = _get_active_helper()
return (helper is not None)
[docs]def IsLegacyCatalystAdaptor():
"""Returns True if the active execution environment is from within a legacy
Catalyst adaptor implementation that uses vtkCPProcessor etc. manually,
instead of the Conduit-based in situ API"""
return _get_active_data_description() is not None
if IsInsitu() and not IsLegacyCatalystAdaptor():
from ..modules.vtkPVInSitu import vtkInSituInitializationHelper
[docs]def IsCatalystInSituAPI():
"""Returns True if the active execution environment is from within
an implementation of the Conduit-based Catalyst In Situ API."""
return IsInsitu() and not IsLegacyCatalystAdaptor() and vtkInSituInitializationHelper.IsInitialized()
[docs]def RegisterView(view):
"""Keeps track of views created inside a specific Catalyst
script. This is useful to ensure we only update the views for the
current script when multiple scripts are being executed in the same run.
"""
assert IsInsitu()
_get_active_helper().RegisterView(view.SMProxy)
if IsCatalystInSituAPI():
view.ViewTime = vtkInSituInitializationHelper.GetTime()
else:
view.ViewTime = _get_active_data_description().GetTime()
[docs]def CreateProducer(name):
assert IsInsituInput(name)
from . import log_level
from .. import log
originalname = name
name = _transform_registration_name(name)
log(log_level(), "creating producer for simulation input named '%s' (original-name=%s)" \
% (name, originalname))
if IsCatalystInSituAPI():
# Catalyst 2.0
from paraview import servermanager
producer = servermanager._getPyProxy(vtkInSituInitializationHelper.GetProducer(name))
# since state file may have arbitrary properties being specified
# on the original source, we ensure we ignore them
producer.IgnoreUnknownSetRequests = True
return producer
# Legacy Catalyst
from paraview import servermanager
helper = _get_active_helper()
producer = helper.GetTrivialProducer(name)
producer = servermanager._getPyProxy(producer)
# since state file may have arbitrary properties being specified
# on the original source, we ensure we ignore them
producer.IgnoreUnknownSetRequests = True
return producer
[docs]def InitializePythonEnvironment():
"""
This is called in vtkCPPythonPipeline to initialize the Python
environment to be more suitable for Catalyst in situ execution.
"""
from . import log_level
from .. import log
log(log_level(), "initializing Python environment for in situ")
# disable writing bytecode, we don't want to generate pyc files
# when running in situ.
log(log_level(), "disable byte-code generation")
import sys
sys.dont_write_bytecode = True
# ensure we can import ParaView servermanager and necessary modules
log(log_level(), "import paraview.servermanager")
from paraview import servermanager
# import Python wrapping for vtkPVCatalyst module
log(log_level(), "import paraview.modules.vtkPVCatalyst")
from paraview.modules import vtkPVCatalyst
[docs]def RegisterPackageFromZip(zipfilename, packagename=None):
from . import importers
zipfilename = _mpi_exchange_if_needed(zipfilename)
return importers.add_file(zipfilename, packagename)
[docs]def RegisterPackageFromDir(path):
import os.path
from . import importers
packagename = os.path.basename(path)
init_py = os.path.join(path, "__init__.py")
return importers.add_file(init_py, packagename)
[docs]def RegisterModuleFromFile(filename):
from . import importers
return importers.add_file(filename)
_temp_directory = None
def _mpi_exchange_if_needed(filename):
global _temp_directory
from ..modules.vtkRemotingCore import vtkProcessModule
pm = vtkProcessModule.GetProcessModule()
if pm.GetNumberOfLocalPartitions() <= 1:
return filename
from mpi4py import MPI
from vtkmodules.vtkParallelMPI4Py import vtkMPI4PyCommunicator
comm = vtkMPI4PyCommunicator.ConvertToPython(pm.GetGlobalController().GetCommunicator())
if comm.Get_rank() == 0:
with open(filename, 'rb') as f:
data = f.read()
else:
data = None
data = comm.bcast(data, root=0)
if comm.Get_rank() == 0:
return filename
else:
if not _temp_directory:
import tempfile, os.path
# we hook the temp-dir to the module so it lasts for the livetime of
# the interpreter and gets cleaned up on exit.
_temp_directory = tempfile.TemporaryDirectory()
filename = os.path.join(_temp_directory.name, os.path.basename(filename))
with open(filename, "wb") as f:
f.write(data)
return filename