various options added to help cartesius runs

da/analysis/expand_fluxes.py

@@ -7,7 +7,7 @@ from datetime import datetime, timedelta
 
 import logging
 import numpy as np
-from pylab import date2num, num2date
+from da.tools.general import date2num, num2date
 import da.tools.io4 as io
 from da.analysis.tools_regions import globarea, state_to_grid
 from da.tools.general import create_dirs

da/analysis/expand_molefractions.py

@@ -9,8 +9,8 @@ import netCDF4
 import numpy as np
 from string import join
 from datetime import datetime, timedelta
-from pylab import date2num, num2date
 sys.path.append('../../')
+from da.tools.general import date2num, num2date
 from da.tools.general import create_dirs
 import da.tools.io4 as io
 

da/analysis/tools_time.py

@@ -3,9 +3,8 @@ import sys
 import calendar
 import copy
 from datetime import datetime, timedelta
-from matplotlib.dates import date2num, num2date
+from da.tools.general import date2num, num2date
 from numpy import array, zeros, newaxis, logical_and, arange
-from pylab import floor, drange, num2date, date2num
     
 
 def Fromdatetime(date):
@@ -62,6 +61,7 @@ def dec2date(dectime):
 
 def dec2num(dectime):
     """ Function converts decimal time from year fraction (e.g., 1991.875 such as in IDL, CCG) to a python decimal numtime """
+    from pylab import floor, drange, num2date, date2num
     if not isinstance(dectime, list): dectime = [dectime]
 
     newdectime = []
@@ -79,6 +79,7 @@ def dec2num(dectime):
 
 def num2dec(numtime):
     """ Function converts python decimal numtime to an IDL decimal time """
+    from pylab import floor, drange, num2date, date2num
     res = date2dec(num2mydate(numtime))
     return res
 

da/platform/cartesius.py

@@ -42,7 +42,7 @@ class CartesiusPlatform(Platform):
             -on Jet/Zeus: return  
 
         """
-        return "queue"
+        return "foreground"
 
     def get_job_template(self, joboptions={}, block=False):
         """ 

da/tm5/observationoperator.py

@@ -474,61 +474,6 @@ class TM5ObservationOperator(ObservationOperator):
 
         return code
 
-    def tm5_under_mpirun(self):
-        """ Method handles the case where a shell runs an MPI process that forks into N TM5 model instances """
-
-        da_platform = self.dacycle.daplatform
-        targetdir = os.path.join(self.tm_settings[self.rundirkey])
-
-        if not os.path.exists(os.path.join(mpi_shell_location, mpi_shell_filename)):
-            logging.error("Cannot find the mpi_shell wrapper needed for completion (%s) in (%s)" % (mpi_shell_filename, mpi_shell_location))
-            logging.error("Please see the %s/readme_wrapper.txt file for instructions to compile it" % mpi_shell_location)
-            raise IOError
-
-        shutil.copy(os.path.join(mpi_shell_location, mpi_shell_filename) , os.path.join(targetdir, mpi_shell_filename))
-
-        # Go to executable directory and start the subprocess, using a new logfile
-
-        os.chdir(targetdir)
-        logging.debug('Changing directory to %s ' % targetdir)
-
-        # Remove the tm5.ok file from a previous run, placed back only if a successful TM5 run is executed
-
-        okfile = 'tm5.ok'
-        if os.path.exists(okfile): 
-            os.remove(okfile)
-
-        nthreads = self.dacycle['da.optimizer.nmembers']
-        jobparams = {'jobname':'tm5',
-                                   'jobnodes':'ncomp %d' % int(nthreads),
-                                   'jobtime':'01:00:00',
-                                   'joblog':os.path.join(self.dacycle['dir.jobs'])
-                                  }
-
-        # file ID and names
-        jobid = 'tm5'
-        targetdir = os.path.join(self.dacycle['dir.exec'])
-        jobfile = os.path.join(targetdir, 'jb.%s.jb' % jobid)
-        #logfile = jobfile.replace('.jb', '.log')
-
-        template = da_platform.get_job_template(jobparams, block=True)
-        template += 'cd %s\n' % targetdir
-        template += 'mpirun -np %d %s ./tm5.x\n' % (int(nthreads), mpi_shell_filename)
-
-        da_platform.write_job(jobfile, template, jobid)
-      
-        logging.info('Submitting job at %s' % datetime.datetime.now())
-        code = da_platform.submit_job(jobfile, joblog=jobfile) 
-        logging.info('Resuming job at %s' % datetime.datetime.now())
-        
-        if not os.path.exists(okfile): 
-            code = -1
-        else:
-            code = 0
-
-        return code
-
-
     def tm5_with_n_tracers(self):
         """ Method handles the case where one TM5 model instance with N tracers does the sampling of all ensemble members"""
 

da/tools/initexit.py

@@ -518,7 +518,7 @@ class CycleControl(dict):
             jobfile = os.path.join(targetdir, 'jb.%s.jb' % jobid)
             logfile = os.path.join(targetdir, 'jb.%s.log' % jobid)
             # Template and commands for job
-            jobparams = {'jobname':"j.%s" % jobid, 'jobtime':'06:00:00', 'logfile': logfile, 'errfile': logfile}
+            jobparams = {'jobname':"j.%s" % jobid, 'jobnodes':self['da.resources.ntasks'], 'jobtime': self['da.resources.ntime'], 'logfile': logfile, 'errfile': logfile}
             template = self.daplatform.get_job_template(jobparams)
             execcommand = os.path.join(self['dir.da_submit'], sys.argv[0]) 
             if '-t' in self.opts:

template.rc

@@ -8,6 +8,11 @@ time.nlag           : 5
 dir.da_run          : templaterundir
 random.seed.init    : ${datadir}/randomseedinit.pickle
 
+! info on the resources needed
+
+da.resources.ntasks : 1
+da.resources.ntime  : 00:02:00
+
 
 ! Info on the DA system used