1# -*- coding: utf-8 -*-
2
3# Copyright (c) 2019 - 2021 Detlev Offenbach <detlev@die-offenbachs.de>
4#
5
6"""
7Package implementing the conda GUI logic.
8"""
9
10import json
11import os
12import contextlib
13
14from PyQt5.QtCore import pyqtSignal, QObject, QProcess, QCoreApplication
15from PyQt5.QtWidgets import QDialog
16
17from E5Gui import E5MessageBox
18
19import Globals
20import Preferences
21
22from . import rootPrefix, condaVersion
23from .CondaExecDialog import CondaExecDialog
24
25
26class Conda(QObject):
27    """
28    Class implementing the conda GUI logic.
29
30    @signal condaEnvironmentCreated() emitted to indicate the creation of
31        a new environment
32    @signal condaEnvironmentRemoved() emitted to indicate the removal of
33        an environment
34    """
35    condaEnvironmentCreated = pyqtSignal()
36    condaEnvironmentRemoved = pyqtSignal()
37
38    RootName = QCoreApplication.translate("Conda", "<root>")
39
40    def __init__(self, parent=None):
41        """
42        Constructor
43
44        @param parent parent
45        @type QObject
46        """
47        super().__init__(parent)
48
49        self.__ui = parent
50
51    #######################################################################
52    ## environment related methods below
53    #######################################################################
54
55    def createCondaEnvironment(self, arguments):
56        """
57        Public method to create a conda environment.
58
59        @param arguments list of command line arguments
60        @type list of str
61        @return tuple containing a flag indicating success, the directory of
62            the created environment (aka. prefix) and the corresponding Python
63            interpreter
64        @rtype tuple of (bool, str, str)
65        """
66        args = ["create", "--json", "--yes"] + arguments
67
68        dlg = CondaExecDialog("create", self.__ui)
69        dlg.start(args)
70        dlg.exec()
71        ok, resultDict = dlg.getResult()
72
73        if ok:
74            if ("actions" in resultDict and
75                    "PREFIX" in resultDict["actions"]):
76                prefix = resultDict["actions"]["PREFIX"]
77            elif "prefix" in resultDict:
78                prefix = resultDict["prefix"]
79            elif "dst_prefix" in resultDict:
80                prefix = resultDict["dst_prefix"]
81            else:
82                prefix = ""
83
84            # determine Python executable
85            if prefix:
86                pathPrefixes = [
87                    prefix,
88                    rootPrefix()
89                ]
90            else:
91                pathPrefixes = [
92                    rootPrefix()
93                ]
94            for pathPrefix in pathPrefixes:
95                python = (
96                    os.path.join(pathPrefix, "python.exe")
97                    if Globals.isWindowsPlatform() else
98                    os.path.join(pathPrefix, "bin", "python")
99                )
100                if os.path.exists(python):
101                    break
102            else:
103                python = ""
104
105            self.condaEnvironmentCreated.emit()
106            return True, prefix, python
107        else:
108            return False, "", ""
109
110    def removeCondaEnvironment(self, name="", prefix=""):
111        """
112        Public method to remove a conda environment.
113
114        @param name name of the environment
115        @type str
116        @param prefix prefix of the environment
117        @type str
118        @return flag indicating success
119        @rtype bool
120        @exception RuntimeError raised to indicate an error in parameters
121
122        Note: only one of name or prefix must be given.
123        """
124        if name and prefix:
125            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
126
127        if not name and not prefix:
128            raise RuntimeError("One of 'name' or 'prefix' must be given.")
129
130        args = [
131            "remove",
132            "--json",
133            "--quiet",
134            "--all",
135        ]
136        if name:
137            args.extend(["--name", name])
138        elif prefix:
139            args.extend(["--prefix", prefix])
140
141        exe = Preferences.getConda("CondaExecutable")
142        if not exe:
143            exe = "conda"
144
145        proc = QProcess()
146        proc.start(exe, args)
147        if not proc.waitForStarted(15000):
148            E5MessageBox.critical(
149                self.__ui,
150                self.tr("conda remove"),
151                self.tr("""The conda executable could not be started."""))
152            return False
153        else:
154            proc.waitForFinished(15000)
155            output = str(proc.readAllStandardOutput(),
156                         Preferences.getSystem("IOEncoding"),
157                         'replace').strip()
158            try:
159                jsonDict = json.loads(output)
160            except Exception:
161                E5MessageBox.critical(
162                    self.__ui,
163                    self.tr("conda remove"),
164                    self.tr("""The conda executable returned invalid data."""))
165                return False
166
167            if "error" in jsonDict:
168                E5MessageBox.critical(
169                    self.__ui,
170                    self.tr("conda remove"),
171                    self.tr("<p>The conda executable returned an error.</p>"
172                            "<p>{0}</p>").format(jsonDict["message"]))
173                return False
174
175            if jsonDict["success"]:
176                self.condaEnvironmentRemoved.emit()
177
178            return jsonDict["success"]
179
180        return False
181
182    def getCondaEnvironmentsList(self):
183        """
184        Public method to get a list of all Conda environments.
185
186        @return list of tuples containing the environment name and prefix
187        @rtype list of tuples of (str, str)
188        """
189        exe = Preferences.getConda("CondaExecutable")
190        if not exe:
191            exe = "conda"
192
193        environmentsList = []
194
195        proc = QProcess()
196        proc.start(exe, ["info", "--json"])
197        if proc.waitForStarted(15000) and proc.waitForFinished(15000):
198            output = str(proc.readAllStandardOutput(),
199                         Preferences.getSystem("IOEncoding"),
200                         'replace').strip()
201            try:
202                jsonDict = json.loads(output)
203            except Exception:
204                jsonDict = {}
205
206            if "envs" in jsonDict:
207                for prefix in jsonDict["envs"][:]:
208                    if prefix == jsonDict["root_prefix"]:
209                        if not jsonDict["root_writable"]:
210                            # root prefix is listed but not writable
211                            continue
212                        name = self.RootName
213                    else:
214                        name = os.path.basename(prefix)
215
216                    environmentsList.append((name, prefix))
217
218        return environmentsList
219
220    #######################################################################
221    ## package related methods below
222    #######################################################################
223
224    def getInstalledPackages(self, name="", prefix=""):
225        """
226        Public method to get a list of installed packages of a conda
227        environment.
228
229        @param name name of the environment
230        @type str
231        @param prefix prefix of the environment
232        @type str
233        @return list of installed packages. Each entry is a tuple containing
234            the package name, version and build.
235        @rtype list of tuples of (str, str, str)
236        @exception RuntimeError raised to indicate an error in parameters
237
238        Note: only one of name or prefix must be given.
239        """
240        if name and prefix:
241            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
242
243        if not name and not prefix:
244            raise RuntimeError("One of 'name' or 'prefix' must be given.")
245
246        args = [
247            "list",
248            "--json",
249        ]
250        if name:
251            args.extend(["--name", name])
252        elif prefix:
253            args.extend(["--prefix", prefix])
254
255        exe = Preferences.getConda("CondaExecutable")
256        if not exe:
257            exe = "conda"
258
259        packages = []
260
261        proc = QProcess()
262        proc.start(exe, args)
263        if proc.waitForStarted(15000) and proc.waitForFinished(30000):
264            output = str(proc.readAllStandardOutput(),
265                         Preferences.getSystem("IOEncoding"),
266                         'replace').strip()
267            try:
268                jsonList = json.loads(output)
269            except Exception:
270                jsonList = []
271
272            for package in jsonList:
273                if isinstance(package, dict):
274                    packages.append((
275                        package["name"],
276                        package["version"],
277                        package["build_string"]
278                    ))
279                else:
280                    parts = package.rsplit("-", 2)
281                    while len(parts) < 3:
282                        parts.append("")
283                    packages.append(tuple(parts))
284
285        return packages
286
287    def getUpdateablePackages(self, name="", prefix=""):
288        """
289        Public method to get a list of updateable packages of a conda
290        environment.
291
292        @param name name of the environment
293        @type str
294        @param prefix prefix of the environment
295        @type str
296        @return list of installed packages. Each entry is a tuple containing
297            the package name, version and build.
298        @rtype list of tuples of (str, str, str)
299        @exception RuntimeError raised to indicate an error in parameters
300
301        Note: only one of name or prefix must be given.
302        """
303        if name and prefix:
304            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
305
306        if not name and not prefix:
307            raise RuntimeError("One of 'name' or 'prefix' must be given.")
308
309        args = [
310            "update",
311            "--json",
312            "--quiet",
313            "--all",
314            "--dry-run",
315        ]
316        if name:
317            args.extend(["--name", name])
318        elif prefix:
319            args.extend(["--prefix", prefix])
320
321        exe = Preferences.getConda("CondaExecutable")
322        if not exe:
323            exe = "conda"
324
325        packages = []
326
327        proc = QProcess()
328        proc.start(exe, args)
329        if proc.waitForStarted(15000) and proc.waitForFinished(30000):
330            output = str(proc.readAllStandardOutput(),
331                         Preferences.getSystem("IOEncoding"),
332                         'replace').strip()
333            try:
334                jsonDict = json.loads(output)
335            except Exception:
336                jsonDict = {}
337
338            if "actions" in jsonDict and "LINK" in jsonDict["actions"]:
339                for linkEntry in jsonDict["actions"]["LINK"]:
340                    if isinstance(linkEntry, dict):
341                        packages.append((
342                            linkEntry["name"],
343                            linkEntry["version"],
344                            linkEntry["build_string"]
345                        ))
346                    else:
347                        package = linkEntry.split()[0]
348                        parts = package.rsplit("-", 2)
349                        while len(parts) < 3:
350                            parts.append("")
351                        packages.append(tuple(parts))
352
353        return packages
354
355    def updatePackages(self, packages, name="", prefix=""):
356        """
357        Public method to update packages of a conda environment.
358
359        @param packages list of package names to be updated
360        @type list of str
361        @param name name of the environment
362        @type str
363        @param prefix prefix of the environment
364        @type str
365        @return flag indicating success
366        @rtype bool
367        @exception RuntimeError raised to indicate an error in parameters
368
369        Note: only one of name or prefix must be given.
370        """
371        if name and prefix:
372            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
373
374        if not name and not prefix:
375            raise RuntimeError("One of 'name' or 'prefix' must be given.")
376
377        if packages:
378            args = [
379                "update",
380                "--json",
381                "--yes",
382            ]
383            if name:
384                args.extend(["--name", name])
385            elif prefix:
386                args.extend(["--prefix", prefix])
387            args.extend(packages)
388
389            dlg = CondaExecDialog("update", self.__ui)
390            dlg.start(args)
391            dlg.exec()
392            ok, _ = dlg.getResult()
393        else:
394            ok = False
395
396        return ok
397
398    def updateAllPackages(self, name="", prefix=""):
399        """
400        Public method to update all packages of a conda environment.
401
402        @param name name of the environment
403        @type str
404        @param prefix prefix of the environment
405        @type str
406        @return flag indicating success
407        @rtype bool
408        @exception RuntimeError raised to indicate an error in parameters
409
410        Note: only one of name or prefix must be given.
411        """
412        if name and prefix:
413            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
414
415        if not name and not prefix:
416            raise RuntimeError("One of 'name' or 'prefix' must be given.")
417
418        args = [
419            "update",
420            "--json",
421            "--yes",
422            "--all"
423        ]
424        if name:
425            args.extend(["--name", name])
426        elif prefix:
427            args.extend(["--prefix", prefix])
428
429        dlg = CondaExecDialog("update", self.__ui)
430        dlg.start(args)
431        dlg.exec()
432        ok, _ = dlg.getResult()
433
434        return ok
435
436    def installPackages(self, packages, name="", prefix=""):
437        """
438        Public method to install packages into a conda environment.
439
440        @param packages list of package names to be installed
441        @type list of str
442        @param name name of the environment
443        @type str
444        @param prefix prefix of the environment
445        @type str
446        @return flag indicating success
447        @rtype bool
448        @exception RuntimeError raised to indicate an error in parameters
449
450        Note: only one of name or prefix must be given.
451        """
452        if name and prefix:
453            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
454
455        if not name and not prefix:
456            raise RuntimeError("One of 'name' or 'prefix' must be given.")
457
458        if packages:
459            args = [
460                "install",
461                "--json",
462                "--yes",
463            ]
464            if name:
465                args.extend(["--name", name])
466            elif prefix:
467                args.extend(["--prefix", prefix])
468            args.extend(packages)
469
470            dlg = CondaExecDialog("install", self.__ui)
471            dlg.start(args)
472            dlg.exec()
473            ok, _ = dlg.getResult()
474        else:
475            ok = False
476
477        return ok
478
479    def uninstallPackages(self, packages, name="", prefix=""):
480        """
481        Public method to uninstall packages of a conda environment (including
482        all no longer needed dependencies).
483
484        @param packages list of package names to be uninstalled
485        @type list of str
486        @param name name of the environment
487        @type str
488        @param prefix prefix of the environment
489        @type str
490        @return flag indicating success
491        @rtype bool
492        @exception RuntimeError raised to indicate an error in parameters
493
494        Note: only one of name or prefix must be given.
495        """
496        if name and prefix:
497            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
498
499        if not name and not prefix:
500            raise RuntimeError("One of 'name' or 'prefix' must be given.")
501
502        if packages:
503            from UI.DeleteFilesConfirmationDialog import (
504                DeleteFilesConfirmationDialog)
505            dlg = DeleteFilesConfirmationDialog(
506                self.parent(),
507                self.tr("Uninstall Packages"),
508                self.tr(
509                    "Do you really want to uninstall these packages and"
510                    " their dependencies?"),
511                packages)
512            if dlg.exec() == QDialog.DialogCode.Accepted:
513                args = [
514                    "remove",
515                    "--json",
516                    "--yes",
517                ]
518                if condaVersion() >= (4, 4, 0):
519                    args.append("--prune",)
520                if name:
521                    args.extend(["--name", name])
522                elif prefix:
523                    args.extend(["--prefix", prefix])
524                args.extend(packages)
525
526                dlg = CondaExecDialog("remove", self.__ui)
527                dlg.start(args)
528                dlg.exec()
529                ok, _ = dlg.getResult()
530            else:
531                ok = False
532        else:
533            ok = False
534
535        return ok
536
537    def searchPackages(self, pattern, fullNameOnly=False, packageSpec=False,
538                       platform="", name="", prefix=""):
539        """
540        Public method to search for a package pattern of a conda environment.
541
542        @param pattern package search pattern
543        @type str
544        @param fullNameOnly flag indicating to search for full names only
545        @type bool
546        @param packageSpec flag indicating to search a package specification
547        @type bool
548        @param platform type of platform to be searched for
549        @type str
550        @param name name of the environment
551        @type str
552        @param prefix prefix of the environment
553        @type str
554        @return flag indicating success and a dictionary with package name as
555            key and list of dictionaries containing detailed data for the found
556            packages as values
557        @rtype tuple of (bool, dict of list of dict)
558        @exception RuntimeError raised to indicate an error in parameters
559
560        Note: only one of name or prefix must be given.
561        """
562        if name and prefix:
563            raise RuntimeError("Only one of 'name' or 'prefix' must be given.")
564
565        args = [
566            "search",
567            "--json",
568        ]
569        if fullNameOnly:
570            args.append("--full-name")
571        if packageSpec:
572            args.append("--spec")
573        if platform:
574            args.extend(["--platform", platform])
575        if name:
576            args.extend(["--name", name])
577        elif prefix:
578            args.extend(["--prefix", prefix])
579        args.append(pattern)
580
581        exe = Preferences.getConda("CondaExecutable")
582        if not exe:
583            exe = "conda"
584
585        packages = {}
586        ok = False
587
588        proc = QProcess()
589        proc.start(exe, args)
590        if proc.waitForStarted(15000) and proc.waitForFinished(30000):
591            output = str(proc.readAllStandardOutput(),
592                         Preferences.getSystem("IOEncoding"),
593                         'replace').strip()
594            with contextlib.suppress(Exception):
595                packages = json.loads(output)
596                ok = "error" not in packages
597
598        return ok, packages
599
600    #######################################################################
601    ## special methods below
602    #######################################################################
603
604    def updateConda(self):
605        """
606        Public method to update conda itself.
607
608        @return flag indicating success
609        @rtype bool
610        """
611        args = [
612            "update",
613            "--json",
614            "--yes",
615            "conda"
616        ]
617
618        dlg = CondaExecDialog("update", self.__ui)
619        dlg.start(args)
620        dlg.exec()
621        ok, _ = dlg.getResult()
622
623        return ok
624
625    def writeDefaultConfiguration(self):
626        """
627        Public method to create a conda configuration with default values.
628        """
629        args = [
630            "config",
631            "--write-default",
632            "--quiet"
633        ]
634
635        exe = Preferences.getConda("CondaExecutable")
636        if not exe:
637            exe = "conda"
638
639        proc = QProcess()
640        proc.start(exe, args)
641        proc.waitForStarted(15000)
642        proc.waitForFinished(30000)
643
644    def getCondaInformation(self):
645        """
646        Public method to get a dictionary containing information about conda.
647
648        @return dictionary containing information about conda
649        @rtype dict
650        """
651        exe = Preferences.getConda("CondaExecutable")
652        if not exe:
653            exe = "conda"
654
655        infoDict = {}
656
657        proc = QProcess()
658        proc.start(exe, ["info", "--json"])
659        if proc.waitForStarted(15000) and proc.waitForFinished(30000):
660            output = str(proc.readAllStandardOutput(),
661                         Preferences.getSystem("IOEncoding"),
662                         'replace').strip()
663            try:
664                infoDict = json.loads(output)
665            except Exception:
666                infoDict = {}
667
668        return infoDict
669
670    def runProcess(self, args):
671        """
672        Public method to execute the conda with the given arguments.
673
674        The conda executable is called with the given arguments and
675        waited for its end.
676
677        @param args list of command line arguments
678        @type list of str
679        @return tuple containing a flag indicating success and the output
680            of the process
681        @rtype tuple of (bool, str)
682        """
683        exe = Preferences.getConda("CondaExecutable")
684        if not exe:
685            exe = "conda"
686
687        process = QProcess()
688        process.start(exe, args)
689        procStarted = process.waitForStarted(15000)
690        if procStarted:
691            finished = process.waitForFinished(30000)
692            if finished:
693                if process.exitCode() == 0:
694                    output = str(process.readAllStandardOutput(),
695                                 Preferences.getSystem("IOEncoding"),
696                                 'replace').strip()
697                    return True, output
698                else:
699                    return (False,
700                            self.tr("conda exited with an error ({0}).")
701                            .format(process.exitCode()))
702            else:
703                process.terminate()
704                process.waitForFinished(2000)
705                process.kill()
706                process.waitForFinished(3000)
707                return False, self.tr("conda did not finish within"
708                                      " 30 seconds.")
709
710        return False, self.tr("conda could not be started.")
711
712    def cleanConda(self, cleanAction):
713        """
714        Public method to update conda itself.
715
716        @param cleanAction cleaning action to be performed (must be one of
717            the command line parameters without '--')
718        @type str
719        """
720        args = [
721            "clean",
722            "--yes",
723            "--{0}".format(cleanAction),
724        ]
725
726        dlg = CondaExecDialog("clean", self.__ui)
727        dlg.start(args)
728        dlg.exec()
729