add conda interpreter

python/src/main/java/org/apache/zeppelin/python/PythonCondaInterpreter.java

@@ -0,0 +1,401 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements.  See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License.  You may obtain a copy of the License at
+*
+*  http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.zeppelin.python;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.zeppelin.interpreter.*;
+import org.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.InterpreterResult.Type;
+import org.apache.zeppelin.scheduler.Scheduler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Conda support
+ */
+public class PythonCondaInterpreter extends Interpreter {
+  Logger logger = LoggerFactory.getLogger(PythonCondaInterpreter.class);
+  public static final String ZEPPELIN_PYTHON = "zeppelin.python";
+  public static final String CONDA_PYTHON_PATH = "/bin/python";
+  public static final String DEFAULT_ZEPPELIN_PYTHON = "python";
+
+  public static final Pattern PATTERN_OUTPUT_ENV_LIST = Pattern.compile("([^\\s]*)[\\s*]*\\s(.*)");
+  public static final Pattern PATTERN_COMMAND_ENV_LIST = Pattern.compile("env\\s*list\\s?");
+  public static final Pattern PATTERN_COMMAND_ENV = Pattern.compile("env\\s*(.*)");
+  public static final Pattern PATTERN_COMMAND_LIST = Pattern.compile("list");
+  public static final Pattern PATTERN_COMMAND_CREATE = Pattern.compile("create\\s*(.*)");
+  public static final Pattern PATTERN_COMMAND_ACTIVATE = Pattern.compile("activate\\s*(.*)");
+  public static final Pattern PATTERN_COMMAND_DEACTIVATE = Pattern.compile("deactivate");
+  public static final Pattern PATTERN_COMMAND_INSTALL = Pattern.compile("install\\s*(.*)");
+  public static final Pattern PATTERN_COMMAND_UNINSTALL = Pattern.compile("uninstall\\s*(.*)");
+  public static final Pattern PATTERN_COMMAND_HELP = Pattern.compile("help");
+  public static final Pattern PATTERN_COMMAND_INFO = Pattern.compile("info");
+
+  public PythonCondaInterpreter(Properties property) {
+    super(property);
+  }
+
+  @Override
+  public void open() {
+
+  }
+
+  @Override
+  public void close() {
+
+  }
+
+  @Override
+  public InterpreterResult interpret(String st, InterpreterContext context) {
+    InterpreterOutput out = context.out;
+    Matcher activateMatcher = PATTERN_COMMAND_ACTIVATE.matcher(st);
+    Matcher createMatcher = PATTERN_COMMAND_CREATE.matcher(st);
+    Matcher installMatcher = PATTERN_COMMAND_INSTALL.matcher(st);
+    Matcher uninstallMatcher = PATTERN_COMMAND_UNINSTALL.matcher(st);
+    Matcher envMatcher = PATTERN_COMMAND_ENV.matcher(st);
+
+    try {
+      if (PATTERN_COMMAND_ENV_LIST.matcher(st).matches()) {
+        String result = runCondaEnvList();
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (envMatcher.matches()) {
+        // `envMatcher` should be used after `listEnvMatcher`
+        String result = runCondaEnv(getRestArgsFromMatcher(envMatcher));
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (PATTERN_COMMAND_LIST.matcher(st).matches()) {
+        String result = runCondaList();
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (createMatcher.matches()) {
+        String result = runCondaCreate(getRestArgsFromMatcher(createMatcher));
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (activateMatcher.matches()) {
+        String envName = activateMatcher.group(1).trim();
+        return runCondaActivate(envName);
+      } else if (PATTERN_COMMAND_DEACTIVATE.matcher(st).matches()) {
+        return runCondaDeactivate();
+      } else if (installMatcher.matches()) {
+        String result = runCondaInstall(getRestArgsFromMatcher(installMatcher));
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (uninstallMatcher.matches()) {
+        String result = runCondaUninstall(getRestArgsFromMatcher(uninstallMatcher));
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else if (st == null || PATTERN_COMMAND_HELP.matcher(st).matches()) {
+        runCondaHelp(out);
+        return new InterpreterResult(Code.SUCCESS);
+      } else if (PATTERN_COMMAND_INFO.matcher(st).matches()) {
+        String result = runCondaInfo();
+        return new InterpreterResult(Code.SUCCESS, Type.HTML, result);
+      } else {
+        return new InterpreterResult(Code.ERROR, "Not supported command: " + st);
+      }
+    } catch (RuntimeException | IOException | InterruptedException e) {
+      throw new InterpreterException(e);
+    }
+  }
+
+  private void changePythonEnvironment(String envName)
+      throws IOException, InterruptedException {
+    PythonInterpreter python = getPythonInterpreter();
+    String binPath = null;
+    if (envName == null) {
+      binPath = getProperty(ZEPPELIN_PYTHON);
+      if (binPath == null) {
+        binPath = DEFAULT_ZEPPELIN_PYTHON;
+      }
+    } else {
+      Map<String, String> envList = getCondaEnvs();
+      for (String name : envList.keySet()) {
+        if (envName.equals(name)) {
+          binPath = envList.get(name) + CONDA_PYTHON_PATH;
+          break;
+        }
+      }
+    }
+    python.setPythonCommand(binPath);
+  }
+
+  private void restartPythonProcess() {
+    PythonInterpreter python = getPythonInterpreter();
+    logger.info("-----------> " + python);
+    python.close();
+    python.open();
+  }
+
+  protected PythonInterpreter getPythonInterpreter() {
+    LazyOpenInterpreter lazy = null;
+    PythonInterpreter python = null;
+    Interpreter p =
+        getInterpreterInTheSameSessionByClassName(PythonInterpreter.class.getName());
+
+    while (p instanceof WrappedInterpreter) {
+      if (p instanceof LazyOpenInterpreter) {
+        lazy = (LazyOpenInterpreter) p;
+      }
+      p = ((WrappedInterpreter) p).getInnerInterpreter();
+    }
+    python = (PythonInterpreter) p;
+
+    if (lazy != null) {
+      lazy.open();
+    }
+    return python;
+  }
+
+  public static String runCondaCommandForTextOutput(String title, List<String> commands)
+      throws IOException, InterruptedException {
+
+    String result = runCommand(commands);
+    return wrapCondaBasicOutputStyle(title, result);
+  }
+
+  private String runCondaCommandForTableOutput(String title, List<String> commands)
+      throws IOException, InterruptedException {
+
+    StringBuilder sb = new StringBuilder();
+    String result = runCommand(commands);
+
+    // use table output for pretty output
+    Map<String, String> envPerName = parseCondaCommonStdout(result);
+    return wrapCondaTableOutputStyle(title, envPerName);
+  }
+
+  protected Map<String, String> getCondaEnvs()
+      throws IOException, InterruptedException {
+    String result = runCommand("conda", "env", "list");
+    Map<String, String> envList = parseCondaCommonStdout(result);
+    return envList;
+  }
+
+  private String runCondaEnvList() throws IOException, InterruptedException {
+    return wrapCondaTableOutputStyle("Environment List", getCondaEnvs());
+  }
+
+  private String runCondaEnv(List<String> restArgs)
+      throws IOException, InterruptedException {
+
+    restArgs.add(0, "conda");
+    restArgs.add(1, "env");
+    restArgs.add(3, "--yes"); // --yes should be inserted after command
+
+    return runCondaCommandForTextOutput(null, restArgs);
+  }
+
+  private InterpreterResult runCondaActivate(String envName)
+      throws IOException, InterruptedException {
+
+    if (null == envName || envName.isEmpty()) {
+      return new InterpreterResult(Code.ERROR, "Env name should be specified");
+    }
+
+    changePythonEnvironment(envName);
+    restartPythonProcess();
+
+    return new InterpreterResult(Code.SUCCESS, "'" + envName + "' is activated");
+  }
+
+  private InterpreterResult runCondaDeactivate()
+      throws IOException, InterruptedException {
+
+    changePythonEnvironment(null);
+    restartPythonProcess();
+    return new InterpreterResult(Code.SUCCESS, "Deactivated");
+  }
+
+  private String runCondaList() throws IOException, InterruptedException {
+    List<String> commands = new ArrayList<String>();
+    commands.add("conda");
+    commands.add("list");
+
+    return runCondaCommandForTableOutput("Installed Package List", commands);
+  }
+
+  private void runCondaHelp(InterpreterOutput out) {
+    try {
+      out.setType(InterpreterResult.Type.HTML);
+      out.writeResource("output_templates/conda_usage.html");
+    } catch (IOException e) {
+      logger.error("Can't print usage", e);
+    }
+  }
+
+  private String runCondaInfo() throws IOException, InterruptedException {
+    List<String> commands = new ArrayList<String>();
+    commands.add("conda");
+    commands.add("info");
+
+    return runCondaCommandForTextOutput("Conda Information", commands);
+  }
+
+  private String runCondaCreate(List<String> restArgs)
+      throws IOException, InterruptedException {
+    restArgs.add(0, "conda");
+    restArgs.add(1, "create");
+    restArgs.add(2, "--yes");
+
+    return runCondaCommandForTextOutput("Environment Creation", restArgs);
+  }
+
+  private String runCondaInstall(List<String> restArgs)
+      throws IOException, InterruptedException {
+
+    restArgs.add(0, "conda");
+    restArgs.add(1, "install");
+    restArgs.add(2, "--yes");
+
+    return runCondaCommandForTextOutput("Package Installation", restArgs);
+  }
+
+  private String runCondaUninstall(List<String> restArgs)
+      throws IOException, InterruptedException {
+
+    restArgs.add(0, "conda");
+    restArgs.add(1, "uninstall");
+    restArgs.add(2, "--yes");
+
+    return runCondaCommandForTextOutput("Package Uninstallation", restArgs);
+  }
+
+  public static String wrapCondaBasicOutputStyle(String title, String content) {
+    StringBuilder sb = new StringBuilder();
+    if (null != title && !title.isEmpty()) {
+      sb.append("<h4>").append(title).append("</h4>\n")
+          .append("</div><br />\n");
+    }
+    sb.append("<div style=\"white-space:pre-wrap;\">\n")
+        .append(content)
+        .append("</div>");
+
+    return sb.toString();
+  }
+
+  public static String wrapCondaTableOutputStyle(String title, Map<String, String> kv) {
+    StringBuilder sb = new StringBuilder();
+
+    if (null != title && !title.isEmpty()) {
+      sb.append("<h4>").append(title).append("</h4>\n");
+    }
+
+    sb.append("<div style=\"display:table;white-space:pre-wrap;\">\n");
+    for (String name : kv.keySet()) {
+      String path = kv.get(name);
+
+      sb.append(String.format("<div style=\"display:table-row\">" +
+              "<div style=\"display:table-cell;width:150px\">%s</div>" +
+              "<div style=\"display:table-cell;\">%s</div>" +
+              "</div>\n",
+          name, path));
+    }
+    sb.append("</div>\n");
+
+    return sb.toString();
+  }
+
+  public static Map<String, String> parseCondaCommonStdout(String out)
+      throws IOException, InterruptedException {
+
+    Map<String, String> kv = new LinkedHashMap<String, String>();
+    String[] lines = out.split("\n");
+    for (String s : lines) {
+      if (s == null || s.isEmpty() || s.startsWith("#")) {
+        continue;
+      }
+      Matcher match = PATTERN_OUTPUT_ENV_LIST.matcher(s);
+
+      if (!match.matches()) {
+        continue;
+      }
+      kv.put(match.group(1), match.group(2));
+    }
+
+    return kv;
+  }
+
+  @Override
+  public void cancel(InterpreterContext context) {
+
+  }
+
+  @Override
+  public FormType getFormType() {
+    return FormType.NONE;
+  }
+
+  @Override
+  public int getProgress(InterpreterContext context) {
+    return 0;
+  }
+
+  /**
+   * Use python interpreter's scheduler.
+   * To make sure %python.conda paragraph and %python paragraph runs sequentially
+   */
+  @Override
+  public Scheduler getScheduler() {
+    PythonInterpreter pythonInterpreter = getPythonInterpreter();
+    if (pythonInterpreter != null) {
+      return pythonInterpreter.getScheduler();
+    } else {
+      return null;
+    }
+  }
+
+  public static String runCommand(List<String> commands)
+      throws IOException, InterruptedException {
+
+    StringBuilder sb = new StringBuilder();
+
+    ProcessBuilder builder = new ProcessBuilder(commands);
+    builder.redirectErrorStream(true);
+    Process process = builder.start();
+    InputStream stdout = process.getInputStream();
+    BufferedReader br = new BufferedReader(new InputStreamReader(stdout));
+    String line;
+    while ((line = br.readLine()) != null) {
+      sb.append(line);
+      sb.append("\n");
+    }
+    int r = process.waitFor(); // Let the process finish.
+
+    if (r != 0) {
+      throw new RuntimeException("Failed to execute `" +
+          StringUtils.join(commands, " ") + "` exited with " + r);
+    }
+
+    return sb.toString();
+  }
+
+  public static String runCommand(String ... command)
+      throws IOException, InterruptedException {
+
+    List<String> list = new ArrayList<>(command.length);
+    for (String arg : command) {
+      list.add(arg);
+    }
+
+    return runCommand(list);
+  }
+
+  public static List<String> getRestArgsFromMatcher(Matcher m) {
+    // Arrays.asList just returns fixed-size, so we should use ctor instead of
+    return new ArrayList<>(Arrays.asList(m.group(1).split(" ")));
+  }
+}

python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java

@@ -60,7 +60,7 @@ import py4j.GatewayServer;
 public class PythonInterpreter extends Interpreter implements ExecuteResultHandler {
   private static final Logger LOG = LoggerFactory.getLogger(PythonInterpreter.class);
   public static final String ZEPPELIN_PYTHON = "python/zeppelin_python.py";
-  public static final String ZEPPELIN_PY4JPATH = "python/py4j-0.8.2.1.zip";
+  public static final String ZEPPELIN_PY4JPATH = "python/py4j-0.9-src.zip";
   public static final String DEFAULT_ZEPPELIN_PYTHON = "python";
   public static final String MAX_RESULT = "zeppelin.python.maxResult";
 
@@ -68,6 +68,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
   private Pattern errorInLastLine = Pattern.compile(".*(Error|Exception): .*$");
   private String pythonPath;
   private int maxResult;
+  private String py4jLibPath;
 
   private String pythonCommand = DEFAULT_ZEPPELIN_PYTHON;
 
@@ -86,6 +87,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
 
   Integer statementSetNotifier = new Integer(0);
 
+
   public PythonInterpreter(Properties property) {
     super(property);
     try {
@@ -96,6 +98,28 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
     }
   }
 
+  private void createPy4jLib() {
+    py4jLibPath = System.getProperty("user.dir") +
+            File.separator + "interpreter" + File.separator + ZEPPELIN_PY4JPATH;
+    File py4jLib = new File(py4jLibPath);
+    if (!py4jLib.exists()) {
+      return;
+    }
+
+    ClassLoader classLoader = getClass().getClassLoader();
+    try {
+      FileOutputStream outStream = new FileOutputStream(py4jLib);
+      IOUtils.copy(
+          classLoader.getResourceAsStream(ZEPPELIN_PY4JPATH),
+          outStream);
+      outStream.close();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
+
+    logger.info("py4j library path : {}", py4jLibPath);
+  }
+
   private void createPythonScript() {
     ClassLoader classLoader = getClass().getClassLoader();
     File out = new File(scriptPath);
@@ -119,6 +143,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
 
   private void createGatewayServerAndStartScript() {
     createPythonScript();
+    createPy4jLib();
 
     port = findRandomOpenPortOnAllLocalInterfaces();
     gatewayServer = new GatewayServer(this, port);
@@ -146,9 +171,10 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
 
     try {
       Map env = EnvironmentUtils.getProcEnvironment();
-      env.put("PYTHONPATH", ZEPPELIN_PY4JPATH);
+      if (!env.containsKey("PYTHONPATH")) {
+        env.put("PYTHONPATH", py4jLibPath);
+      }
       executor.execute(cmd, env, this);
-      //executor.execute(cmd);
       pythonscriptRunning = true;
     } catch (IOException e) {
       throw new InterpreterException(e);
@@ -169,7 +195,6 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
     if (intpGroup != null && intpGroup.getInterpreterHookRegistry() != null) {
       registerHook(HookType.POST_EXEC_DEV, "z._displayhook()");
     }
-
     // Add matplotlib display hook
     createGatewayServerAndStartScript();
   }
@@ -269,6 +294,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
 
   @Override
   public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) {
+    logger.info("origial python interpreter---->" + this);
     if (cmd == null || cmd.isEmpty()) {
       return new InterpreterResult(Code.SUCCESS, "");
     }

python/src/test/java/org/apache/zeppelin/python/PythonCondaInterpreterTest.java

@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.python;
+
+
+import org.apache.zeppelin.display.GUI;
+import org.apache.zeppelin.interpreter.*;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.regex.Matcher;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.*;
+
+public class PythonCondaInterpreterTest {
+  private PythonCondaInterpreter conda;
+  private PythonInterpreter python;
+
+  @Before
+  public void setUp() {
+    conda = spy(new PythonCondaInterpreter(new Properties()));
+    python = mock(PythonInterpreter.class);
+
+    InterpreterGroup group = new InterpreterGroup();
+    group.put("note", Arrays.asList(python, conda));
+    python.setInterpreterGroup(group);
+    conda.setInterpreterGroup(group);
+
+    doReturn(python).when(conda).getPythonInterpreter();
+  }
+
+  private void setMockCondaEnvList() throws IOException, InterruptedException {
+    Map<String, String> envList = new LinkedHashMap<String, String>();
+    envList.put("env1", "/path1");
+    envList.put("env2", "/path2");
+    doReturn(envList).when(conda).getCondaEnvs();
+  }
+
+  @Test
+  public void testListEnv() throws IOException, InterruptedException {
+    setMockCondaEnvList();
+
+    // list available env
+    InterpreterContext context = getInterpreterContext();
+    InterpreterResult result = conda.interpret("env list", context);
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+
+    assertTrue(result.toString().contains(">env1<"));
+    assertTrue(result.toString().contains("/path1<"));
+    assertTrue(result.toString().contains(">env2<"));
+    assertTrue(result.toString().contains("/path2<"));
+  }
+
+  @Test
+  public void testActivateEnv() throws IOException, InterruptedException {
+    setMockCondaEnvList();
+
+    InterpreterContext context = getInterpreterContext();
+    conda.interpret("activate env1", context);
+    verify(python, times(1)).open();
+    verify(python, times(1)).close();
+    verify(python).setPythonCommand("/path1/bin/python");
+  }
+
+  @Test
+  public void testDeactivate() {
+    InterpreterContext context = getInterpreterContext();
+    conda.interpret("deactivate", context);
+    verify(python, times(1)).open();
+    verify(python, times(1)).close();
+    verify(python).setPythonCommand("python");
+  }
+
+  @Test
+  public void testParseCondaCommonStdout()
+      throws IOException, InterruptedException {
+
+    StringBuilder sb = new StringBuilder()
+        .append("# comment1\n")
+        .append("# comment2\n")
+        .append("env1     /location1\n")
+        .append("env2     /location2\n");
+
+    Map<String, String> locationPerEnv =
+        PythonCondaInterpreter.parseCondaCommonStdout(sb.toString());
+
+    assertEquals("/location1", locationPerEnv.get("env1"));
+    assertEquals("/location2", locationPerEnv.get("env2"));
+  }
+
+  @Test
+  public void testGetRestArgsFromMatcher() {
+    Matcher m =
+        PythonCondaInterpreter.PATTERN_COMMAND_ENV.matcher("env remove --name test --yes");
+    m.matches();
+
+    List<String> restArgs = PythonCondaInterpreter.getRestArgsFromMatcher(m);
+    List<String> expected = Arrays.asList(new String[]{"remove", "--name", "test", "--yes"});
+    assertEquals(expected, restArgs);
+  }
+
+  private InterpreterContext getInterpreterContext() {
+    return new InterpreterContext(
+        "noteId",
+        "paragraphId",
+        null,
+        "paragraphTitle",
+        "paragraphText",
+        new AuthenticationInfo(),
+        new HashMap<String, Object>(),
+        new GUI(),
+        null,
+        null,
+        null,
+        new InterpreterOutput(null));
+  }
+
+
+}