src/pydefx/pystudy.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright (C) 2019  EDF R&D
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20 import inspect
  21 import pathlib
  22 import tempfile
  23 import os
  24 import json
  25 from . import salome_proxy
  26 from . import samplecsvmanager
  27 from . import parameters
  28 from . import configuration
  29 from . import defaultschemabuilder
  30 from .studyexception import StudyUseException, StudyRunException
  31 from .studyresult import StudyResult
  32
  33 def defaultSampleManager():
  34   return samplecsvmanager.SampleManager()
  35
  36 class PyStudy:
  37   JOB_DUMP_NAME = "jobDump.xml"
  38   def __init__(self, sampleManager=None, schemaBuilder=None):
  39     self.job_id = -1
  40     self.global_result = StudyResult()
  41     if sampleManager is None:
  42       self.sampleManager = defaultSampleManager()
  43     else:
  44       self.sampleManager = sampleManager
  45     if schemaBuilder is None:
  46       self.schemaBuilder = defaultschemabuilder.DefaultSchemaBuilder()
  47     else:
  48       self.schemaBuilder = schemaBuilder
  49
  50   # Study creation functions
  51   def createNewJob(self, script, sample, params):
  52     """
  53     Create a new job out of those parameters:
  54     script : script / pyscript type
  55     sample : sample to be evaluated (Sample class)
  56     params : job submission parameters (Parameters class)
  57     The result directory will contain all the files needed for a launch and a
  58     job is created but not launched.
  59     """
  60     self._check(script,sample)
  61     self.sample = sample
  62     self.params = params
  63     self.params.salome_parameters.job_type = self.jobType()
  64     tmp_workdir = self.params.salome_parameters.result_directory
  65     schema_path, extra_files = self._prepareDirectoryForLaunch(tmp_workdir,
  66                                                                script)
  67     # this list manipulation is needed because in_files is not a python list
  68     # if we don't use a salome session. In that case swig uses a python tuple
  69     # in order to map a std::list as a parameter of a structure.
  70     in_files_as_list = list(self.params.salome_parameters.in_files)
  71     self.params.salome_parameters.in_files = in_files_as_list + extra_files
  72     self.params.salome_parameters.job_file = schema_path
  73     launcher = salome_proxy.getLauncher()
  74     self.job_id = launcher.createJob(self.params.salome_parameters)
  75     return self.job_id
  76
  77   def loadFromDirectory(self, path):
  78     """
  79     Recover a study from a result directory where a previous study was launched.
  80     """
  81     self.sample = self.sampleManager.restoreSample(path)
  82     job_string = loadJobString(path)
  83     launcher = salome_proxy.getLauncher()
  84     self.job_id = launcher.restoreJob(job_string)
  85     if job_id >= 0:
  86       salome_params = launcher.getJobParameters(job_id)
  87       self.params = parameters.Parameters(salome_parameters=salome_params)
  88       self.getResult()
  89     return self.job_id
  90
  91   def loadFromString(self, jobstring):
  92     """
  93     Recover a study from a string which contains the description of the job.
  94     This string can be obtained by launcher.dumpJob.
  95     """
  96     launcher = salome_proxy.getLauncher()
  97     self.job_id = launcher.restoreJob(jobstring)
  98     self.params = None
  99     self.sample = None
 100     if self.job_id >= 0:
 101       salome_params = launcher.getJobParameters(self.job_id)
 102       self.params = parameters.Parameters(salome_parameters=salome_params)
 103       #TODO: sampleManager should be loaded from result_directory
 104       self.sample = self.sampleManager.restoreSample(
 105                                                  salome_params.result_directory)
 106       self.getResult()
 107     else:
 108       raise StudyRunException("Failed to restore the job.")
 109
 110   def loadFromId(self, jobid):
 111     """
 112     Connect the study to an already created job.
 113     The result directory of the job must be already prepared for launch.
 114     """
 115     if jobid < 0:
 116       return
 117     self.job_id = jobid
 118     launcher = salome_proxy.getLauncher()
 119     salome_params = launcher.getJobParameters(job_id)
 120     self.params = parameters.Parameters(salome_parameters=salome_params)
 121     #TODO: sampleManager should be loaded from result_directory
 122     self.sample=self.sampleManager.restoreSample(salome_params.result_directory)
 123     self.script = None
 124     return
 125
 126   # launch parameters functions
 127   def jobType(self):
 128     return "yacs_file"
 129
 130   # TODO: may be deprecated
 131   def createDefaultParameters(self, resource="localhost",
 132                               nb_branches=None,
 133                               result_base_dir=None):
 134     """
 135     Create the Parameters structure and the result directory.
 136     The result directory created here is needed by the job.
 137     """
 138     newParams = parameters.Parameters(resource, nb_branches)
 139     newParams.salome_parameters.job_type = self.jobType()
 140     newParams.salome_parameters.job_name = "idefix_job"
 141     newParams.salome_parameters.result_directory = configuration.newResultDirectory(result_base_dir)
 142     return newParams
 143
 144   # Job management functions
 145   def launch(self):
 146     """
 147     The job should have been already created.
 148     """
 149     if self.job_id < 0 :
 150       raise StudyUseException("Nothing to launch! Job is not created!")
 151     tmp_workdir = self.params.salome_parameters.result_directory
 152     # run the job
 153     launcher = salome_proxy.getLauncher()
 154     launcher.launchJob(self.job_id)
 155     #save the job
 156     job_string = launcher.dumpJob(self.job_id)
 157     jobDumpPath = os.path.join(tmp_workdir, PyStudy.JOB_DUMP_NAME)
 158     with open(jobDumpPath, "w") as f:
 159       f.write(job_string)
 160
 161   def getResult(self):
 162     """
 163     Try to get the result file and if it was possible the results are loaded in
 164     the sample.
 165     An exception may be thrown if it was not possible to get the file.
 166     Return a StudyResult object.
 167     """
 168     self.global_result = StudyResult()
 169     if self.job_id < 0 :
 170       raise StudyUseException("Cannot get the results if the job is not created!")
 171     launcher = salome_proxy.getLauncher()
 172     state = launcher.getJobState(self.job_id)
 173     tmp_workdir = self.params.salome_parameters.result_directory
 174     searchResults = False
 175     errorIfNoResults = False
 176     errorMessage = ""
 177     if state == "CREATED" :
 178       raise StudyUseException("Cannot get the results if the job is not launched!")
 179     elif state ==  "QUEUED" or state == "IN_PROCESS":
 180       # no results available at this point. Try again later! Not an error.
 181       searchResults = False
 182     elif state == "FINISHED" :
 183       # verify the return code of the execution
 184       searchResults = True
 185       if(launcher.getJobWorkFile(self.job_id, "logs/exit_code.log", tmp_workdir)):
 186         exit_code_file = os.path.join(tmp_workdir, "exit_code.log")
 187         exit_code = ""
 188         if os.path.isfile(exit_code_file):
 189           with open(exit_code_file) as myfile:
 190             exit_code = myfile.read()
 191             exit_code = exit_code.strip()
 192         self.global_result.exit_code = exit_code
 193         if exit_code == "0" :
 194           errorIfNoResults = True # we expect to have full results
 195         else:
 196           errorMessage = "An error occured during the execution of the YACS schema."
 197       else:
 198         errorMessage = "Failed to get the exit code of the YACS schema execution."
 199
 200     elif state == "RUNNING" or state == "PAUSED" or state == "ERROR" :
 201       # partial results may be available
 202       searchResults = True
 203     elif state == "FAILED":
 204       # We may have some partial results because the job could have been
 205       # canceled or stoped by timeout.
 206       searchResults = True
 207       errorMessage = "Job execution failed!"
 208     if searchResults :
 209       if 1 == launcher.getJobWorkFile(self.job_id,
 210                                       self.sampleManager.getResultFileName(),
 211                                       tmp_workdir):
 212         try:
 213           res = self.sampleManager.loadResult(self.sample, tmp_workdir)
 214           self.global_result.result = res
 215         except Exception as err:
 216           if errorIfNoResults:
 217             raise err
 218       elif errorIfNoResults:
 219         errorMessage = "The job is finished but we cannot get the result file!"
 220     if len(errorMessage) > 0 :
 221       warningMessage = """
 222 The results you get may be incomplete or incorrect.
 223 For further details, see {}/logs directory on {}.""".format(
 224                           self.params.salome_parameters.work_directory,
 225                           self.params.salome_parameters.resource_required.name)
 226       errorMessage += warningMessage
 227       self.global_result.error_message = errorMessage
 228       raise StudyRunException(errorMessage)
 229     return self.global_result
 230
 231   def resultAvailable(self):
 232     """
 233     Try to get the result and return True in case of success with no exception.
 234     In case of success the results are loaded in the sample.
 235     """
 236     resultFound = False
 237     try:
 238       self.getResult()
 239       resultFound = True
 240     except:
 241       resultFound = False
 242     return resultFound
 243
 244   def getJobState(self):
 245     if self.job_id < 0:
 246       return "NOT_CREATED"
 247     launcher = salome_proxy.getLauncher()
 248     return launcher.getJobState(self.job_id)
 249
 250   def getProgress(self):
 251     if self.job_id < 0:
 252       return 0.0
 253     state = self.getJobState()
 254     if state == "CREATED" or state == "QUEUED" :
 255       return 0.0
 256     if not self.resultAvailable():
 257       return 0.0
 258     return self.sample.progressRate()
 259
 260   def dump(self):
 261     if self.job_id < 0 :
 262       raise StudyUseException("Cannot dump the job if it is not created!")
 263     launcher = salome_proxy.getLauncher()
 264     return launcher.dumpJob(self.job_id)
 265
 266   def wait(self, sleep_delay=10):
 267     """ Wait for the end of the job """
 268     launcher = salome_proxy.getLauncher()
 269     job_id = self.job_id
 270     jobState = launcher.getJobState(job_id)
 271     import time
 272     while jobState=="QUEUED" or jobState=="IN_PROCESS" or jobState=="RUNNING" :
 273       time.sleep(sleep_delay)
 274       jobState = launcher.getJobState(job_id)
 275
 276   def _prepareDirectoryForLaunch(self, result_directory, script):
 277     """
 278     result_directory : path to a result working directory.
 279     script : script / pyscript type
 280     return:
 281       yacs_schema_path: path to the yacs schema (xml file).
 282       extra_in_files: list of files to add to salome_parameters.in_files
 283     """
 284     if not os.path.exists(result_directory):
 285       os.makedirs(result_directory)
 286     # export sample to result_directory
 287     inputFiles = self.sampleManager.prepareRun(self.sample, result_directory)
 288
 289     # export nbbranches
 290     configpath = os.path.join(result_directory, "idefixconfig.json")
 291     dicconfig = {}
 292     dicconfig["nbbranches"]  = self.params.nb_branches
 293     dicconfig["studymodule"] = "idefixstudy"
 294     dicconfig["sampleIterator"] = self.sampleManager.getModuleName()
 295     with open(configpath, "w") as f:
 296       json.dump(dicconfig, f, indent=2)
 297     studypath = os.path.join(result_directory, "idefixstudy.py")
 298     with open(studypath, "w") as f:
 299       f.write(script.script)
 300     schema_path, extra_files = self.schemaBuilder.buildSchema(result_directory)
 301
 302     extra_files.extend([configpath, studypath])
 303     extra_files.extend(inputFiles)
 304     return schema_path, extra_files
 305
 306   def _check(self, script, sample):
 307     "Raise StudyUseException if the sample does not match with the sample."
 308     script_params = script.getInputNames()
 309     sample_inputs = sample.getInputNames()
 310     if len(script_params) < 1:
 311       raise StudyUseException("The study function should have at least one parameter. None found.")
 312     if len(script_params) != len(sample_inputs):
 313       m="The study function should have the same number of parameters as the input variables in the sample ({} != {})."
 314       raise StudyUseException(m.format(len(script_params), len(sample_inputs)))
 315     for nm in script_params:
 316       if nm not in sample_inputs:
 317         raise StudyUseException("Parameter {} not found in the sample.".format(nm))
 318
 319 ### Deprecated!!!!
 320 def dumpJob(result_directory, jobString):
 321   """
 322   Save the jobString to a file into result_directory.
 323   result_directory is a string representing a path to a directory.
 324   jobString is a string representing the serialization of a job.
 325   Use loadJobString for reloading the string saved here.
 326   """
 327   jobDumpPath = os.path.join(result_directory, PyStudy.JOB_DUMP_NAME)
 328   with open(jobDumpPath, "w") as f:
 329     f.write(job_string)
 330
 331 def loadJobString(result_directory):
 332   """
 333   Return the jobString saved by the dumpJob function into a directory.
 334   Use dumpJob for saving a the string.
 335   """
 336   jobDumpPath = os.path.join(result_directory, PyStudy.JOB_DUMP_NAME)
 337   with open(jobDumpPath, "r") as f:
 338     job_string = f.read()
 339   return job_string
 340