feat: add transformer

a transformer is pretty dumb right now - it takes the matches
output, a selection algorithm, and passes the result back
to the solver (that nows how to parse the match details
for metadata) and then return a template. We will eventually
not require a solver and just use an LLM or similar

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

Author: vsoch

examples/fractale/README.md

@@ -120,6 +120,45 @@ fractale satisfy ./examples/fractale/jobspec-containment-unsatisfied.yaml
 ```
 We likely want to have a more structured query syntax that can handle AND, OR, and other specifics. The actual search should remain general to support any generic key/value pair of attributes. My database structure and queries are also bad. 
 
+## Script Request
+
+A script request is going to:
+
+```console
+=> 1. take input (right now jobspec) "what clusters can satisfy this request?
+=> 2. subsystem solver "these clusters can"
+=> 3. selection plugin (defaults to random) "how many and how do you want to choose from this set?"
+=> 4. I have chosen N, transform them appropriately to submit
+=> 5. render the matches based on the subsystem
+```
+
+Right now we have this done rather manually, and the idea is that an LLM can eventually more elegantly do it. 
+Here is an example.
+
+```bash
+$ fractale script ./examples/fractale/software-curl.yaml
+```
+```console
+=> 🍇 Loading cluster "a" subsystem "containment"
+=> 🍇 Loading cluster "a" subsystem "modules"
+=> 🍇 Loading cluster "a" subsystem "spack"
+    => Exploring cluster "a" containment subsystem
+          (1/1) satisfied resource core 
+Cluster "a" is a match
+{
+│   'version': 1,
+│   'resources': [{'type': 'slot', 'count': 1, 'with': [{'type': 'core', 'count': 1}], 'label': 'task'}],
+│   'tasks': [{'command': ['gmx'], 'slot': 'task', 'count': {'per_slot': 1}}],
+│   'attributes': {
+│   │   'system': {
+│   │   │   'duration': 0,
+│   │   │   'requires': {'software': [{'name': 'curl', 'type': 'binary'}]},
+│   │   │   'files': {'batch-script': {'mode': 33216, 'data': '#!/bin/bash\n\nspack load curl\ngmx', 'encoding': 'utf-8'}}
+│   │   }
+│   }
+}
+```
+
 ## Save
 
 We can save an image of our subystem for a cluster. E.g.,

fractale/cli/__init__.py

@@ -89,8 +89,22 @@ def get_parser():
         formatter_class=argparse.RawTextHelpFormatter,
         description="determine clusters that satisfy a jobspec based on user subsystems",
     )
-    satisfy.add_argument("jobspec", help="jobspec yaml or json file")
-    for cmd in [satisfy, save]:
+    # Generate a jobspec script
+    script = subparsers.add_parser(
+        "script",
+        formatter_class=argparse.RawTextHelpFormatter,
+        description="generate a batch script after satisfy",
+    )
+    script.add_argument(
+        "--selector", help="selection algorithm to use", default="random", choices=["random"]
+    )
+    script.add_argument(
+        "--transformer", help="transformer to use", default="flux", choices=["flux"]
+    )
+    for cmd in [satisfy, script]:
+        cmd.add_argument("jobspec", help="jobspec yaml or json file")
+
+    for cmd in [satisfy, save, script]:
         cmd.add_argument(
             "--solver",
             help="subsystem solved backend",
@@ -148,6 +162,8 @@ def help(return_code=0):
         from .generate_subsystem import main
     elif args.command == "satisfy":
         from .satisfy import main
+    elif args.command == "script":
+        from .script import main
     elif args.command == "save":
         from .save import main
     else:

fractale/selector/__init__.py

@@ -0,0 +1,9 @@
+import fractale.selector.algorithms as algorithms
+
+plugins = {"random": algorithms.random_selection}
+
+
+def get_selector(name):
+    if name not in plugins:
+        raise ValueError(f"{name} is not a valid selection algorithm.")
+    return plugins[name]

fractale/selector/algorithms.py

@@ -0,0 +1,11 @@
+import random
+
+# Selection algorithms for choosing clusters
+# Each selection algorithm should return a list
+
+
+def random_selection(choices):
+    """
+    Select randomly
+    """
+    return [random.choice(choices)]

fractale/subsystem/match.py

@@ -0,0 +1,75 @@
+import json
+from dataclasses import dataclass
+
+
+@dataclass
+class Match:
+    """
+    A single match is a datum that goes into a match set.
+    """
+
+    cluster: str
+    subsystem: str
+    requires: dict
+    details: dict
+
+
+class MatchSet:
+    """
+    A MatchSet includes one or more matches of clusters-> subsystems to a spec
+    """
+
+    def __init__(self):
+        # Lookup by cluster, subsystem, and then requirements and matches.
+        # This needs to be enough to return to the called and generate
+        # templates to submit jobs.
+        self.matches = {}
+
+    @property
+    def count(self):
+        """
+        Return the number of cluster matches.
+        """
+        return len(self.matches.keys())
+
+    @property
+    def clusters(self):
+        """
+        Return a list of clusters.
+        """
+        return list(self.matches)
+
+    def all(self):
+        """
+        Custom function to iterate over matches
+        """
+        return list(self.iterset())
+
+    def remove(self, cluster):
+        """
+        Remove a cluster from a match.
+        """
+        if cluster in self.matches:
+            del self.matches[cluster]
+
+    def iterset(self):
+        """
+        Custom function to iterate over matches
+        """
+        for cluster, by_subsystem in self.matches.items():
+            for subsystem, matches in by_subsystem.items():
+                for match in matches:
+                    yield match
+
+    def add(self, cluster, subsystem, requires, details):
+        """
+        Add a match, including cluster, subsystem, requirements, and details.
+        """
+        new_match = Match(cluster, subsystem, requires, details)
+        if cluster not in self.matches:
+            self.matches[cluster] = {}
+        if subsystem not in self.matches[cluster]:
+            self.matches[cluster][subsystem] = []
+        # this could be given directly, but I don't want to assume
+        # that a cluster and subsystem only has one possible match.
+        self.matches[cluster][subsystem].append(new_match)

fractale/subsystem/solver/base.py

@@ -1,8 +1,10 @@
+import json
 import os
 
 import fractale.jobspec as jspec
 import fractale.utils as utils
 from fractale.logger import LogColors, logger
+from fractale.subsystem.match import MatchSet
 
 
 class Solver:
@@ -32,14 +34,20 @@ def prepare_requirements(self, jobspec):
         requires["containment"] = jspec.flatten_jobspec_resources(js)
         return requires
 
-    def satisfied(self, jobspec):
+    def render(self, subsystems):
+        """
+        Take in a set of cluster matches and
+        """
+        return []
+
+    def satisfied(self, jobspec, return_results=False):
         """
         Determine if a jobspec is satisfied by user-space subsystems.
         """
         requires = self.prepare_requirements(jobspec)
 
         # These clusters will satisfy the request
-        matches = set()
+        matches = MatchSet()
 
         # We don't care about the association with tasks - the requires are matching clusters to entire jobs
         # We could optimize this to be fewer queries, but it's likely trivial for now
@@ -71,12 +79,15 @@ def satisfied(self, jobspec):
                 nodes = self.find_nodes(cluster, name, items)
                 if not nodes:
                     continue
-                matches.add((cluster, name))
+                # This is adding cluster, subsystem name, match criteria, and node ids
+                matches.add(cluster, name, items, nodes)
 
             if matches:
-                print(f"\n{LogColors.OKBLUE}({len(matches)}) Matches {LogColors.ENDC}")
-                for match in matches:
-                    print(f"cluster ({match[0]}) subsystem ({match[1]})")
+                print(f"\n{LogColors.OKBLUE}({matches.count}) Matches {LogColors.ENDC}")
+                for match in matches.iterset():
+                    print(f"cluster ({match.cluster}) subsystem ({match.subsystem})")
+                if return_results:
+                    return matches
                 return True
             else:
                 print(f"{LogColors.RED}=> No Matches{LogColors.ENDC}")
@@ -85,9 +96,14 @@ def satisfied(self, jobspec):
     def load(self, path):
         """
         Load a group of subsystem files, typically json JGF.
+
+        We also are careful to store metadata here that might be needed for
+        rendering.
         """
         from fractale.subsystem.subsystem import Subsystem
 
+        self.metadata = {}
+
         if not os.path.exists(path):
             raise ValueError(f"User subsystem directory {path} does not exist.")
         files = utils.recursive_find(path, "graph[.]json")
@@ -96,3 +112,4 @@ def load(self, path):
         for filename in files:
             new_subsystem = Subsystem(filename)
             self.load_subsystem(new_subsystem)
+            self.metadata[new_subsystem.name] = new_subsystem.metadata

fractale/subsystem/solver/database.py

@@ -117,6 +117,27 @@ def get_subsystem_nodes(self, cluster, subsystem):
         labels = self.query(statement)
         return [f"'{x[0]}'" for x in labels]
 
+    def render(self, subsystems):
+        """
+        Yield lines for the transformer.
+        """
+        for subsystem, items in subsystems.items():
+            for item in items:
+                # This is actually easier to do than a query!
+                if subsystem == "spack":
+                    for require in item.requires:
+                        item_type = require.get("type")
+                        item_name = require.get("name")
+                        if item_type == "binary" and item_name is not None:
+                            yield f"\nspack load {item_name}"
+
+                        elif subsystem == "environment-modules":
+                            item_type = require.get("type")
+                            # TODO we need to test if this will with with <.>
+                            item_name = require.get("attribute.name")
+                            if item_type == "module" and item_name is not None:
+                                yield f"\nmodule load {item_name}"
+
     def find_nodes(self, cluster, name, items):
         """
         Given a list of node labels, find children (attributes)