Single stream processing

cups.py

@@ -190,7 +190,6 @@ def getLatestId( tablename, dstname, run, seg ):
     print("[CUPS FATAL]: cupsid is not defined")
     exit(0) # operating without a cupsid is now a fatal error
 
-
 @subcommand()
 def info( args ):
     start = datetime.datetime.now(datetime.timezone.utc)        
@@ -207,6 +206,24 @@ def info( args ):
 
 
 
+def update_production_status( update_query, retries=10, delay=10.0 ):
+    print(update_query)
+    for itry in range(0,retries):
+        time.sleep( delay * (itry + 1 ) * random.random() )
+        try:
+            with pyodbc.connect("DSN=ProductionStatusWrite") as statusdb:
+                curs=statusdb.cursor()
+                curs.execute(update_query)
+                curs.commit()
+                print(f"Applied after {itry+1} attempts")
+                return
+        except:
+            print(f"Failed {itry+1} attempts...")
+
+    print("Update failed")
+
+
+
 @subcommand()
 def started(args):
     """
@@ -234,7 +251,6 @@ def started(args):
 
     return 'result', ntries, start, finish, ex, nm, sv
 
-
 @subcommand([
     argument(     "--nsegments",help="Number of segments produced",dest="nsegments",default=1),
 ])
@@ -255,13 +271,13 @@ def running(args):
     set status='running',running='{timestamp}',nsegments={nsegments}
     where id={id_}
     """
-
     curs, ntries, start, finish, ex, nm, sv = dbQuery( cnxn_string_map[ 'statw' ], update )
     if curs:
         curs.commit()
 
     return 'result', ntries, start, finish, ex, nm, sv
 
+
 #_______________________________________________________________________________________________________
 @subcommand([
     argument("-e","--exit",help="Exit code of the payload macro",dest="exit",default=-1),
@@ -286,13 +302,13 @@ def finished(args):
     state='finished'
     if ec>0:
         state='failed'
+    update = None
     if args.inc:
         update = f"""
         update {tablename}
         set status='{state}',ended='{timestamp}',nsegments={ns},exit_code={ec},nevents=nevents+{ne}
         where id={id_}
         """
-#        where dstname='{dstname}' and run={run} and segment={seg} and id={id_}
     else:
         update = f"""
         update {tablename}
@@ -359,7 +375,6 @@ def nevents(args):
         set nevents=nevents+{ne}
         where id={id_}
         """
-#       where dstname='{dstname}' and run={run} and segment={seg} and id={id_}
     else:
         update = f"""
         update {tablename}
@@ -372,7 +387,7 @@ def nevents(args):
         curs.commit()
 
     return 'result', ntries, start, finish, ex, nm, sv
-        
+
 
 @subcommand([
 ])

kaedama.py

@@ -55,23 +55,8 @@
 arg_parser.add_argument( '--print-query',dest='printquery',help="Print the query after parameter substitution and exit", action="store_true", default=False )
 
 
-# TODO: physics/run2pp/ana449_2024p008/”run range”/DST_TRKR_CLUSTER
-#_default_filesystem = {
-#        'outdir'  :           "/sphenix/lustre01/sphnxpro/production/$(runtype)/$(runname)/$(name)/$(build)_$(tag)/run_$(rungroup)"
-#    ,   'logdir'  : "file:///sphenix/data/data02/sphnxpro/production/$(runtype)/$(runname)/$(name)/$(build)_$(tag)/run_$(rungroup)"
-#    ,   'histdir' :       "/sphenix/data/data02/sphnxpro/production/$(runtype)/$(runname)/$(name)/$(build)_$(tag)/run_$(rungroup)"
-#    ,   'condor'  :                                 "/tmp/production/$(runtype)/$(runname)/$(name)/$(build)_$(tag)/run_$(rungroup)"
-#}
-
-
-# TODO: physics/run2pp/ana449_2024p008/”run range”/DST_TRKR_CLUSTER
-#       runtype runname build_tag  runrange DST
-#_default_filesystem = {
-#        'outdir'  :           "/sphenix/lustre01/sphnxpro/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/$(name)"
-#    ,   'logdir'  : "file:///sphenix/data/data02/sphnxpro/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/$(name)"
-#    ,   'histdir' :       "/sphenix/data/data02/sphnxpro/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/$(name)"
-#    ,   'condor'  :                                 "/tmp/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/$(name)"
-#}
+arg_parser.add_argument( '--streamname', help="Name of the data stream for single-stream jobs" ) #TODO:  May not need these arguments...
+arg_parser.add_argument( '--streamfile', help="Filename (not incl run number) for the data stream" )
 
 _default_filesystem = {
         'outdir'  :           "/sphenix/lustre01/sphnxpro/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/{leafdir}"
@@ -80,7 +65,6 @@
     ,   'condor'  :                                 "/tmp/production/$(runtype)/$(runname)/$(build)_$(tag)/run_$(rungroup)/{leafdir}"
 }
 
-
 def sanity_checks( params, inputq ):
     result = True
 
@@ -100,9 +84,9 @@ def sanity_checks( params, inputq ):
     #
 
     # Name should be of the form DST_NAME_runXauau
-    if re.match( "[A-Z][A-Z][A-Z]_([A-Z]+_)+[a-z0-9]+", params['name'] ) == None:
-        logging.error( f'params.name {params["name"]} does not respect the sPHENIX convention:  DST_NAME_run<N>species' )
-        result = False
+    #if re.match( "[A-Z][A-Z][A-Z]_([A-Z]+_)+[a-z0-9]+", params['name'] ) == None:
+    #    logging.warn( f'params.name {params["name"]} does not respect the sPHENIX convention:  DST_NAME_run<N>species' )
+    #    result = False
 
     # Build and dbtag should not contain a "_"
     if re.match("_",params['build']):
@@ -237,6 +221,8 @@ def main():
     elif len(args.segments)>=3:
         seg_condition = "and segment in ( %s )" % ','.join( args.segments )
 
+    streamname = args.streamname
+    streamfile = args.streamfile
 
     RUNFMT = slurp.RUNFMT
     SEGFMT = slurp.SEGFMT
@@ -277,6 +263,8 @@ def main():
 
     if params:
 
+        params['name']=params['name'].format( **locals() )
+
         if args.mangle_dstname:
             params['name']=params['name'].replace('DST',args.mangle_dstname)
             logging.info(f"DST name is mangled to {params['name']}")

production-rules/DST_STREAMING_EVENT_run2pp_ana435_2024p007.yaml

@@ -0,0 +1,202 @@
+PHYS_DST_SINGLE_STREAMING_EVENT_run2pp:
+
+   params:
+     name:       DST_STREAMING_EVENT_{streamname}_run2pp
+     build:      ana.435
+     build_name: ana435
+     dbtag:      2024p007
+     logbase :   $(name)_$(build)_$(tag)-$INT(run,{RUNFMT})-$INT(seg,{SEGFMT})
+     outbase :   $(name)_$(build)_$(tag)
+     script  :   run_cosmics.sh
+     payload :   ./slurp-examples/sPHENIX/cosmics/
+     mem     :   20480MB
+     neventsper: 1000
+     comment :    "---"
+     rsync   : "./slurp-examples/sPHENIX/cosmics/*,cups.py,bachi.py,odbc.ini"
+
+   input:
+      db: daqdb
+      direct_path: /sphenix/lustre01/sphnxpro/{mode}/*/physics/
+      query: |-
+         with partialrun as (
+         select 'daqdb/filelist'                                                                                    as source      , 
+                runnumber                                                                                                          , 
+                0                                                                                                   as segment     , 
+                string_agg( distinct split_part(filename,'/',-1), ' ' )                                             as files       ,   
+                string_agg( distinct split_part(filename,'/',-1) || ':' || firstevent || ':' || lastevent, ' ' )    as fileranges  
+
+         from filelist
+         where 
+           ( 
+             (filename  like '/bbox%/{streamfile}%-0000.evt'   and lastevent>2 ) or
+             (filename  like '/bbox%/GL1_physics%-0000.evt'    and lastevent>2 ) 
+           )
+           {run_condition}
+
+         group by runnumber
+         having
+                every(transferred_to_sdcc)   and
+                max(lastevent)>1000          and
+
+                sum( case when filename like '/bbox%/GL1_physics%'  then 1 else 0 end )>0 and
+                sum( case when filename like '/bbox%/{streamfile}%' then 1 else 0 end )>0
+
+         order by runnumber
+         ),
+
+         fullrun as (
+         select 
+                'daqdb/filelist'                                                                                    as source      , 
+                runnumber                                                                                                          , 
+                0                                                                                                   as segment     , 
+                string_agg( distinct split_part(filename,'/',-1), ' ' )                                             as files       ,   
+                string_agg( distinct split_part(filename,'/',-1) || ':' || firstevent || ':' || lastevent, ' ' )    as fileranges  
+         from  
+                filelist
+         where 
+           ( 
+             (filename  like '/bbox%/{streamfile}%.evt'   and lastevent>2 ) or
+             (filename  like '/bbox%/GL1_physics%.evt'   and lastevent>2 ) 
+
+           )
+           {run_condition}
+
+         group by runnumber
+         having
+                every(transferred_to_sdcc)   and
+                max(lastevent)>1000          and
+
+                sum( case when filename like '/bbox%/GL1_physics%'  then 1 else 0 end )>0 and
+                sum( case when filename like '/bbox%/{streamfile}%' then 1 else 0 end )>0 
+
+         order by runnumber
+         )
+
+         select *,'partial run' as runtype from partialrun where runnumber not in ( select runnumber from fullrun ) 
+              union all 
+         select *,'full run' as runtype from fullrun where true 
+
+              ;
+
+   # TODO:  Need to add error checking to make sure that outdir, logdir, etc... are quoted properly.  Else, this will cause problems with argument substitution
+   filesystem:  
+     outdir  : "/sphenix/lustre01/sphnxpro/physics/slurp/streaming/physics/$(build)_$(tag)/run_$(rungroup)"
+     logdir  : "file:///sphenix/data/data02/sphnxpro/streaminglogs/$(build)_$(tag)/run_$(rungroup)"
+     histdir :       "/sphenix/data/data02/sphnxpro/streamhist/$(build)_$(tag)/run_$(rungroup)"
+     condor :        "/tmp/testlogs/$(build)_$(tag)/run_$(rungroup)"
+
+   #
+   # Again I note the need to ensure that the arguments are properly specified given the
+   # definition of the payload script.
+   #
+   job:
+     executable            : "{payload}/run_cosmics.sh"
+     arguments             : "$(nevents) {outbase} {logbase} $(run) $(seg) {outdir} $(build) $(tag) $(inputs) $(ranges) {neventsper} {logdir} {comment} {histdir} {PWD} {rsync}"
+     output_destination    : '{logdir}'
+     log                   : '{condor}/{logbase}.condor'
+     accounting_group      : "group_sphenix.mdc2"
+     accounting_group_user : "sphnxpro"
+     priority : '4000'
+     request_xferslots: '0'
+
+
+
+#_____________________________________________________________________________________________________________________________
+
+PHYS_DST_SINGLE_TRKR_HIT_SET_physics_2024p007:
+   # DST_EVENT works from a pre-built set of run lists.
+   params:
+     name:       DST_TRKR_HIT_{streamname}_run2pp
+     build:      new
+     build_name: new
+     dbtag:      2024p007
+     logbase :   $(name)_$(build)_$(tag)-$INT(run,{RUNFMT})-$INT(seg,{SEGFMT})
+     outbase :   $(name)_$(build)_$(tag)
+     script  :   run.sh
+     payload :   ./slurp-examples/sPHENIX/TrackingProduction/
+     mem     :   2048MB
+     rsync   : "./slurp-examples/sPHENIX/TrackingProduction/*,cups.py,bachi.py,odbc.ini"
+
+   input:
+      db: fc
+      query: |-
+         select 
+                'filecatalog/datasets'   as source       ,
+                runnumber                                ,
+                segment                                  ,
+                filename                    as files     ,
+                'X'                         as fileranges
+         from  
+                datasets
+         where
+                filename like 'DST_STREAMING_EVENT_{streamname}_run2pp_ana435_2024p007%'
+                {run_condition}
+                and runnumber>=49700
+         order by runnumber
+                {limit_condition}
+              ;              
+   filesystem:  
+     outdir : "/sphenix/lustre01/sphnxpro/physics/slurp/tracking/$(build)_$(tag)/run_$(rungroup)"
+     logdir : "file:///sphenix/data/data02/sphnxpro/trackinglogs/$(build)_$(tag)/run_$(rungroup)"
+     histdir :       "/sphenix/data/data02/sphnxpro/hitsethist/$(build)_$(tag)/run_$(rungroup)"
+     condor :        "/tmp/trkrogs/$(build)_$(tag)/run_$(rungroup)"
+
+   job:
+     executable            : "{payload}/run.sh"
+     arguments             : "$(nevents) {outbase} {logbase} $(run) $(seg) {outdir} $(build) $(tag) $(inputs) $(ranges) {logdir} {histdir} {PWD} {rsync}"
+     output_destination    : '{logdir}'
+     log                   : '{condor}/{logbase}.condor'
+     accounting_group      : "group_sphenix.mdc2"
+     accounting_group_user : "sphnxpro"
+     priority : '3800'
+
+
+#_____________________________________________________________________________________________________________________________
+
+DST_TRKR_CLUSTER_SET_run2pp_2024p007:
+   # DST_EVENT works from a pre-built set of run lists.
+   params:
+     name:       DST_TRKR_CLUSTER_run2pp
+     build:      new
+     build_name: new
+     dbtag:      2024p007
+     logbase :   $(name)_$(build)_$(tag)-$INT(run,{RUNFMT})-$INT(seg,{SEGFMT})
+     outbase :   $(name)_$(build)_$(tag)
+     script  :   run_job0.sh
+     payload :   ./slurp-examples/sPHENIX/TrackingProduction/
+     mem     :   2048MB
+     nevents :  0
+     rsync   : "./slurp-examples/sPHENIX/TrackingProduction/*,cups.py,bachi.py,odbc.ini"
+
+   input:
+      db: fc
+      query: |-
+         select 
+                'filecatalog/datasets'   as source       ,
+                runnumber                                ,
+                segment                                  ,
+                filename                    as files     ,
+                'X' as fileranges
+         from  
+                datasets
+         where
+                filename like 'DST_TRKR_HIT_run2pp_new_2024p007%'
+                {run_condition}
+                and runnumber>=49700
+         order by runnumber
+                {limit_condition}
+              ;              
+   filesystem:  
+     outdir : "/sphenix/lustre01/sphnxpro/physics/slurp/tracking/$(build)_$(tag)/run_$(rungroup)"
+     logdir : "file:///sphenix/data/data02/sphnxpro/trackinglogs/$(build)_$(tag)/run_$(rungroup)"
+     histdir :       "/sphenix/data/data02/sphnxpro/clusterhist/$(build)_$(tag)/run_$(rungroup)"
+     condor :        "/tmp/trkrlogs/$(build)_$(tag)/run_$(rungroup)"
+
+   job:
+     executable            : "{payload}/run_job0.sh"
+     arguments             : "{nevents} {outbase} {logbase} $(run) $(seg) {outdir} $(build) $(tag) $(inputs) $(ranges) {logdir} {histdir} {PWD} {rsync}"
+     output_destination    : '{logdir}'
+     log                   : '{condor}/{logbase}.condor'
+     accounting_group      : "group_sphenix.mdc2"
+     accounting_group_user : "sphnxpro"
+     priority : '3800'

setup

@@ -6,6 +6,7 @@ export PATH=${PATH}:${HOME}/bin
 export ODBCINI=./odbc.ini
 
 if [[ $OS =~ "Alma" ]]; then
+
    export PATH=/usr/bin:${PATH}
    export PYTHONPATH=/opt/sphenix/core/lib/python3.9/site-packages
    alias python=/usr/bin/python
@@ -17,3 +18,4 @@ echo Using $(python --version)
 
 
 
+