updates to moab to handle missed directives

Signed-off-by: vsoch <[email protected]>

Author: vsoch

fractale/transformer/__init__.py

@@ -4,10 +4,10 @@
 from .flux import Transformer as FluxTransformer
 from .kubernetes import Transformer as KubernetesTransformer
 from .lsf import Transformer as LSFTransformer
+from .moab import Transformer as MoabTransformer
 from .oar import Transformer as OARTransformer
 from .pbs import Transformer as PBSTransformer
 from .slurm import Transformer as SlurmTransformer
-from .moab import Transformer as MoabTransformer
 
 plugins = {
     "kubernetes": KubernetesTransformer,

fractale/transformer/cobalt/transform.py

@@ -99,7 +99,7 @@ def parse_cobalt_command(command_lines, spec):
         if line and not line.startswith("#"):
             main_command = line
             break
-            
+
     if not main_command:
         return []
 
@@ -146,12 +146,14 @@ def convert(self, spec):
         bt = epoch_to_cobalt_begin_time(spec.begin_time)
         if bt:
             qsub_cmd.extend(["--at", bt])
-        
+
         # Dependencies are specified with a colon-separated list of job IDs
         if spec.depends_on:
-            dep_str = ":".join(spec.depends_on) if isinstance(spec.depends_on, list) else spec.depends_on
+            dep_str = (
+                ":".join(spec.depends_on) if isinstance(spec.depends_on, list) else spec.depends_on
+            )
             qsub_cmd.extend(["--dependencies", dep_str])
-            
+
         # Node constraints are handled by --attrs
         if spec.constraints:
             qsub_cmd.extend(["--attrs", ":".join(spec.constraints)])
@@ -163,19 +165,18 @@ def convert(self, spec):
             qsub_cmd.extend(["-o", spec.output_file])
         if spec.error_file:
             qsub_cmd.extend(["-e", spec.error_file])
-            
+
         # Email notifications
         if spec.mail_user:
             qsub_cmd.extend(["-M", spec.mail_user])
             # Cobalt has a simple '--notify user' flag, equivalent to ALL in Slurm.
             if spec.mail_type:
                 qsub_cmd.append("--notify user")
 
-
         if spec.environment:
             for k, v in spec.environment.items():
                 qsub_cmd.extend(["--env", f"{k}={v}"])
-        
+
         # Note: Cobalt exclusive access is often handled by queue policy or `--mode script`.
         # We omit a direct flag to avoid conflicting with system-specific setups.
 
@@ -192,16 +193,16 @@ def convert(self, spec):
 
         if spec.container_image:
             aprun_cmd.extend(["singularity", "exec", spec.container_image])
-        
+
         # If spec.script is defined, it takes precedence over executable/arguments
         if spec.script:
-             exec_script_parts.extend(spec.script)
+            exec_script_parts.extend(spec.script)
         else:
-             if spec.executable:
-                 aprun_cmd.append(spec.executable)
-             if spec.arguments:
-                 aprun_cmd.extend(spec.arguments)
-             exec_script_parts.append(" ".join(aprun_cmd))
+            if spec.executable:
+                aprun_cmd.append(spec.executable)
+            if spec.arguments:
+                aprun_cmd.extend(spec.arguments)
+            exec_script_parts.append(" ".join(aprun_cmd))
 
         exec_script = "\n".join(exec_script_parts)
 
@@ -239,7 +240,7 @@ def _parse(self, content, return_unhandled=False):
             i = 0
             while i < len(args):
                 arg = args[i]
-                
+
                 # Logic to handle both `--key value` and `--key=value`
                 key, value = None, None
                 if "=" in arg:
@@ -251,10 +252,10 @@ def _parse(self, content, return_unhandled=False):
                     if i + 1 < len(args) and not args[i + 1].startswith("-"):
                         value = args[i + 1]
                         i += 2
-                    else: # It's a boolean flag
+                    else:  # It's a boolean flag
                         value = True
                         i += 1
-                
+
                 key = key.lstrip("-")
 
                 if key == "A":
@@ -270,8 +271,10 @@ def _parse(self, content, return_unhandled=False):
                 elif key == "O":
                     # This sets the job name AND the output file prefix
                     spec.job_name = value
-                    if not spec.output_file: spec.output_file = f"{value}.output"
-                    if not spec.error_file: spec.error_file = f"{value}.error"
+                    if not spec.output_file:
+                        spec.output_file = f"{value}.output"
+                    if not spec.error_file:
+                        spec.error_file = f"{value}.error"
                 elif key == "o":
                     spec.output_file = value
                 elif key == "e":
@@ -285,13 +288,13 @@ def _parse(self, content, return_unhandled=False):
                 elif key == "M":
                     spec.mail_user = value
                 elif key == "notify" and value == "user":
-                    spec.mail_type = ["ALL"] # Simple mapping
+                    spec.mail_type = ["ALL"]  # Simple mapping
                 elif key == "env":
-                     env_key, env_val = value.split("=", 1)
-                     spec.environment[env_key] = env_val
+                    env_key, env_val = value.split("=", 1)
+                    spec.environment[env_key] = env_val
                 else:
                     not_handled.add(key)
-        
+
         # We again assume a block of text here.
         if script_body:
             spec.script = script_body
@@ -315,7 +318,7 @@ def _parse(self, content, return_unhandled=False):
                     temp_args.pop(idx)
                     temp_args.pop(idx)
             except (ValueError, IndexError):
-                pass # Ignore if parsing aprun fails
+                pass  # Ignore if parsing aprun fails
 
             if temp_args:
                 spec.executable = temp_args[0]

fractale/transformer/lsf/transform.py

@@ -98,14 +98,14 @@ def parse_lsf_command(command_lines, spec):
     """
     if not command_lines:
         return []
-    
+
     main_command = ""
     for line in command_lines:
         line = line.strip()
         if line and not line.startswith("#"):
             main_command = line
             break
-    
+
     if not main_command:
         return []
 
@@ -141,7 +141,7 @@ def convert(self, spec):
         script.add("q", spec.queue)
         script.add("o", spec.output_file)
         script.add("e", spec.error_file)
-        
+
         # Mail notifications
         if spec.mail_user:
             script.add("u", spec.mail_user)
@@ -162,11 +162,11 @@ def convert(self, spec):
         r_parts = []
         if spec.constraints:
             r_parts.append(f'select[{":".join(spec.constraints)}]')
-        
+
         if spec.num_nodes > 1 and spec.num_tasks > 0:
             tasks_per_node = spec.num_tasks // spec.num_nodes
             if tasks_per_node > 0:
-                r_parts.append(f'span[ptile={tasks_per_node}]')
+                r_parts.append(f"span[ptile={tasks_per_node}]")
 
         rusage_parts = []
         if spec.mem_per_task:
@@ -179,7 +179,7 @@ def convert(self, spec):
         if spec.gpus_per_task > 0:
             # ngpus_excl_p = GPUs per process (task) in exclusive mode.
             rusage_parts.append(f"ngpus_excl_p={spec.gpus_per_task}")
-            
+
         if rusage_parts:
             r_parts.append(f'rusage[{":".join(rusage_parts)}]')
 
@@ -188,9 +188,9 @@ def convert(self, spec):
 
         if spec.exclusive_access:
             script.add_flag("x")
-        
+
         if spec.requeue:
-            script.add_flag("r") # -r makes the job rerunnable
+            script.add_flag("r")  # -r makes the job rerunnable
 
         # --- Priority and Scheduling ---
         lsf_prio = priority_to_lsf_priority(spec.priority)
@@ -199,10 +199,14 @@ def convert(self, spec):
 
         bt = epoch_to_lsf_begin_time(spec.begin_time)
         script.add("b", bt)
-        
+
         # Dependencies
         if spec.depends_on:
-            dep_str = " && ".join([f"ended({job_id})" for job_id in spec.depends_on]) if isinstance(spec.depends_on, list) else f"ended({spec.depends_on})"
+            dep_str = (
+                " && ".join([f"ended({job_id})" for job_id in spec.depends_on])
+                if isinstance(spec.depends_on, list)
+                else f"ended({spec.depends_on})"
+            )
             script.add("w", f'"{dep_str}"')
 
         script.newline()
@@ -212,7 +216,7 @@ def convert(self, spec):
             for key, value in spec.environment.items():
                 script.add_line(f"export {key}='{value}'")
             script.newline()
-        
+
         # If spec.script is defined, it takes precedence.
         if spec.script:
             script.add_lines(spec.script)
@@ -226,7 +230,7 @@ def convert(self, spec):
                 cmd_parts.append(f"--rs_per_host {spec.num_tasks // spec.num_nodes}")
                 cmd_parts.append(f"--tasks_per_rs 1")
                 cmd_parts.append(f"--cpu_per_rs {spec.cpus_per_task}")
-            
+
             if spec.container_image:
                 cmd_parts.extend(["singularity", "exec", spec.container_image])
             if spec.executable:
@@ -255,9 +259,9 @@ def _parse(self, content, return_unhandled=False):
             m = bsub_re.match(line)
             if m:
                 key, val = m.groups()
-                key = key.lstrip('-')
+                key = key.lstrip("-")
                 if val:
-                    val = val.split("#", 1)[0] # Remove trailing comments
+                    val = val.split("#", 1)[0]  # Remove trailing comments
 
                 val = val.strip().strip('"') if val else True
 
@@ -290,14 +294,14 @@ def _parse(self, content, return_unhandled=False):
                 elif key == "N":
                     spec.mail_type.append("END")
                 elif key == "w":
-                    ended_jobs = re.findall(r'ended\(([^)]+)\)', val)
+                    ended_jobs = re.findall(r"ended\(([^)]+)\)", val)
                     spec.depends_on = ended_jobs
                 elif key == "R":
                     # Parse complex -R string
-                    rusage_match = re.search(r'rusage\[(.*?)\]', val)
-                    span_match = re.search(r'span\[ptile=(\d+)\]', val)
-                    select_match = re.search(r'select\[(.*?)\]', val)
-                    
+                    rusage_match = re.search(r"rusage\[(.*?)\]", val)
+                    span_match = re.search(r"span\[ptile=(\d+)\]", val)
+                    select_match = re.search(r"select\[(.*?)\]", val)
+
                     if rusage_match:
                         for part in rusage_match.group(1).split(":"):
                             k, v = part.split("=", 1)
@@ -310,7 +314,7 @@ def _parse(self, content, return_unhandled=False):
                         if spec.num_tasks > 0 and tasks_per_node > 0:
                             spec.num_nodes = spec.num_tasks // tasks_per_node
                     if select_match:
-                        spec.constraints.extend(select_match.group(1).split(':'))
+                        spec.constraints.extend(select_match.group(1).split(":"))
                 else:
                     not_handled.add(key)
                 continue