CondaInterface/Conda.py

# -*- coding: utf-8 -*-

# Copyright (c) 2019 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Package implementing the conda GUI logic.
"""

from __future__ import unicode_literals
try:
    str = unicode       # __IGNORE_EXCEPTION__
except NameError:
    pass

import json
import os

from PyQt5.QtCore import pyqtSignal, QObject, QProcess, QCoreApplication
from PyQt5.QtWidgets import QDialog

from E5Gui import E5MessageBox

import Globals
import Preferences

from . import rootPrefix, condaVersion
from .CondaExecDialog import CondaExecDialog


class Conda(QObject):
    """
    Class implementing the conda GUI logic.
    
    @signal condaEnvironmentCreated() emitted to indicate the creation of
        a new environment
    @signal condaEnvironmentRemoved() emitted to indicate the removal of
        an environment
    """
    condaEnvironmentCreated = pyqtSignal()
    condaEnvironmentRemoved = pyqtSignal()
    
    RootName = QCoreApplication.translate("Conda", "<root>")
    
    def __init__(self, parent=None):
        """
        Constructor
        
        @param parent parent
        @type QObject
        """
        super(Conda, self).__init__(parent)
        
        self.__ui = parent
    
    #######################################################################
    ## environment related methods below
    #######################################################################
    
    def createCondaEnvironment(self, arguments):
        """
        Public method to create a conda environment.
        
        @param arguments list of command line arguments
        @type list of str
        @return tuple containing a flag indicating success, the directory of
            the created environment (aka. prefix) and the corresponding Python
            interpreter
        @rtype tuple of (bool, str, str)
        """
        args = ["create", "--json", "--yes"] + arguments
        
        dlg = CondaExecDialog("create", self.__ui)
        dlg.start(args)
        dlg.exec_()
        ok, resultDict = dlg.getResult()
        
        if ok:
            if "actions" in resultDict and \
                    "PREFIX" in resultDict["actions"]:
                prefix = resultDict["actions"]["PREFIX"]
            elif "prefix" in resultDict:
                prefix = resultDict["prefix"]
            elif "dst_prefix" in resultDict:
                prefix = resultDict["dst_prefix"]
            else:
                prefix = ""
            
            # determine Python executable
            if prefix:
                pathPrefixes = [
                    prefix,
                    rootPrefix()
                ]
            else:
                pathPrefixes = [
                    rootPrefix()
                ]
            for pathPrefix in pathPrefixes:
                if Globals.isWindowsPlatform():
                    python = os.path.join(pathPrefix, "python.exe")
                else:
                    python = os.path.join(pathPrefix, "bin", "python")
                if os.path.exists(python):
                    break
            else:
                python = ""
            
            self.condaEnvironmentCreated.emit()
            return True, prefix, python
        else:
            return False, "", ""
    
    def removeCondaEnvironment(self, name="", prefix=""):
        """
        Public method to remove a conda environment.
        
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success
        @rtype bool
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        args = [
            "remove",
            "--json",
            "--quiet",
            "--all",
        ]
        if name:
            args.extend(["--name", name])
        elif prefix:
            args.extend(["--prefix", prefix])
        
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        proc = QProcess()
        proc.start(exe, args)
        if not proc.waitForStarted(15000):
            E5MessageBox.critical(
                self.__ui,
                self.tr("conda remove"),
                self.tr("""The conda executable could not be started."""))
            return False
        else:
            proc.waitForFinished(15000)
            output = str(proc.readAllStandardOutput(),
                         Preferences.getSystem("IOEncoding"),
                         'replace').strip()
            try:
                jsonDict = json.loads(output)
            except Exception:
                E5MessageBox.critical(
                    self.__ui,
                    self.tr("conda remove"),
                    self.tr("""The conda executable returned invalid data."""))
                return False
            
            if "error" in jsonDict:
                E5MessageBox.critical(
                    self.__ui,
                    self.tr("conda remove"),
                    self.tr("<p>The conda executable returned an error.</p>"
                            "<p>{0}</p>").format(jsonDict["message"]))
                return False
            
            if jsonDict["success"]:
                self.condaEnvironmentRemoved.emit()
            
            return jsonDict["success"]
        
        return False
    
    def getCondaEnvironmentsList(self):
        """
        Public method to get a list of all Conda environments.
        
        @return list of tuples containing the environment name and prefix
        @rtype list of tuples of (str, str)
        """
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        environmentsList = []
        
        proc = QProcess()
        proc.start(exe, ["info", "--json"])
        if proc.waitForStarted(15000):
            if proc.waitForFinished(15000):
                output = str(proc.readAllStandardOutput(),
                             Preferences.getSystem("IOEncoding"),
                             'replace').strip()
                try:
                    jsonDict = json.loads(output)
                except Exception:
                    jsonDict = {}
                
                if "envs" in jsonDict:
                    for prefix in jsonDict["envs"][:]:
                        if prefix == jsonDict["root_prefix"]:
                            if not jsonDict["root_writable"]:
                                # root prefix is listed but not writable
                                continue
                            name = self.RootName
                        else:
                            name = os.path.basename(prefix)
                        
                        environmentsList.append((name, prefix))
        
        return environmentsList
    
    #######################################################################
    ## package related methods below
    #######################################################################
    
    def getInstalledPackages(self, name="", prefix=""):
        """
        Public method to get a list of installed packages of a conda
        environment.
        
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        args = [
            "list",
            "--json",
        ]
        if name:
            args.extend(["--name", name])
        elif prefix:
            args.extend(["--prefix", prefix])
        
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        packages = []
        
        proc = QProcess()
        proc.start(exe, args)
        if proc.waitForStarted(15000):
            if proc.waitForFinished(30000):
                output = str(proc.readAllStandardOutput(),
                             Preferences.getSystem("IOEncoding"),
                             'replace').strip()
                try:
                    jsonList = json.loads(output)
                except Exception:
                    jsonList = []
                
                for package in jsonList:
                    if isinstance(package, dict):
                        packages.append((
                            package["name"],
                            package["version"],
                            package["build_string"]
                        ))
                    else:
                        packages.append(tuple(package.rsplit("-", 2)))
        
        return packages
    
    def getUpdateablePackages(self, name="", prefix=""):
        """
        Public method to get a list of updateable packages of a conda
        environment.
        
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        args = [
            "update",
            "--json",
            "--quiet",
            "--all",
            "--dry-run",
        ]
        if name:
            args.extend(["--name", name])
        elif prefix:
            args.extend(["--prefix", prefix])
        
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        packages = []
        
        proc = QProcess()
        proc.start(exe, args)
        if proc.waitForStarted(15000):
            if proc.waitForFinished(30000):
                output = str(proc.readAllStandardOutput(),
                             Preferences.getSystem("IOEncoding"),
                             'replace').strip()
                try:
                    jsonDict = json.loads(output)
                except Exception:
                    jsonDict = {}
                
                if "actions" in jsonDict and "LINK" in jsonDict["actions"]:
                    for linkEntry in jsonDict["actions"]["LINK"]:
                        if isinstance(linkEntry, dict):
                            packages.append((
                                linkEntry["name"],
                                linkEntry["version"],
                                linkEntry["build_string"]
                            ))
                        else:
                            package = linkEntry.split()[0]
                            packages.append(tuple(package.rsplit("-", 2)))
        
        return packages
    
    def updatePackages(self, packages, name="", prefix=""):
        """
        Public method to update packages of a conda environment.
        
        @param packages list of package names to be updated
        @type list of str
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success
        @rtype bool
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        if packages:
            args = [
                "update",
                "--json",
                "--yes",
            ]
            if name:
                args.extend(["--name", name])
            elif prefix:
                args.extend(["--prefix", prefix])
            args.extend(packages)
            
            dlg = CondaExecDialog("update", self.__ui)
            dlg.start(args)
            dlg.exec_()
            ok, _ = dlg.getResult()
        else:
            ok = False
        
        return ok
    
    def updateAllPackages(self, name="", prefix=""):
        """
        Public method to update all packages of a conda environment.
        
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success
        @rtype bool
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        args = [
            "update",
            "--json",
            "--yes",
            "--all"
        ]
        if name:
            args.extend(["--name", name])
        elif prefix:
            args.extend(["--prefix", prefix])
        
        dlg = CondaExecDialog("update", self.__ui)
        dlg.start(args)
        dlg.exec_()
        ok, _ = dlg.getResult()
        
        return ok
    
    def installPackages(self, packages, name="", prefix=""):
        """
        Public method to install packages into a conda environment.
        
        @param packages list of package names to be installed
        @type list of str
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success
        @rtype bool
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        if packages:
            args = [
                "install",
                "--json",
                "--yes",
            ]
            if name:
                args.extend(["--name", name])
            elif prefix:
                args.extend(["--prefix", prefix])
            args.extend(packages)
            
            dlg = CondaExecDialog("install", self.__ui)
            dlg.start(args)
            dlg.exec_()
            ok, _ = dlg.getResult()
        else:
            ok = False
        
        return ok
    
    def uninstallPackages(self, packages, name="", prefix=""):
        """
        Public method to uninstall packages of a conda environment (including
        all no longer needed dependencies).
        
        @param packages list of package names to be uninstalled
        @type list of str
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success
        @rtype bool
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        if not name and not prefix:
            raise RuntimeError("One of 'name' or 'prefix' must be given.")
        
        if packages:
            from UI.DeleteFilesConfirmationDialog import \
                DeleteFilesConfirmationDialog
            dlg = DeleteFilesConfirmationDialog(
                self.parent(),
                self.tr("Uninstall Packages"),
                self.tr(
                    "Do you really want to uninstall these packages and"
                    " their dependencies?"),
                packages)
            if dlg.exec_() == QDialog.Accepted:
                args = [
                    "remove",
                    "--json",
                    "--yes",
                ]
                if condaVersion() >= (4, 4, 0):
                    args.append("--prune",)
                if name:
                    args.extend(["--name", name])
                elif prefix:
                    args.extend(["--prefix", prefix])
                args.extend(packages)
                
                dlg = CondaExecDialog("remove", self.__ui)
                dlg.start(args)
                dlg.exec_()
                ok, _ = dlg.getResult()
            else:
                ok = False
        else:
            ok = False
        
        return ok
    
    def searchPackages(self, pattern, fullNameOnly=False, packageSpec=False,
                       platform="", name="", prefix=""):
        """
        Public method to search for a package pattern of a conda environment.
        
        @param pattern package search pattern
        @type str
        @param fullNameOnly flag indicating to search for full names only
        @type bool
        @param packageSpec flag indicating to search a package specification
        @type bool
        @param platform type of platform to be searched for
        @type str
        @param name name of the environment
        @type str
        @param prefix prefix of the environment
        @type str
        @return flag indicating success and a dictionary with package name as
            key and list of dictionaries containing detailed data for the found
            packages as values
        @rtype tuple of (bool, dict of list of dict)
        @exception RuntimeError raised to indicate an error in parameters
        
        Note: only one of name or prefix must be given.
        """
        if name and prefix:
            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
        
        args = [
            "search",
            "--json",
        ]
        if fullNameOnly:
            args.append("--full-name")
        if packageSpec:
            args.append("--spec")
        if platform:
            args.extend(["--platform", platform])
        if name:
            args.extend(["--name", name])
        elif prefix:
            args.extend(["--prefix", prefix])
        args.append(pattern)
        
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        packages = {}
        ok = False
        
        proc = QProcess()
        proc.start(exe, args)
        if proc.waitForStarted(15000):
            if proc.waitForFinished(30000):
                output = str(proc.readAllStandardOutput(),
                             Preferences.getSystem("IOEncoding"),
                             'replace').strip()
                try:
                    packages = json.loads(output)
                    ok = "error" not in packages
                except Exception:
                    # return values for errors is already set
                    pass
        
        return ok, packages
    
    #######################################################################
    ## special methods below
    #######################################################################
    
    def updateConda(self):
        """
        Public method to update conda itself.
        
        @return flag indicating success
        @rtype bool
        """
        args = [
            "update",
            "--json",
            "--yes",
            "conda"
        ]
        
        dlg = CondaExecDialog("update", self.__ui)
        dlg.start(args)
        dlg.exec_()
        ok, _ = dlg.getResult()
        
        return ok
    
    def writeDefaultConfiguration(self):
        """
        Public method to create a conda configuration with default values.
        """
        args = [
            "config",
            "--write-default",
            "--quiet"
        ]
        
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        proc = QProcess()
        proc.start(exe, args)
        proc.waitForStarted(15000)
        proc.waitForFinished(30000)
    
    def getCondaInformation(self):
        """
        Public method to get a dictionary containing information about conda.
        
        @return dictionary containing information about conda
        @rtype dict
        """
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        infoDict = {}
        
        proc = QProcess()
        proc.start(exe, ["info", "--json"])
        if proc.waitForStarted(15000):
            if proc.waitForFinished(30000):
                output = str(proc.readAllStandardOutput(),
                             Preferences.getSystem("IOEncoding"),
                             'replace').strip()
                try:
                    infoDict = json.loads(output)
                except Exception:
                    infoDict = {}
        
        return infoDict
    
    def runProcess(self, args):
        """
        Public method to execute the conda with the given arguments.
        
        The conda executable is called with the given arguments and
        waited for its end.
        
        @param args list of command line arguments
        @type list of str
        @return tuple containing a flag indicating success and the output
            of the process
        @rtype tuple of (bool, str)
        """
        exe = Preferences.getConda("CondaExecutable")
        if not exe:
            exe = "conda"
        
        process = QProcess()
        process.start(exe, args)
        procStarted = process.waitForStarted(15000)
        if procStarted:
            finished = process.waitForFinished(30000)
            if finished:
                if process.exitCode() == 0:
                    output = str(process.readAllStandardOutput(),
                                 Preferences.getSystem("IOEncoding"),
                                 'replace').strip()
                    return True, output
                else:
                    return (False,
                            self.tr("conda exited with an error ({0}).")
                            .format(process.exitCode()))
            else:
                process.terminate()
                process.waitForFinished(2000)
                process.kill()
                process.waitForFinished(3000)
                return False, self.tr("conda did not finish within"
                                      " 30 seconds.")
        
        return False, self.tr("conda could not be started.")
    
    def cleanConda(self, cleanAction):
        """
        Public method to update conda itself.
        
        @param cleanAction cleaning action to be performed (must be one of
            the command line parameters without '--')
        @type str
        """
        args = [
            "clean",
            "--yes",
            "--{0}".format(cleanAction),
        ]
        
        dlg = CondaExecDialog("clean", self.__ui)
        dlg.start(args)
        dlg.exec_()