File: C:/Users/fred/anaconda3/Lib/site-packages/navigator_updater/widgets/dialogs/main_dialog.py
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright 2017 Anaconda, Inc.
#
# May be copied and distributed freely only as part of an Anaconda or
# Miniconda installation.
# -----------------------------------------------------------------------------
"""Anaconda Navigator Updater main dialog."""
from __future__ import annotations
__all__ = ['MainDialog']
from collections.abc import Callable, Mapping, Sequence
from enum import auto, IntEnum
from html import escape
from json import loads
from os.path import basename
from shlex import join
from subprocess import check_output, Popen, SubprocessError # nosec
from sys import platform
from typing import Any, cast, Final, overload, TYPE_CHECKING, TypeVar
from qtpy.QtCore import QObject, QSize, Qt, QThread, QTimer, Signal
from qtpy.QtSvg import QSvgWidget
from qtpy.QtWidgets import QHBoxLayout, QVBoxLayout
from navigator_updater.config import NAVIGATOR_LOCKFILE
from navigator_updater.external import filelock
from navigator_updater.static.images import ANACONDA_LOGO
from navigator_updater.utils.logs import logger as LOGGER
from navigator_updater.utils.misc import set_windows_app_user_model_id
from navigator_updater.utils.version_utils import Version
from navigator_updater.widgets import ButtonNormal, ButtonPrimary, LabelBase
from navigator_updater.widgets.dialogs import DialogBase
if TYPE_CHECKING:
from typing_extensions import Self
class CondaMeta: # pylint: disable=too-few-public-methods
"""
Metadata on conda command to run.
Collected by :class:`~CondaCall` and used by :class:`~CondaThread`.
:param arguments: Method to generate arguments with.
:param converters: List of functions to generate values for attributes of a parent :class:`~CondaState`.
"""
__slots__ = ('arguments', 'converters')
def __init__(
self,
arguments: Callable[[CondaState], Sequence[str]],
*converters: Callable[[CondaState, Any], Any],
) -> None:
"""Initialize new instances of :class:`~CondaMeta`."""
self.arguments: Final[Callable[[CondaState], Sequence[str]]] = arguments
self.converters: Final[list[Callable[[CondaState, Any], Any]]] = list(converters)
@property
def name(self) -> str:
"""
Name of the method/thread.
Used to identify :class:`~CondaThread`.
"""
return self.arguments.__name__
class CondaThread(QThread):
"""Thread for conda commands."""
done = Signal(object)
def __init__(self, parent: CondaState, meta: CondaMeta) -> None:
"""Initialize new instance of a :class:`~CondaThread`."""
super().__init__()
self.__meta: Final[CondaMeta] = meta
self.__parent: Final[CondaState] = parent
self.done.connect(self.__done)
self.finished.connect(self.__finished)
LOGGER.debug('<%s> created', self.__meta.name)
def run(self) -> None:
"""Execute the thread."""
arguments: list[str] = ['conda', *self.__meta.arguments(self.__parent)]
prefix: str = f'<{self.__meta.name}> [{join(arguments)}]'
try:
LOGGER.debug('%s calling command', prefix)
result: bytes = check_output(arguments).rsplit(b'\x00', 1)[-1] # nosec
except SubprocessError as exception:
LOGGER.error(
'%s failed with %s; stdout: %s; stderr: %s',
prefix,
str(getattr(exception, 'returncode', '<timeout>')),
str(getattr(exception, 'stdout', None) or '<empty>'),
str(getattr(exception, 'stderr', None) or '<empty>'),
)
self.done.emit(None)
return
try:
output: Any = loads(result)
except ValueError:
LOGGER.error('%s generated invalid output', prefix)
self.done.emit(None)
return
self.done.emit(output)
def start(self, priority: int = QThread.InheritPriority) -> None:
"""Start the thread."""
if self.__meta.name in self.__parent._busy: # pylint: disable=protected-access
raise TypeError(f'thread <{self.__meta.name}> is already running')
self.__parent._busy[self.__meta.name] = self # pylint: disable=protected-access
LOGGER.debug('<%s> starting thread', self.__meta.name)
super().start(priority)
def __done(self, content: Any) -> None:
"""Register results of conda command."""
converter: Callable[[CondaState, Any], Any]
for converter in self.__meta.converters:
value: Any = converter(self.__parent, content)
setattr(self.__parent, converter.__name__, value)
LOGGER.debug('<%s> updated %s = %r', self.__meta.name, converter.__name__, value)
def __finished(self) -> None:
"""Finalize thread execution."""
LOGGER.debug('<%s> thread finished', self.__meta.name)
if self.__parent._busy.pop(self.__meta.name) is not self: # pylint: disable=protected-access
raise RuntimeError(f'unexpected thread finished for <{self.__meta.name!r}>')
if self.__meta.converters:
self.__parent.updated.emit(self.__meta.name)
def __call__(self, callback: Callable[[Any | None], Any] | None = None) -> None:
"""Start the thread."""
if callback is not None:
def wrapper(value: Any) -> None:
callback(value)
self.done.disconnect(wrapper)
self.done.connect(wrapper)
self.start()
T = TypeVar('T')
class CondaCall:
"""
Wrapper for :class:`~CondaState` methods that start conda processes.
Wrapped function must return list of conda arguments (skipping :code:`conda`, only what follows it).
"""
__slots__ = ('__meta',)
def __init__(self, arguments: Callable[[CondaState], Sequence[str]]) -> None:
"""Initialize new instance of a :class:`~CondaCall`."""
self.__meta: Final[CondaMeta] = CondaMeta(arguments)
def output(self, default: T) -> Callable[[Callable[[CondaState, Any], T]], T]:
"""
Register conda call output.
This will create a lazy-property-like attribute. Should be used as a decorator for a function that converts
output of a conda command to the target value.
"""
def wrapper(converter: Callable[[CondaState, Any], T]) -> T:
self.__meta.converters.append(converter)
return default
return wrapper
@overload
def __get__(self, instance: None, owner: type[CondaState]) -> Self:
"""Retrieve value of an attribute."""
@overload
def __get__(self, instance: CondaState, owner: type[CondaState]) -> CondaThread:
"""Retrieve value of an attribute."""
def __get__(self, instance: CondaState | None, owner: type[CondaState]) -> Self | CondaThread:
"""Retrieve value of an attribute."""
if instance is None:
return self
return CondaThread(meta=self.__meta, parent=instance)
CallableT = TypeVar('CallableT', bound=Callable)
class Keyword(IntEnum):
"""
State of a missing package version.
- **WAITING** - value is yet to be fetched.
- **EMPTY** - package is not installed or missing in repository.
- **FAILED** - there was an error fetching package version.
"""
EMPTY = auto()
FAILED = auto()
WAITING = auto()
FALLBACK: Final[Mapping[Keyword | Version, Keyword | Version]] = {Keyword.WAITING: Keyword.FAILED}
class CondaState(QObject):
"""Container for conda-related data."""
updated = Signal(str)
def __init__(self, package: str = 'anaconda-navigator', application: str | None = None) -> None:
"""Initialize new instance of a :class:`~CondaState`."""
if application is None:
application = package
super().__init__()
self.application: Final[str] = application
self.package: Final[str] = package
self._busy: Final[dict[str, QThread]] = {}
@CondaCall
def fetch_info(self) -> list[str]:
"""Fetch primary details of conda environment from conda."""
return ['info', '--json']
@fetch_info.output(default='base')
def environment(self, content: dict[str, Any] | None) -> str:
"""Name of the environment currently used by conda."""
if content is None:
return self.environment
return basename(
content.get('active_prefix', '') or content.get('root_prefix', '') or self.environment
)
@fetch_info.output(default=False)
def read_only(self, content: dict[str, Any] | None) -> bool:
"""Whether root conda environment is read-only."""
if content is None:
return self.read_only
result: bool | None = content.get('root_writable', None)
if result is None:
return self.read_only
return not result
@CondaCall
def find_current(self) -> list[str]:
"""Fetch :attr:`~CondaState.current` from conda."""
return ['list', '--json']
@find_current.output(default=cast('Keyword | Version', Keyword.WAITING))
def current(self, content: list[dict[str, Any]] | None) -> Keyword | Version:
"""Version of the installed package."""
if content is None:
LOGGER.error('failed to fetch current version of the navigator')
return FALLBACK.get(self.current, self.current)
item: dict[str, Any]
for item in content:
if item['name'] == self.package:
result: Version = Version(item['version'])
LOGGER.debug('found current version of the navigator: %s', str(result))
return result
LOGGER.debug('looks like there is no current version of the navigator')
return Keyword.EMPTY
@CondaCall
def find_latest(self) -> list[str]:
"""Fetch :attr:`~CondaState.latest` from conda."""
return ['search', '--json', self.package]
@find_latest.output(default=cast('Keyword | Version', Keyword.WAITING))
def latest(self, content: dict[str, list[dict[str, Any]]] | None) -> Keyword | Version:
"""Version of the package available in the repository."""
if content is None:
LOGGER.error('failed to fetch latest version of the navigator')
return FALLBACK.get(self.latest, self.latest)
if self.package in content:
result: Version = max(Version(item['version']) for item in content[self.package])
LOGGER.debug('found latest version of the navigator: %s', str(self.latest))
return result
LOGGER.debug('looks like there is no latest version of the navigator')
return Keyword.EMPTY
@CondaCall
def install(self) -> list[str]:
"""Install :attr:`~CondaState.latest` version of a package in current environment."""
latest: Keyword | Version = self.latest
package: str = self.package
if isinstance(latest, Version):
package += f'={latest}'
return ['install', '--json', '--yes', package]
@install.output(default=False)
def install_failed(self, content: Any) -> bool:
"""Whether last attempt to install navigator failed or not."""
return content is None
def run(self) -> None:
"""Run application from current repository."""
kwargs: dict[str, Any] = {}
if platform == 'win32':
from subprocess import STARTUPINFO # type: ignore # nosec # pylint: disable=import-outside-toplevel
kwargs['close_fds'] = True
kwargs['creationflags'] = 0x08000200
kwargs['startupinfo'] = STARTUPINFO(dwFlags=0x01)
else:
kwargs['start_new_session'] = True
LOGGER.debug('running %s', self.application)
# pylint: disable=consider-using-with
Popen(['conda', 'run', self.application], **kwargs) # nosec
@property
def active(self) -> bool:
"""Conda is currently active doing something."""
return bool(self._busy)
@property
def busy(self) -> bool:
"""Conda is busy updating environment."""
return 'install' in self._busy
@property
def ready(self) -> bool:
"""Primary details were already fetched."""
return Keyword.WAITING not in (self.current, self.latest)
class MainDialog(DialogBase): # pylint: disable=too-many-instance-attributes
"""Main dialog for the anaconda navgator updater."""
def __init__(self) -> None: # pylint: disable=too-many-statements
"""Initialize new instance of a :class:`~MainDialog`."""
super().__init__()
self.state: Final[CondaState] = CondaState()
self.state.updated.connect(self.update)
# Layout
self.label_icon: Final[QSvgWidget] = QSvgWidget()
self.label_icon.load(ANACONDA_LOGO)
self.label_icon.setMaximumSize(QSize(64, 64))
self.label_icon.setMinimumSize(QSize(64, 64))
layout_icon = QVBoxLayout()
layout_icon.addWidget(self.label_icon)
layout_icon.addStretch()
self.label_message: Final[LabelBase] = LabelBase(
'We strongly recommend you to keep Anaconda Navigator up to date.',
)
self.label_message.setAlignment(Qt.AlignLeft | Qt.AlignTop)
self.label_message.setWordWrap(True)
self.label_status = LabelBase('')
self.label_status.setWordWrap(True)
layout_status: Final[QHBoxLayout] = QHBoxLayout()
layout_status.addWidget(self.label_status)
layout_status.addSpacing(12)
layout_text: Final[QVBoxLayout] = QVBoxLayout()
layout_text.addWidget(self.label_message)
layout_text.addStretch()
layout_text.addSpacing(12)
layout_text.addLayout(layout_status)
layout_top: Final[QHBoxLayout] = QHBoxLayout()
layout_top.addLayout(layout_icon)
layout_top.addSpacing(12)
layout_top.addLayout(layout_text)
self.button_cancel: Final[ButtonNormal] = ButtonNormal('Dismiss')
self.button_cancel.setFocusPolicy(Qt.NoFocus)
self.button_cancel.clicked.connect(self.reject)
self.button_update: Final[ButtonPrimary] = ButtonPrimary('Update now')
self.button_update.setAutoDefault(True)
self.button_update.setDisabled(True)
self.button_update.setFocusPolicy(Qt.NoFocus)
self.button_update.clicked.connect(self.install)
self.button_launch = ButtonPrimary('Launch Navigator')
self.button_launch.setAutoDefault(True)
self.button_launch.setDisabled(True)
self.button_launch.setFocusPolicy(Qt.NoFocus)
self.button_launch.clicked.connect(self.launch)
layout_buttons: Final[QHBoxLayout] = QHBoxLayout()
layout_buttons.addStretch()
layout_buttons.addWidget(self.button_cancel)
layout_buttons.addSpacing(12)
layout_buttons.addWidget(self.button_update)
layout_buttons.addSpacing(12)
layout_buttons.addWidget(self.button_launch)
layout: Final[QVBoxLayout] = QVBoxLayout()
layout.addLayout(layout_top)
layout.addSpacing(24)
layout.addStretch()
layout.addLayout(layout_buttons)
self.setWindowTitle('Anaconda Navigator Updater')
self.setFixedWidth(450)
self.setMinimumHeight(200)
self.setLayout(layout)
if platform == 'win32':
set_windows_app_user_model_id()
# Timers
self.refresh_countdown: int = 0
self.refresh_timer: Final[QTimer] = QTimer()
self.refresh_timer.setInterval(3_000)
self.refresh_timer.timeout.connect(self.refresh)
# Launch processes
LOGGER.debug('starting processes of the main window')
self.state.fetch_info()
self.refresh()
self.refresh_timer.start()
@staticmethod
def check_running() -> bool:
"""Check if anaconda navigator is running."""
try:
with filelock.FileLock(NAVIGATOR_LOCKFILE).acquire(timeout=0.01): # type: ignore
return False
except filelock.Timeout: # type: ignore
return True
def install(self) -> None:
"""Install latest version of anaconda navigator."""
self.reset_timer()
self.state.install()
self.update()
def launch(self) -> None:
"""Launch anaconda navigator and close current dialog."""
self.state.run()
self.close()
def refresh(self) -> None:
"""Check if anything changed in environment."""
if self.state.active:
return
self.refresh_countdown -= 1
if self.refresh_countdown <= 0:
LOGGER.debug('attempting full refresh')
self.state.find_current()
self.state.find_latest()
self.reset_timer()
else:
LOGGER.debug('attempting quick refresh')
self.update()
def reset_timer(self) -> None:
"""Reset :attr:`~MainWindow.refresh_countdown` to delay full refresh."""
self.refresh_countdown = 10
def update(self, field: str = '') -> None: # pylint: disable=too-many-branches
"""Refresh text of status labels and block/unblock primary buttons."""
if field == 'install':
self.state.find_current()
self.reset_timer()
return
if not self.state.ready:
return
current: Version | Keyword = self.state.current
launchable: bool
message: str = 'Current version: <i>'
if launchable := isinstance(current, Version):
message += escape(str(current))
elif current is Keyword.EMPTY:
message += 'Not installed'
else:
message += '-'
message += '</i><br>Available version: <b>'
latest: Version | Keyword = self.state.latest
updatable: bool
if updatable := isinstance(latest, Version):
message += escape(str(latest))
if isinstance(current, Version):
updatable = latest > current
else:
message += '-'
self.label_message.setText(message + '</b><br>')
if self.state.busy:
if isinstance(self.state.latest, Version):
self.label_status.setText(f'Updating package on <b>{escape(self.state.environment)}</b>...')
else:
self.label_status.setText(f'Installing package on <b>{escape(self.state.environment)}</b>...')
launchable = updatable = False
elif self.check_running():
if updatable:
self.label_status.setText('Please close Anaconda Navigator before updating.')
else:
self.label_status.setText('Anaconda Navigator is up to date and running.')
launchable = updatable = False
elif updatable:
if self.state.read_only and (platform == 'win32'):
self.label_status.setText('Need to run with elevated privileges')
updatable = False
elif self.state.install_failed: # pylint: disable=using-constant-test
self.label_status.setText(
'Failed to install update, please try running "conda install anaconda-navigator" in a terminal',
)
else:
self.label_status.setText('')
else:
self.label_status.setText('Anaconda Navigator is up to date.')
self.button_launch.setEnabled(launchable)
self.button_update.setEnabled(updatable)