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 QObject, QProcess

from E5Gui import E5MessageBox

import Globals
import Preferences

from . import rootPrefix


class Conda(QObject):
    """
    Class implementing the conda GUI logic.
    """
    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)
        """
        from .CondaExecDialog import CondaExecDialog
        
        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 = ""
            
            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
            
            return jsonDict["success"]
    
    def getCondaEnvironmentsList(self, listPrefixes=False):
        """
        Public method to get a list of all Conda environments.
        
        @param listPrefixes flag indicating to return prefixes instead of names
        @type bool
        @return list of environment names or prefixes
        @rtype list of str
        """
        # TODO: return environment name and prefix
        # TODO: return root environment only, if writable ('root_prefix', 'root_writable')
        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:
                    if listPrefixes:
                        environmentsList = jsonDict["envs"][:]
                    else:
                        environmentsList = [
                            os.path.basename(e) for e in jsonDict["envs"]]
        
        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)
        @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 = [
            "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(15000):
                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)
        @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",
            "--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(15000):
                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 list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @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.")
        
        # TODO: not implemented yet
    
    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 list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @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.")
        
        # TODO: not implemented yet
    
    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 list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @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.")
        
        # TODO: not implemented yet
    
    def uninstallPackages(self, packages, name="", prefix=""):
        """
        Public method to uninstall packages of a conda environment.
        
        @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 list of installed packages. Each entry is a tuple containing
            the package name, version and build.
        @rtype list of tuples of (str, str, str)
        @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.")
        
        # TODO: not implemented yet