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 salome
  25 import json
  26 from . import samplecsvmanager
  27 from . import parameters
  28 from . import configuration
  29 from . import defaultschemabuilder
  30
  31 def defaultSampleManager():
  32   return samplecsvmanager.SampleManager()
  33
  34 class PyStudy:
  35   JOB_DUMP_NAME = "jobDump.xml"
  36   def __init__(self, sampleManager=None, schemaBuilder=None):
  37     self.job_id = -1
  38     if sampleManager is None:
  39       self.sampleManager = defaultSampleManager()
  40     else:
  41       self.sampleManager = sampleManager
  42     if schemaBuilder is None:
  43       self.schemaBuilder = defaultschemabuilder.DefaultSchemaBuilder()
  44     else:
  45       self.schemaBuilder = schemaBuilder
  46
  47   # Study creation functions
  48   def createNewJob(self, script, sample, params):
  49     """
  50     Create a new job out of those parameters:
  51     script : script / pyscript type
  52     sample : sample to be evaluated (Sample class)
  53     params : job submission parameters (Parameters class)
  54     The result directory will contain all the files needed for a launch and a
  55     job is created but not launched.
  56     """
  57     self.sample = sample
  58     self.params = params
  59     self.params.salome_parameters.job_type = self.jobType()
  60     tmp_workdir = self.params.salome_parameters.result_directory
  61     schema_path, extra_files = self._prepareDirectoryForLaunch(tmp_workdir,
  62                                                                script)
  63
  64     self.params.salome_parameters.in_files.extend(extra_files)
  65     self.params.salome_parameters.job_file = schema_path
  66     launcher = salome.naming_service.Resolve('/SalomeLauncher')
  67     self.job_id = launcher.createJob(self.params.salome_parameters)
  68     return self.job_id
  69
  70   def loadFromDirectory(self, path):
  71     """
  72     Recover a study from a result directory where a previous study was launched.
  73     """
  74     self.sample = self.sampleManager.restoreSample(path)
  75     job_string = loadJobString(path)
  76     launcher = salome.naming_service.Resolve('/SalomeLauncher')
  77     self.job_id = launcher.restoreJob(job_string)
  78     if job_id >= 0:
  79       salome_params = launcher.getJobParameters(job_id)
  80       self.params = parameters.Parameters(salome_parameters=salome_params)
  81       self.getResult()
  82     return self.job_id
  83
  84   def loadFromString(self, jobstring):
  85     """
  86     Recover a study from a string which contains the description of the job.
  87     This string can be obtained by launcher.dumpJob.
  88     """
  89     launcher = salome.naming_service.Resolve('/SalomeLauncher')
  90     self.job_id = launcher.restoreJob(jobstring)
  91     self.params = None
  92     self.sample = None
  93     if self.job_id >= 0:
  94       salome_params = launcher.getJobParameters(self.job_id)
  95       self.params = parameters.Parameters(salome_parameters=salome_params)
  96       #TODO: sampleManager should be loaded from result_directory
  97       self.sample = self.sampleManager.restoreSample(
  98                                                  salome_params.result_directory)
  99       self.getResult()
 100     else:
 101       raise Exception("Failed to restore the job.")
 102
 103   def loadFromId(self, jobid):
 104     """
 105     Connect the study to an already created job.
 106     The result directory of the job must be already prepared for launch.
 107     """
 108     if jobid < 0:
 109       return
 110     self.job_id = jobid
 111     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 112     salome_params = launcher.getJobParameters(job_id)
 113     self.params = parameters.Parameters(salome_parameters=salome_params)
 114     #TODO: sampleManager should be loaded from result_directory
 115     self.sample=self.sampleManager.restoreSample(salome_params.result_directory)
 116     self.script = None
 117     return
 118
 119   # launch parameters functions
 120   def jobType(self):
 121     return "yacs_file"
 122
 123   # TODO: may be deprecated
 124   def createDefaultParameters(self, resource="localhost",
 125                               nb_branches=None,
 126                               result_base_dir=None):
 127     """
 128     Create the Parameters structure and the result directory.
 129     The result directory created here is needed by the job.
 130     """
 131     newParams = parameters.Parameters(resource, nb_branches)
 132     newParams.salome_parameters.job_type = self.jobType()
 133     newParams.salome_parameters.job_name = "idefix_job"
 134     newParams.salome_parameters.result_directory = configuration.newResultDirectory(result_base_dir)
 135     return newParams
 136
 137   # Job management functions
 138   def launch(self):
 139     """
 140     The job should have been already created.
 141     """
 142     if self.job_id < 0 :
 143       raise Exception("Nothing to launch! Job is not created!")
 144     tmp_workdir = self.params.salome_parameters.result_directory
 145     # run the job
 146     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 147     launcher.launchJob(self.job_id)
 148     #save the job
 149     job_string = launcher.dumpJob(self.job_id)
 150     jobDumpPath = os.path.join(tmp_workdir, PyStudy.JOB_DUMP_NAME)
 151     with open(jobDumpPath, "w") as f:
 152       f.write(job_string)
 153
 154   def getResult(self):
 155     """
 156     Try to get the result file and if it was possible the results are loaded in
 157     the sample.
 158     An exception may be thrown if it was not possible to get the file.
 159     """
 160     if self.job_id < 0 :
 161       raise Exception("Cannot get the results if the job is not created!")
 162     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 163     state = launcher.getJobState(self.job_id)
 164     tmp_workdir = self.params.salome_parameters.result_directory
 165     searchResults = False
 166     errorIfNoResults = False
 167     errorMessage = ""
 168     if state == "CREATED" :
 169       raise Exception("Cannot get the results if the job is not launched!")
 170     elif state ==  "QUEUED" or state == "IN_PROCESS":
 171       # no results available at this point. Try again later! Not an error.
 172       searchResults = False
 173     elif state == "FINISHED" :
 174       # verify the return code of the execution
 175       searchResults = True
 176       if(launcher.getJobWorkFile(self.job_id, "logs/exit_code.log", tmp_workdir)):
 177         exit_code_file = os.path.join(tmp_workdir, "exit_code.log")
 178         exit_code = ""
 179         if os.path.isfile(exit_code_file):
 180           with open(exit_code_file) as myfile:
 181             exit_code = myfile.read()
 182             exit_code = exit_code.strip()
 183         if exit_code == "0" :
 184           errorIfNoResults = True # we expect to have full results
 185         else:
 186           errorMessage = "An error occured during the execution of the YACS schema."
 187       else:
 188         errorMessage = "Failed to get the exit code of the YACS schema execution."
 189
 190     elif state == "RUNNING" or state == "PAUSED" or state == "ERROR" :
 191       # partial results may be available
 192       searchResults = True
 193     elif state == "FAILED":
 194       # We may have some partial results because the job could have been
 195       # canceled or stoped by timeout.
 196       searchResults = True
 197       errorMessage = "Job execution failed!"
 198     if searchResults :
 199       if 1 == launcher.getJobWorkFile(self.job_id,
 200                                       self.sampleManager.getResultFileName(),
 201                                       tmp_workdir):
 202         try:
 203           self.sampleManager.loadResult(self.sample, tmp_workdir)
 204         except Exception as err:
 205           if errorIfNoResults:
 206             raise err
 207       elif errorIfNoResults:
 208         errorMessage = "The job is finished but we cannot get the result file!"
 209     if len(errorMessage) > 0 :
 210       warningMessage = """
 211 The results you get may be incomplete or incorrect.
 212 For further details, see {}/logs directory on {}.""".format(
 213                           self.params.salome_parameters.work_directory,
 214                           self.params.salome_parameters.resource_required.name)
 215       errorMessage += warningMessage
 216       raise Exception(errorMessage)
 217     return self.sample
 218
 219   def resultAvailable(self):
 220     """
 221     Try to get the result and return True in case of success with no exception.
 222     In case of success the results are loaded in the sample.
 223     """
 224     resultFound = False
 225     try:
 226       self.getResult()
 227       resultFound = True
 228     except:
 229       resultFound = False
 230     return resultFound
 231
 232   def getJobState(self):
 233     if self.job_id < 0:
 234       return "NOT_CREATED"
 235     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 236     return launcher.getJobState(self.job_id)
 237
 238   def getProgress(self):
 239     if self.job_id < 0:
 240       return 0.0
 241     state = self.getJobState()
 242     if state == "CREATED" or state == "QUEUED" :
 243       return 0.0
 244     if not self.resultAvailable():
 245       return 0.0
 246     return self.sample.progressRate()
 247
 248   def dump(self):
 249     if self.job_id < 0 :
 250       raise Exception("Cannot dump the job if it is not created!")
 251     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 252     return launcher.dumpJob(self.job_id)
 253
 254   def wait(self, sleep_delay=10):
 255     """ Wait for the end of the job """
 256     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 257     job_id = self.job_id
 258     jobState = launcher.getJobState(job_id)
 259     import time
 260     while jobState=="QUEUED" or jobState=="IN_PROCESS" or jobState=="RUNNING" :
 261       time.sleep(sleep_delay)
 262       jobState = launcher.getJobState(job_id)
 263
 264   def _prepareDirectoryForLaunch(self, result_directory, script):
 265     """
 266     result_directory : path to a result working directory.
 267     script : script / pyscript type
 268     return:
 269       yacs_schema_path: path to the yacs schema (xml file).
 270       extra_in_files: list of files to add to salome_parameters.in_files
 271     """
 272     if not os.path.exists(result_directory):
 273       os.makedirs(result_directory)
 274     # export sample to result_directory
 275     inputFiles = self.sampleManager.prepareRun(self.sample, result_directory)
 276
 277     # export nbbranches
 278     configpath = os.path.join(result_directory, "idefixconfig.json")
 279     dicconfig = {}
 280     dicconfig["nbbranches"]  = self.params.nb_branches
 281     dicconfig["studymodule"] = "idefixstudy"
 282     dicconfig["sampleIterator"] = self.sampleManager.getModuleName()
 283     with open(configpath, "w") as f:
 284       json.dump(dicconfig, f, indent=2)
 285     studypath = os.path.join(result_directory, "idefixstudy.py")
 286     with open(studypath, "w") as f:
 287       f.write(script.script)
 288     schema_path, extra_files = self.schemaBuilder.buildSchema(result_directory)
 289
 290     extra_files.extend([configpath, studypath])
 291     extra_files.extend(inputFiles)
 292     return schema_path, extra_files
 293
 294 ### Deprecated!!!!
 295 def dumpJob(result_directory, jobString):
 296   """
 297   Save the jobString to a file into result_directory.
 298   result_directory is a string representing a path to a directory.
 299   jobString is a string representing the serialization of a job.
 300   Use loadJobString for reloading the string saved here.
 301   """
 302   jobDumpPath = os.path.join(result_directory, PyStudy.JOB_DUMP_NAME)
 303   with open(jobDumpPath, "w") as f:
 304     f.write(job_string)
 305
 306 def loadJobString(result_directory):
 307   """
 308   Return the jobString saved by the dumpJob function into a directory.
 309   Use dumpJob for saving a the string.
 310   """
 311   jobDumpPath = os.path.join(result_directory, PyStudy.JOB_DUMP_NAME)
 312   with open(jobDumpPath, "r") as f:
 313     job_string = f.read()
 314   return job_string
 315