Merge pull request #11 from cedadev/cci_cases

Changes from CCI branch to main

extensions/templates/setup-cci.sh → config/setup-cci.sh

@@ -4,4 +4,5 @@ export WORKDIR=/gws/nopw/j04/esacci_portal/kerchunk_conversion/
 export SRCDIR=/home/users/dwest77/Documents/kerchunk_dev/kerchunk-builder
 export KVENV=/home/users/dwest77/Documents/kerchunk_dev/kerchunk-builder/build_venv
 
-source $KVENV/bin/activate
+source $KVENV/bin/activate
+module load jaspy

extensions/templates/phase.sbatch.template

@@ -1,14 +1,10 @@
 #!/bin/bash
-#SBATCH --partition=short-serial-4hr
-#SBATCH --account=short4hr
-#SBATCH --job-name={}
+#SBATCH --partition=short-serial
+#SBATCH --job-name={}_kerchunk
 
 #SBATCH --time={}
 #SBATCH --mem={}
 
-#SBATCH -o {}
-#SBATCH -e {}
-
 module add jaspy
 source {}/bin/activate
 

group_run.py

@@ -2,6 +2,7 @@
 import json
 import os
 import argparse
+import subprocess
 
 from pipeline.logs import init_logger, BypassSwitch
 
@@ -28,8 +29,15 @@ def get_attribute(env, args, var):
     else:
         print(f'Error: Missing attribute {var}')
         return None
+
 
 def main(args):
+
+    logger = init_logger(args.verbose, 0, 'main-group')
+
+    deploy(args, logger)
+
+def deploy(args, logger, get_id=False, dependent_id=False):
     """Assemble sbatch script for parallel running jobs"""
 
     logger = init_logger(args.verbose, 0, 'main-group')
@@ -98,8 +106,6 @@ def main(args):
         f'{group}_{phase}_array',             # Job name
         time,                                 # Time
         mem,                                  # Memory
-        f'{GROUPDIR}/outs/%A_{label}/%a.out', # Outs
-        f'{GROUPDIR}/errs/%A_{label}/%a.err', # Errs
         VENV,
         WORKDIR,
         GROUPDIR,
@@ -115,6 +121,8 @@ def main(args):
         sb += ' -Q'
     if args.backtrack:
         sb += ' -B'
+    if args.dryrun:
+        sb += ' -d'
 
     if 'X' in args.bypass:
         logger.warning('Running with XK Shape Bypass flag "X" is experimental and should only be used with approval.')
@@ -130,7 +138,17 @@ def main(args):
         logger.info('DRYRUN: sbatch command: ')
         print(f'sbatch --array=0-{group_len-1} {group_phase_sbatch}')
     else:
-        os.system(f'sbatch --array=0-{group_len-1} {group_phase_sbatch}')
+        if get_id: # Unused section to save the ID of the process
+            result = subprocess.run(['sbatch', f'--array=0-{group_len-1}', group_phase_sbatch], stdout=subprocess.PIPE)
+            try:
+                id = result.stdout.decode('utf-8').split(' ')[3].strip() # Check!
+                assert len(id) == 8
+                return id
+            except:
+                logger.error('Slurm submission failed')
+                return None
+        else:
+            os.system(f'sbatch --array=0-{group_len-1} {group_phase_sbatch}')
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(description='Run a pipeline step for a group of datasets')

pipeline/allocator.py

@@ -0,0 +1,23 @@
+__author__    = "Daniel Westwood"
+__contact__   = "[email protected]"
+__copyright__ = "Copyright 2023 United Kingdom Research and Innovation"
+
+# Job Subset Allocation Script
+# - Calculate Expected Utilisation for each Job (500 + Kerchunk Size*5 MB)
+
+# First-Fit Bin Packing Algorithm
+# - Sort Utilisations from largest to smallest
+# - Bin Capacity is largest size rounded up to next memory cap (1, 2, 3, 4 GB)
+# - Allocate item to first bin with space remaining
+# - End with N bins (subsets) - write list of project codes for each subset to a separate file in proj_code_subsets/set_N.txt
+# - Run array with number of subsets already set.
+
+# Utilisation estimate is (total_chunks * 835) + 500 (MB)
+"""
+for proj_code in (repeat_id set):
+    open detail-cfg (for this code)
+    calculate utilisation
+    add to dict [utilisation, proj_code]
+    keep track of max/min
+get bins using binpacking (pypi)
+"""