The analysis pipeline was modified to:

(1) Start up assuming that the statevector and CycleControl still need to be setup. This allows
stand-alone operation of this pipeline

(2) The analysis pipeline no longer needs the obsoperator object

(3) The pipeline crashes on errors, there is no try/except anymore for each part

da/tools/pipeline.py

@@ -114,7 +114,7 @@ def forward_pipeline(dacycle, platform, dasystem, samples, statevector, obsopera
     logging.info("Cycle finished...exiting pipeline")
 ####################################################################################################
 
-def analysis_pipeline(dacycle, platform, dasystem, samples, statevector, obsoperator):
+def analysis_pipeline(dacycle, platform, dasystem, samples, statevector):
     """ Main entry point for analysis of ctdas results """
 
     from da.analysis.expand_fluxes import save_weekly_avg_1x1_data, save_weekly_avg_state_data, save_weekly_avg_tc_data, save_weekly_avg_ext_tc_data, save_weekly_avg_agg_data
@@ -124,34 +124,35 @@ def analysis_pipeline(dacycle, platform, dasystem, samples, statevector, obsoper
 
     logging.info(header + "Starting analysis" + footer) 
 
-    try:
-        save_weekly_avg_1x1_data(dacycle, statevector)
-        save_weekly_avg_state_data(dacycle, statevector)
-    except:
-        logging.error("Error in analysis of weekly 1x1 fluxes and/or statevector, continuing analysis...")
-    try:
-        save_weekly_avg_tc_data(dacycle, statevector)
-        save_weekly_avg_ext_tc_data(dacycle)
-    except:
-        logging.error("Error in analysis of weekly tc fluxes, continuing analysis...")
-    try:
-        save_weekly_avg_agg_data(dacycle,region_aggregate='transcom')
-        save_weekly_avg_agg_data(dacycle,region_aggregate='olson')
-        save_weekly_avg_agg_data(dacycle,region_aggregate='country')
-    except:
-        logging.error("Error in analysis of weekly aggregated fluxes, continuing analysis...")
-    try:
-        time_avg(dacycle,'flux1x1')
-        time_avg(dacycle,'transcom')
-        time_avg(dacycle,'olson')
-        time_avg(dacycle,'country')
-    except:
-        logging.error("Error in analysis of time averaged fluxes, continuing analysis...")
-    try:
-        write_mole_fractions(dacycle)
-        summarize_obs(dacycle)
-    except:
-        logging.error("Error in analysis of mole fractions, continuing analysis...")
+    dasystem.validate()                               
+    dacycle.dasystem = dasystem                       
+    dacycle.daplatform = platform                    
+    dacycle.setup()                              
+    statevector.setup(dacycle)   
+
+    logging.info(header + "Starting mole fractions" + footer) 
+
+    write_mole_fractions(dacycle)
+    summarize_obs(dacycle['dir.analysis'])
+
+    logging.info(header + "Starting weekly averages" + footer) 
+
+    save_weekly_avg_1x1_data(dacycle, statevector)
+    save_weekly_avg_state_data(dacycle, statevector)
+    save_weekly_avg_tc_data(dacycle, statevector)
+    save_weekly_avg_ext_tc_data(dacycle)
+    save_weekly_avg_agg_data(dacycle,region_aggregate='transcom')
+    save_weekly_avg_agg_data(dacycle,region_aggregate='olson')
+    save_weekly_avg_agg_data(dacycle,region_aggregate='country')
+
+    logging.info(header + "Starting monthly and yearly averages" + footer) 
+
+    time_avg(dacycle,'flux1x1')
+    time_avg(dacycle,'transcom')
+    time_avg(dacycle,'olson')
+    time_avg(dacycle,'country')
+
+    logging.info(header + "Finished analysis" + footer) 
 
 def archive_pipeline(dacycle, platform, dasystem):
     """ Main entry point for archiving of output from one disk/system to another """