Merge pull request #370 from pkjr002/bugfix/total

🐛 multi-threaded seems to cause idle/sleep

Author: kemccusker

modules/facts/total/total_workflow.py

@@ -7,6 +7,8 @@
 import yaml
 import xarray as xr
 import dask.array as da
+import dask.diagnostics
+import warnings
 
 
 def TotalSamplesInDirectory(directory, pyear_start, pyear_end, pyear_step, chunksize):
@@ -108,7 +110,7 @@ def TotalSamples(infiles, outfile, targyears, chunksize):
 		target_infiles, 
 	    combine="nested", 
 	    concat_dim="file", 
-	    chunks={"locations":chunksize},
+	    chunks=None,
 	)
 	ds = ds.sel(years=targyears)
 	# Sums everything across the new "file" dimension.
@@ -135,7 +137,30 @@ def TotalSamples(infiles, outfile, targyears, chunksize):
     # This actually carries out the delayed calculations and operations.
     # SBM: FYI Double check the numbers to ensure everything is summing across dims correctly.
     # SBM: FYI Also, check to see if output as something huge like float64.
-	total_out.to_netcdf(outfile, encoding={"sea_level_change": {"dtype": "f4", "zlib": True, "complevel":4, "_FillValue": nc_missing_value}})
+	#total_out.to_netcdf(outfile, encoding={"sea_level_change": {"dtype": "f4", "zlib": True, "complevel":4, "_FillValue": nc_missing_value}})
+
+	# New .to_netcdf run mechanic that allows dask the ability to control the chunking and reduces runtime, also places a progress bar in the task.out section.
+	dask.config.set({"array.slicing.split_large_chunks": True})
+	warnings.filterwarnings("ignore", category=FutureWarning)
+
+	write_job = total_out.to_netcdf(
+		outfile,
+		encoding={
+			"sea_level_change": {
+				"dtype": "f4",
+				"zlib": True,
+				"complevel": 4,
+				"_FillValue": nc_missing_value,
+			}
+		},
+		compute=False,
+	)
+
+	with dask.config.set(scheduler="single-threaded"):
+		with dask.diagnostics.ProgressBar():
+			print("            >> Writing to File...")
+			write_job.compute()
+	
 
 	return(outfile)
 
@@ -167,4 +192,4 @@ def TotalSamples(infiles, outfile, targyears, chunksize):
 		# Total up the workflow in the provided directory
 		TotalSamplesInDirectory(args.directory, args.pyear_start, args.pyear_end, args.pyear_step, args.chunksize)
 
-	exit()
+	exit()