Use regex to unfold arguments.

Author: MarcCote

scripts/smart_dispatch.py

@@ -36,9 +36,9 @@ def main():
             commands = smartdispatch.get_commands_from_file(args.commandsFile)
         else:
             # Commands that needs to be parsed and unfolded.
-            arguments = map(smartdispatch.unfold_argument, args.commandAndOptions)
+            arguments = smartdispatch.unfold_arguments(args.commandAndOptions)
             jobname = smartdispatch.generate_name_from_arguments(arguments, max_length=235)
-            commands = smartdispatch.get_commands_from_arguments(" ".join(args.commandAndOptions))
+            commands = smartdispatch.get_commands_from_arguments(arguments)
 
         commands = smartdispatch.replace_uid_tag(commands)
 

smartdispatch/argument.py

@@ -0,0 +1,32 @@
+import re
+
+
+class Argument(object):
+    def __init__(self):
+        self.name = ""
+        self.regex = ""
+
+    def unfold(self, match):
+        raise NotImplementedError("Subclass must implement this method!")
+
+
+class EnumerationArgument(object):
+    def __init__(self):
+        self.name = "enumeration"
+        self.regex = "\[[^]]*\]"
+
+    def unfold(self, match):
+        return match[1:-1].split(',')
+
+
+class RangeArgument(object):
+    def __init__(self):
+        self.name = "range"
+        self.regex = "\[\d+:\d+(:\d+)?\]"
+
+    def unfold(self, match):
+        groups = re.search(self.regex, match).groups()
+        start = int(groups[0])
+        end = int(groups[1])
+        step = 1 if groups[2] is None else int(groups[2])
+        return range(start, end, step)

smartdispatch/smartdispatch.py

@@ -1,10 +1,13 @@
 from __future__ import absolute_import
 
 import os
+import re
+import itertools
 from datetime import datetime
 
 import smartdispatch
 from smartdispatch import utils
+from smartdispatch.argument import EnumerationArgument, RangeArgument
 
 UID_TAG = "{UID}"
 
@@ -98,48 +101,29 @@ def get_commands_from_file(fileobj):
 
 
 def get_commands_from_arguments(arguments):
-    unfolded_commands = [arguments]
-    unfolded_commands_stard_idx = [0]
-
-    while True:
-        new_unfolded_commands = []
-        new_unfolded_commands_stard_idx = []
-
-        for idx in range(len(unfolded_commands)):
-            start_bracket_idx = unfolded_commands[idx].find("[", unfolded_commands_stard_idx[idx])
-
-            if start_bracket_idx == -1:
-                new_unfolded_commands_stard_idx = [-1]
-                break
-
-            while unfolded_commands[idx][start_bracket_idx + 1] == "[":
-                start_bracket_idx += 1
-
-            stop_bracket_idx = unfolded_commands[idx].find("]", start_bracket_idx)
-
-            for argument in unfolded_commands[idx][start_bracket_idx + 1:stop_bracket_idx].split(" "):
-                new_unfolded_commands_stard_idx += [start_bracket_idx + len(argument)]
-                new_unfolded_commands += [unfolded_commands[idx][0:start_bracket_idx] + argument + unfolded_commands[idx][stop_bracket_idx + 1:]]
-
-        if -1 in new_unfolded_commands_stard_idx:
-            break
-
-        unfolded_commands = new_unfolded_commands
-        unfolded_commands_stard_idx = new_unfolded_commands_stard_idx
-
-    return unfolded_commands
+    ''' Obtains commands from the product of every unfolded arguments.
+    Parameters
+    ----------
+    arguments : list of list of str
+        list of unfolded arguments
+    Returns
+    -------
+    commands : list of str
+        commands resulting from the product of every unfolded arguments
+    '''
+    return ["".join(argvalues) for argvalues in itertools.product(*arguments)]
 
 
-def unfold_argument(argument):
-    ''' Unfolds a folded argument into a list of unfolded arguments.
+def unfold_arguments(arguments):
+    ''' Unfolds folded arguments into a list of unfolded arguments.
 
-    An argument can be folded e.g. a list of unfolded arguments separated by spaces.
+    An argument can be folded e.g. a list of unfolded arguments separated by commas.
     An unfolded argument unfolds to itself.
 
     Parameters
     ----------
-    argument : str
-        argument to unfold
+    arguments : list of str
+        arguments to unfold
 
     Returns
     -------
@@ -148,11 +132,34 @@ def unfold_argument(argument):
 
     Complex arguments
     -----------------
-    *list (space)*: "item1 item2 ... itemN"
+    *enumeration*: "[item1,item2,...,itemN]"
+    *range*: "[start:end]" or "[start:end:step]"
     '''
+    text = utils.escape(" ".join(arguments))
+
+    # Order matter, if some regex is more greedy than another, the it should go after
+    arguments = [EnumerationArgument(), RangeArgument()]
+
+    # Build the master regex with all argument's regex
+    regex = "(" + "|".join(["(?P<{0}>{1})".format(arg.name, arg.regex) for arg in arguments]) + ")"
+
+    pos = 0
+    unfolded_arguments = []
+    for match in re.finditer(regex, text):
+        # Add already unfolded argument
+        unfolded_arguments.append([text[pos:match.start()]])
+
+        # Unfold argument
+        groupdict = match.groupdict()
+        for argument in arguments:
+            if groupdict[argument.name] is not None:
+                unfolded_arguments.append(argument.unfold(groupdict[argument.name]))
+
+        pos = match.end()
 
-    # Suppose `argument`is a space separated list
-    return argument.split(" ")
+    unfolded_arguments.append([text[pos:]])  # Add remaining unfolded arguments
+    unfolded_arguments = [map(utils.hex2str, argvalues) for argvalues in unfolded_arguments]
+    return unfolded_arguments
 
 
 def replace_uid_tag(commands):

smartdispatch/tests/test_smartdispatch.py

@@ -78,27 +78,28 @@ def test_get_commands_from_file():
 
 
 def test_get_commands_from_arguments():
-    # Test single unfolded arguments
-    args = "ls"
+    # Test with one argument
+    args = [["ls"]]
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["ls"])
 
-    args = "echo 1"
+    args = [["echo 1"]]
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["echo 1"])
 
-    args = "echo [1 2]"
+    # Test two arguments
+    args = smartdispatch.unfold_arguments(["echo [1,2]"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["echo 1", "echo 2"])
 
-    args = "echo test [1 2] yay"
+    args = smartdispatch.unfold_arguments(["echo test [1,2] yay"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["echo test 1 yay", "echo test 2 yay"])
 
-    args = "echo test[1 2]"
+    args = smartdispatch.unfold_arguments(["echo test[1,2]"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["echo test1", "echo test2"])
 
-    args = "echo test[1 2]yay"
+    args = smartdispatch.unfold_arguments(["echo test[1,2]yay"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["echo test1yay", "echo test2yay"])
 
-    # Test multiple unfolded arguments
-    args = "python my_command.py [0.01 0.000001 0.00000000001] -1 [omicron mu]"
+    # Test multiple folded arguments
+    args = smartdispatch.unfold_arguments(["python my_command.py", "[0.01,0.000001,0.00000000001]", "-1", "[omicron,mu]"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["python my_command.py 0.01 -1 omicron",
                                                                    "python my_command.py 0.01 -1 mu",
                                                                    "python my_command.py 0.000001 -1 omicron",
@@ -107,7 +108,7 @@ def test_get_commands_from_arguments():
                                                                    "python my_command.py 0.00000000001 -1 mu"])
 
     # Test multiple unfolded arguments and not unfoldable brackets
-    args = "python my_command.py [0.01 0.000001 0.00000000001] -1 [[42 133,666]] slow [omicron mu]"
+    args = smartdispatch.unfold_arguments(["python my_command.py [0.01,0.000001,0.00000000001] -1 \[[42,133\,666]\] slow [omicron,mu]"])
     assert_equal(smartdispatch.get_commands_from_arguments(args), ["python my_command.py 0.01 -1 [42] slow omicron",
                                                                    "python my_command.py 0.01 -1 [42] slow mu",
                                                                    "python my_command.py 0.01 -1 [133,666] slow omicron",
@@ -122,16 +123,6 @@ def test_get_commands_from_arguments():
                                                                    "python my_command.py 0.00000000001 -1 [133,666] slow mu"])
 
 
-def test_unfold_argument():
-    # Test simple argument
-    for arg in ["arg1", "[arg1"]:
-        assert_array_equal(smartdispatch.unfold_argument(arg), [arg])
-
-    # Test list (space)
-    for arg in ["arg1 arg2", "arg1 ", " arg1"]:
-        assert_array_equal(smartdispatch.unfold_argument(arg), arg.split(" "))
-
-
 def test_replace_uid_tag():
     command = "command without uid tag"
     assert_array_equal(smartdispatch.replace_uid_tag([command]), [command])

smartdispatch/utils.py

@@ -35,6 +35,24 @@ def slugify(value):
     return str(re.sub('[-\s]+', '_', value))
 
 
+def escape(text, escaping_character="\\"):
+    """ Escape the escaped character using its hex representation """
+    def hexify(match):
+        return "\\x{0}".format(match.group()[-1].encode("hex"))
+
+    return re.sub(r"\\.", hexify, text)
+
+
+def hex2str(text):
+    """ Convert hex representation to the character it represents """
+    if len(text) == 0:
+        return ''
+
+    def unhexify(match):
+        return match.group()[2:].decode("hex")
+
+    return re.sub(r"\\x..", unhexify, text)
+
 @contextmanager
 def open_with_lock(*args, **kwargs):
     """ Context manager for opening file with an exclusive lock. """