defaultschemabuilder.py
studyexception.py
studyresult.py
+ salome_proxy.py
)
INSTALL(FILES ${SCRIPTS} DESTINATION ${SALOME_INSTALL_PYTHON}/pydefx)
from .sample import Sample
from .defaultschemabuilder import DefaultSchemaBuilder
-import salome
-salome.salome_init()
+from .salome_proxy import forceSalomeServers, forceNoSalomeServers
#
# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
#
-import salome
+from . import salome_proxy
import tempfile
import pathlib
def defaultWorkingDir(resource):
- resManager= salome.lcc.getResourcesManager()
+ resManager = salome_proxy.getResourcesManager()
resource_definition = resManager.GetResourceDefinition(resource)
return resource_definition.working_directory
def defaultNbBranches(resource):
- resManager= salome.lcc.getResourcesManager()
+ resManager = salome_proxy.getResourcesManager()
resource_definition = resManager.GetResourceDefinition(resource)
ret = resource_definition.nb_node
if ret < 1:
return result
def availableResources():
- resManager = salome.lcc.getResourcesManager()
- params = salome.ResourceParameters()
+ resManager = salome_proxy.getResourcesManager()
+ params = salome_proxy.createSalomeParameters().resource_required
params.can_launch_batch_jobs = True
return resManager.GetFittingResources(params)
#
# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
#
-import salome
-from . import configuration
import tempfile
+from . import salome_proxy
+from . import configuration
class Parameters:
def __init__(self, resource="localhost",
nb_branches=None, salome_parameters=None):
if salome_parameters is None :
- job_params = salome.JobParameters()
+ job_params = salome_proxy.createSalomeParameters()
job_params.job_type = "yacs_file"
- job_params.resource_required = salome.ResourceParameters()
job_params.resource_required.name = resource
job_params.job_name = "idefix_job"
job_params.wckey = configuration.defaultWckey(resource)
import pathlib
import tempfile
import os
-import salome
import json
+from . import salome_proxy
from . import samplecsvmanager
from . import parameters
from . import configuration
tmp_workdir = self.params.salome_parameters.result_directory
schema_path, extra_files = self._prepareDirectoryForLaunch(tmp_workdir,
script)
-
- self.params.salome_parameters.in_files.extend(extra_files)
+ # this list manipulation is needed because in_files is not a python list
+ # if we don't use a salome session. In that case swig uses a python tuple
+ # in order to map a std::list as a parameter of a structure.
+ in_files_as_list = list(self.params.salome_parameters.in_files)
+ self.params.salome_parameters.in_files = in_files_as_list + extra_files
self.params.salome_parameters.job_file = schema_path
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
self.job_id = launcher.createJob(self.params.salome_parameters)
return self.job_id
"""
self.sample = self.sampleManager.restoreSample(path)
job_string = loadJobString(path)
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
self.job_id = launcher.restoreJob(job_string)
if job_id >= 0:
salome_params = launcher.getJobParameters(job_id)
Recover a study from a string which contains the description of the job.
This string can be obtained by launcher.dumpJob.
"""
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
self.job_id = launcher.restoreJob(jobstring)
self.params = None
self.sample = None
if jobid < 0:
return
self.job_id = jobid
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
salome_params = launcher.getJobParameters(job_id)
self.params = parameters.Parameters(salome_parameters=salome_params)
#TODO: sampleManager should be loaded from result_directory
raise StudyUseException("Nothing to launch! Job is not created!")
tmp_workdir = self.params.salome_parameters.result_directory
# run the job
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
launcher.launchJob(self.job_id)
#save the job
job_string = launcher.dumpJob(self.job_id)
self.global_result = StudyResult()
if self.job_id < 0 :
raise StudyUseException("Cannot get the results if the job is not created!")
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
state = launcher.getJobState(self.job_id)
tmp_workdir = self.params.salome_parameters.result_directory
searchResults = False
def getJobState(self):
if self.job_id < 0:
return "NOT_CREATED"
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
return launcher.getJobState(self.job_id)
def getProgress(self):
def dump(self):
if self.job_id < 0 :
raise StudyUseException("Cannot dump the job if it is not created!")
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
return launcher.dumpJob(self.job_id)
def wait(self, sleep_delay=10):
""" Wait for the end of the job """
- launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ launcher = salome_proxy.getLauncher()
job_id = self.job_id
jobState = launcher.getJobState(job_id)
import time
--- /dev/null
+import salome
+import pylauncher
+import os
+
+_use_salome_servers = None
+
+def _default():
+ global _use_salome_servers
+ if _use_salome_servers is None:
+ try:
+ salome.salome_init()
+ _use_salome_servers = True
+ except RuntimeError:
+ _use_salome_servers = False
+
+def forceSalomeServers():
+ global _use_salome_servers
+ if not _use_salome_servers:
+ salome.salome_init()
+ _use_salome_servers = True
+
+def forceNoSalomeServers():
+ global _use_salome_servers
+ _use_salome_servers = False
+
+def createSalomeParameters():
+ _default()
+ if _use_salome_servers:
+ result = salome.JobParameters()
+ result.resource_required = salome.ResourceParameters()
+ else:
+ result = pylauncher.JobParameters_cpp()
+ result.resource_required = pylauncher.resourceParams()
+ return result
+
+_resourceManager = None
+def getResourcesManager():
+ global _resourceManager
+ _default()
+ if _resourceManager is None:
+ if _use_salome_servers:
+ _resourceManager = salome.lcc.getResourcesManager()
+ else:
+ catalog_path = os.environ.get("USER_CATALOG_RESOURCES_FILE", "")
+ _resourceManager = pylauncher.ResourcesManager_cpp(catalog_path)
+ return _resourceManager
+
+_launcher = None
+def getLauncher():
+ global _launcher
+ _default()
+ if _launcher is None:
+ if _use_salome_servers:
+ _launcher = salome.naming_service.Resolve('/SalomeLauncher')
+ else:
+ _launcher = pylauncher.Launcher_cpp()
+ _launcher.SetResourcesManager(getResourcesManager())
+ return _launcher
SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES
LABELS "${COMPONENT_NAME}"
)
+
+SET(TEST_NAME ${COMPONENT_NAME}_NoSession_PyExampleTest)
+ADD_TEST(${TEST_NAME} ./runUnitTest.sh)
+SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES
+ LABELS "${COMPONENT_NAME}"
+ )
import unittest
import insitu.insitumanager
import os
+import time
+
class TestYdefx(unittest.TestCase):
def test_insitu(self):
myParams = pydefx.Parameters()
myParams.configureResource("localhost")
mywd = os.path.join(myParams.salome_parameters.work_directory,
- "prescript_test")
+ "insitu_test" +
+ time.strftime("%Y-%m-%d-%H-%M-%S", time.localtime()))
myParams.salome_parameters.work_directory = mywd
myParams.createResultDirectory("/tmp")
import unittest
import os
+import time
class TestYdefx(unittest.TestCase):
def test_prescript(self):
myParams = pydefx.Parameters()
myParams.configureResource("localhost")
mywd = os.path.join(myParams.salome_parameters.work_directory,
- "prescript_test")
+ "prescript_test" +
+ time.strftime("%Y-%m-%d-%H-%M-%S", time.localtime()))
myParams.salome_parameters.work_directory = mywd
myParams.createResultDirectory("/tmp")
