Dim. Reduction and auto-gating on t-SNE plots

Author: arunviswanathan91

app/src/main/java/org/streamflow/bridge/RPaths.java

@@ -70,9 +70,34 @@ public static Path rscript() {
             Path exe = Paths.get(envHome, "bin", RSCRIPT);
             if (Files.exists(exe)) return exe;
         }
+        // A normal Windows R install ("C:\Program Files\R\R-4.4.2") puts Rscript in bin\ but does NOT
+        // add it to PATH, so without this the R plugins would be invisible on a machine that has R.
+        // Prefer the highest version found.
+        Path sys = newestSystemR();
+        if (sys != null) return sys;
         return Paths.get(RSCRIPT); // fall back to PATH lookup
     }
 
+    /** Newest {@code <ProgramFiles>\R\R-x.y.z\bin\Rscript.exe} on Windows, or null. */
+    private static Path newestSystemR() {
+        if (!IS_WINDOWS) return null;
+        Path best = null; String bestVer = null;
+        for (String base : new String[]{System.getenv("ProgramFiles"), System.getenv("ProgramW6432"), "C:\\R"}) {
+            if (base == null || base.isBlank()) continue;
+            Path rRoot = Paths.get(base, "R");
+            if (!Files.isDirectory(rRoot)) continue;
+            try (java.util.stream.Stream<Path> s = Files.list(rRoot)) {
+                for (Path dir : s.toList()) {
+                    Path exe = dir.resolve("bin").resolve(RSCRIPT);
+                    if (!Files.exists(exe)) continue;
+                    String ver = dir.getFileName().toString();   // e.g. "R-4.4.2"
+                    if (bestVer == null || ver.compareTo(bestVer) > 0) { bestVer = ver; best = exe; }
+                }
+            } catch (java.io.IOException ignored) { }
+        }
+        return best;
+    }
+
     /** Absolute path to streamflow_engine.R. */
     public static Path engineScript() {
         String override = System.getProperty("streamflow.engine");

app/src/main/java/org/streamflow/plugins/PluginManifest.java

@@ -0,0 +1,37 @@
+package org.streamflow.plugins;
+
+import java.nio.file.Path;
+import java.util.List;
+
+/**
+ * A parsed {@code plugin.json}. Plugins are R packages driven through {@code engine/run_plugin.R};
+ * this is purely the declarative half — what the plugin is called, what R packages it needs, which
+ * entry function to invoke, and what parameters the UI should collect.
+ *
+ * @param dir      the plugin directory (contains plugin.json + the entry script)
+ * @param id       stable identifier, e.g. {@code peacoqc}
+ * @param entryScript  script file inside {@code dir}, sourced by run_plugin.R
+ * @param entryFunction  function in that script, called as {@code fn(request)}
+ * @param rPackages  R packages that must be installed for this plugin to run
+ * @param params   parameters the UI collects and passes through in the request JSON
+ */
+public record PluginManifest(
+        Path dir,
+        String id,
+        String name,
+        String version,
+        String description,
+        String status,
+        String entryScript,
+        String entryFunction,
+        List<String> rPackages,
+        List<Param> params,
+        List<String> outputs) {
+
+    /** One declared parameter. {@code type} drives which control the Plugins UI renders. */
+    public record Param(String id, String label, String type, boolean required,
+                        String help, Double min, Double max, Object defaultValue) {}
+
+    /** Convenience: the entry script's absolute path. */
+    public Path entryScriptPath() { return dir.resolve(entryScript); }
+}

app/src/main/java/org/streamflow/plugins/PluginRegistry.java

@@ -0,0 +1,116 @@
+package org.streamflow.plugins;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.streamflow.bridge.RPaths;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+/**
+ * Discovers R plugins by scanning for {@code <root>/engine/plugins/<id>/plugin.json}.
+ *
+ * Plugins live in this repo and are bundled only into an R-enabled build (see release.yml's
+ * {@code bundle_r}); the default installer is Python-only, so {@link #rAvailable()} is what the UI
+ * uses to decide whether to offer them at all. Discovery itself never needs R, which keeps it
+ * unit-testable on a machine without an R install.
+ */
+public final class PluginRegistry {
+
+    private static final ObjectMapper JSON = new ObjectMapper();
+
+    private PluginRegistry() {}
+
+    /** Candidate roots: the packaged app dir first, then the dev repo checkout. */
+    static List<Path> pluginRoots() {
+        List<Path> roots = new ArrayList<>();
+        try { roots.add(RPaths.appDir().resolve("engine").resolve("plugins")); } catch (Exception ignored) {}
+        try { roots.add(RPaths.repoRoot().resolve("engine").resolve("plugins")); } catch (Exception ignored) {}
+        return roots;
+    }
+
+    /** True when an Rscript executable can be resolved (bundled R-Portable, R_HOME, a system R, or PATH). */
+    public static boolean rAvailable() {
+        try {
+            Path rs = RPaths.rscript();
+            // A bare "Rscript"/"Rscript.exe" means "hope it's on PATH" — only trust an absolute hit.
+            return rs != null && rs.isAbsolute() && Files.exists(rs);
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    /** All plugins found under the first existing root. Never throws; a bad manifest is skipped. */
+    public static List<PluginManifest> discover() {
+        for (Path root : pluginRoots()) {
+            if (root != null && Files.isDirectory(root)) return discoverIn(root);
+        }
+        return List.of();
+    }
+
+    /** Parse the {@code plugin.json} of every sub-directory. Malformed manifests are skipped, not fatal. */
+    public static List<PluginManifest> discoverIn(Path root) {
+        List<PluginManifest> out = new ArrayList<>();
+        try (Stream<Path> dirs = Files.list(root)) {
+            for (Path dir : dirs.filter(Files::isDirectory).toList()) {
+                Path manifest = dir.resolve("plugin.json");
+                if (!Files.isRegularFile(manifest)) continue;
+                try {
+                    out.add(parse(dir, manifest));
+                } catch (Exception ignored) {
+                    // A single broken plugin must not hide the healthy ones.
+                }
+            }
+        } catch (IOException ignored) { }
+        out.sort(java.util.Comparator.comparing(PluginManifest::name, String.CASE_INSENSITIVE_ORDER));
+        return out;
+    }
+
+    static PluginManifest parse(Path dir, Path manifestFile) throws IOException {
+        JsonNode n = JSON.readTree(Files.readString(manifestFile));
+        JsonNode entry = n.path("entry");
+        String script = entry.path("script").asText(null);
+        String function = entry.path("function").asText(null);
+        if (script == null || function == null) {
+            throw new IOException("plugin.json must declare entry.script and entry.function");
+        }
+        String id = n.path("id").asText(dir.getFileName().toString());
+
+        List<String> pkgs = new ArrayList<>();
+        n.path("requires").path("packages").forEach(p -> pkgs.add(p.asText()));
+
+        List<PluginManifest.Param> params = new ArrayList<>();
+        for (JsonNode p : n.path("params")) {
+            params.add(new PluginManifest.Param(
+                    p.path("id").asText(),
+                    p.path("label").asText(p.path("id").asText()),
+                    p.path("type").asText("string"),
+                    p.path("required").asBoolean(false),
+                    p.path("help").asText(null),
+                    p.has("min") ? p.path("min").asDouble() : null,
+                    p.has("max") ? p.path("max").asDouble() : null,
+                    p.has("default") ? toPlain(p.path("default")) : null));
+        }
+
+        List<String> outputs = new ArrayList<>();
+        n.path("outputs").forEach(o -> outputs.add(o.asText()));
+
+        return new PluginManifest(dir, id,
+                n.path("name").asText(id),
+                n.path("version").asText("0.0.0"),
+                n.path("description").asText(""),
+                n.path("status").asText("experimental"),
+                script, function, pkgs, params, outputs);
+    }
+
+    private static Object toPlain(JsonNode v) {
+        if (v.isBoolean()) return v.asBoolean();
+        if (v.isInt() || v.isLong()) return v.asLong();
+        if (v.isNumber()) return v.asDouble();
+        return v.asText();
+    }
+}

app/src/main/java/org/streamflow/plugins/PluginRunner.java

@@ -0,0 +1,115 @@
+package org.streamflow.plugins;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import javafx.concurrent.Task;
+import org.streamflow.bridge.RPaths;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Executes an R plugin through {@code engine/run_plugin.R}.
+ *
+ * The shim always writes a JSON response — including {@code {"ok": false, "error": ...}} — so this
+ * never has to parse an R stack trace. Anything R prints to stdout/stderr is captured as the plugin
+ * log and surfaced to the user when a run fails.
+ */
+public final class PluginRunner {
+
+    private static final ObjectMapper JSON = new ObjectMapper();
+
+    private PluginRunner() {}
+
+    /** Thrown when the plugin ran but reported {@code ok:false}, or the runner itself failed. */
+    public static class PluginException extends RuntimeException {
+        public PluginException(String message) { super(message); }
+    }
+
+    /** Locate {@code engine/run_plugin.R} next to the packaged app or in the dev checkout. */
+    static Path runnerScript() {
+        for (Path base : List.of(RPaths.appDir(), RPaths.repoRoot())) {
+            if (base == null) continue;
+            Path p = base.resolve("engine").resolve("run_plugin.R");
+            if (Files.isRegularFile(p)) return p;
+        }
+        throw new PluginException("engine/run_plugin.R not found (is the R runway installed?)");
+    }
+
+    /**
+     * Background task that runs {@code plugin} with {@code request} and yields the parsed response.
+     * Never runs on the FX thread — plugins can take minutes.
+     */
+    public static Task<JsonNode> run(PluginManifest plugin, Map<String, Object> request) {
+        return new Task<>() {
+            @Override protected JsonNode call() throws Exception {
+                Path rscript = RPaths.rscript();
+                if (rscript == null) throw new PluginException("No R installation found.");
+
+                Path work = Files.createTempDirectory("sf-plugin-");
+                Path reqFile = work.resolve("request.json");
+                Path respFile = work.resolve("response.json");
+                try {
+                    ObjectNode req = JSON.createObjectNode();
+                    request.forEach((k, v) -> req.set(k, JSON.valueToTree(v)));
+                    Files.writeString(reqFile, JSON.writeValueAsString(req));
+
+                    ProcessBuilder pb = new ProcessBuilder(
+                            rscript.toString(), runnerScript().toString(),
+                            plugin.dir().toString(), reqFile.toString(), respFile.toString());
+                    pb.redirectErrorStream(true);
+                    Process proc = pb.start();
+
+                    StringBuilder log = new StringBuilder();
+                    try (var r = proc.inputReader()) {
+                        String line;
+                        while ((line = r.readLine()) != null) {
+                            log.append(line).append('\n');
+                            updateMessage(line);            // surface progress in the UI
+                            if (isCancelled()) { proc.destroy(); throw new InterruptedException("cancelled"); }
+                        }
+                    }
+                    int exit = proc.waitFor();
+
+                    if (!Files.isRegularFile(respFile)) {
+                        throw new PluginException("Plugin produced no response (exit " + exit + ").\n"
+                                + tail(log.toString()));
+                    }
+                    JsonNode resp = JSON.readTree(Files.readString(respFile));
+                    if (!resp.path("ok").asBoolean(false)) {
+                        throw new PluginException(resp.path("error").asText("Plugin failed.")
+                                + "\n" + tail(log.toString()));
+                    }
+                    return resp;
+                } finally {
+                    deleteQuietly(respFile);
+                    deleteQuietly(reqFile);
+                    deleteQuietly(work);
+                }
+            }
+        };
+    }
+
+    /** 0-based row indices from a plugin response field (e.g. {@code kept_indices}). */
+    public static int[] indices(JsonNode resp, String field) {
+        JsonNode arr = resp.path(field);
+        if (!arr.isArray()) return new int[0];
+        int[] out = new int[arr.size()];
+        for (int i = 0; i < out.length; i++) out[i] = arr.get(i).asInt();
+        return out;
+    }
+
+    private static String tail(String s) {
+        String[] lines = s.strip().split("\n");
+        int from = Math.max(0, lines.length - 12);
+        return String.join("\n", java.util.Arrays.copyOfRange(lines, from, lines.length));
+    }
+
+    private static void deleteQuietly(Path p) {
+        try { Files.deleteIfExists(p); } catch (IOException ignored) { }
+    }
+}