new pipeline on the road

py/mains/age_to_density.py

@@ -77,7 +77,7 @@ def selectAges(snap,dt,f_hsml): #,times):
     
 # -----------------------------------------------------
 # compute, is the core function
-def process(simname,out,dt): #,times):
+def process(simname,out,dt,hsml): #,times):
     components="stars" 
     verbose=False
 
@@ -108,7 +108,7 @@ def process(simname,out,dt): #,times):
             ok,snap.mass = uns.getArrayF("stars","mass")
 
             # select ages according to dt
-            selectAges(snap,dt)
+            selectAges(snap,dt,hsml)
             
             # instantiate output object
             unso=CunsOut(out,"nemo");    # output file

py/modules/simulations/csnapshot.py

@@ -9,7 +9,7 @@ import numpy as np
 
 #
 class CSnapshot:
-    """Operations on snapshots"""
+    """Operations on UNS snapshots"""
     __uns=None
     __verbose=False
     __float32=True

py/modules/simulations/cuns_analysis.py

 #!/usr/bin/python
-
+from __future__ import print_function
 import numpy as np
-import time
 import sys
-sys.path.append('/home/jcl/works/GIT/uns_projects/py/modules/')
-sys.path.append('/home/jcl/works/GIT/uns_projects/py/modules/simulations')
+from multiprocessing import Process, Lock,Pool
+import multiprocessing
+import Queue # necessary to raise Queue.Empty signal
+import time
+import os
+import signal
+
 from py_unstools import *         # import py_unstools package
 from uns_simu import *
-from cfalcon import *
 from csnapshot import *
 
 
 class CUnsAnalysis:
     'Analysis pipeline'
-    #snapshot=None
-    #falcon=None
+    __sql3=None
+    simname=None
+    __r=None
+    __slist=None # snap list
+    __verbose=False
+    __analysis_script=None
+    __select="all"
     # constructor
-    def  __init__(self) :
-        self.snapshot=CSnapshot();
-        self.falcon=CFalcon()
+    def  __init__(self,simname,script,dbname=None,verbose=False,verbose_debug=False):
+        """
+        simname must be a UNS simulation belonging to a uns sqlite3 database.
+        infos regarding to simname simulation are loaded into privates variables
+        """
+        self.__vdebug=verbose_debug
+        self.__dbname=dbname
+        self.__analysis_script=script
+        self.__verbose=verbose
+        if self.__vdebug:
+            print ("simname = ", simname)
+        self.simname = simname
+        self.__sql3 = UnsSimu(simname,dbname=self.__dbname,verbose=self.__verbose)
+        self.__r = self.__sql3.getInfo() # return None if does not exist
+        
+        if self.__vdebug:
+            self.__sql3.printInfo(simname)
+        self.__slist = self.__sql3.getSnapshotList()
+
+
+    #
+    # ----
+    #        
+    # start analysis
+    def compute(self,ncores=None,select="all"):
+        """
+        Start parallel computation on nores
+        """
+        # selected components
+        self.__select=select
+
+        # compute cores
+        if ncores is None:
+            ncores=multiprocessing.cpu_count()
+        if self.__vdebug:
+            print ("#cores used :",ncores)
+        
+        # Feed QUEUE with list of snapshots
+        q = multiprocessing.Queue()
+        for snap in self.__slist: # loop on list os snsphots
+            #print "SNAP put :",snap
+            q.put(snap) # out a snapshot in the list
+
+        # start process jobs
+        processes=[]  # list of processes
+        n=0
+        
+        # manage signint signal
+        #original_sigint_handler = signal.signal(signal.SIGINT, signal.SIG_IGN) # IGNOROE before process creation
+        
+        # lock to control access to file
+        lock=Lock()
+
+        # loop on all #cores and start a process
+        for p in range(ncores):
+            p = Process(target=self.__analysisProcess, args=(q,n,lock,))  # create
+            p.start()  # start
+            processes.append(p) # append list of process, used for joining
+            n += 1
+
+        # wait all processes to complete
+        try:
+            for p in processes:
+                print ("waiting.. ",p)
+                p.join()
+        except KeyboardInterrupt: # allow to interupt all workers with  CTRL+C
+            for p in processes:
+                print ("Terminating.. ",p)
+                p.terminate()
+                p.join()
+        while not q.empty():
+            q.get(block=False)
+     
+    #
+    # ----
+    #        
+    # start analysis
+    def __analysisProcess(self,queue_list,n,lock):
+        """
+        Get a new file from list
+        Load UNS snapshot
+        Start analysis script
+        """
+
+        class data:
+            uns_snap=None
+            first=True
+
+        stop=False
+        cpt=0
+        first=True
+        while (not stop): 
+            try:
+                print ("Core [",n,"] waiting...")
+                my_snap=queue_list.get(True,0.05) # we must use this line
+                                                      # True means bloc during 0.05 sec,
+                                                      # if nothing, then get raise Queue.Empty exception
+                #my_snap=queue_list.get()  # do not use this, could block if nothing to get
+                #time.sleep(0.01)
+                data.uns_snap=CSnapshot(my_snap,self.__select)
+                ok=data.uns_snap.nextFrame("") # load in memory
+                if ok:
+                    # start analysis script
+                    execfile(self.__analysis_script)
+                    
+                cpt+=1
+                print ("Core [",n,"] got snap : ",my_snap,cpt)
+            except Queue.Empty:
+                stop = True # mo more data
+                if self.__verbose:
+                    print ("Queue.empty execption trapped...")
+
+        print ("Core [",n,"] DONE !",cpt)
+
+        #print "Core [",n,"] got snap : ",queue_list.get()
 
+        
+  

py/modules/uns_simu.py

@@ -26,7 +26,14 @@ class UnsSimu:
                 self.__status = 1
         else : # we must parse .unsio file
             status,self.dbname=self.parseDotUnsio("")
-            self.__conn = sqlite3.connect(self.dbname)
+            try:
+                self.__conn = sqlite3.connect(self.dbname)
+                c = self.__conn.cursor()
+                c.execute("select name from eps") # try a request to check if sdb3 is valid
+            except sqlite3.Error:
+                print ("File [",self.dbname,"] is not a valid sqlite3 DB")
+                sys.exit()
+                
 
         #print "CONN = ",self.__conn
         if (self.__conn != 0):
@@ -37,28 +44,29 @@ class UnsSimu:
         inputfile =file
         if (file == "" ):
             inputfile=os.environ['HOME']+"/"+self.__dbunsio
-            #print "Input file:",inputfile
+            #print ("Input file:",inputfile)
         
         gparam={}
         try:
             gp=open(inputfile,"r")
         except IOError:
-            #print "no file ["+inputfile+"], will use default :"+self.dbname
+            print ("no file ["+inputfile+"], will use default :"+self.dbname)
             return False,self.dbname
 
         for line in gp:
             data=line.split()
-            if (len(data)>1 and data[0]!='%' and data[0]!='#'):
-                gparam[data[0]] = data[1]
-            #print data
+            if (len(data)>1 and data[0][0]!='%' and data[0][0]!='#'):
+                gparam[data[0]] = data[2]
+                #print(">>", data[0], " -- ",gparam[data[0]])
 
-        if (gparam['dbname']!="") :
+        if ('dbname' in gparam and gparam['dbname']!="") :
             if (not os.path.isfile(gparam['dbname'])) :
+                print ("FILE [",gparam['dbname'],"] does not exist !!")
                 return False, self.dbname
             else :
                 return True,gparam['dbname']
         else:
-            False,self.dbname
+            return False,self.dbname
 
 
     def printInfo(self,name=None):

py/rsync_to_pythonpath.sh

 
 
 rsync  -av mains/*.py ${PYTHONPATHDYNAM}/bin/
-rsync  -Rav modules/*/*.{py,pyc} ${PYTHONPATHDYNAM}
+rsync  -Rav modules/*.{py,pyc} modules/*/*.{py,pyc} ${PYTHONPATHDYNAM}

py/test/test.py

@@ -2,10 +2,10 @@
 from __future__ import print_function
 
 import sys
-sys.path.append('/home/jcl/works/SVN/uns_projects/trunk/py/modules/')
-sys.path.append('/home/jcl/works/GIT/uns_projects/py/modules/simulations')
+sys.path.insert(0,'/home/jcl/works/GIT/uns_projects/py/modules/simulations')
+sys.path.insert(0,'/home/jcl/works/GIT/uns_projects/py/modules')
 from uns_simu import *
-from csnapshot import *
+from simulations.csnapshot import *
 import argparse
 
 class myargs: