models.py

# -*- coding: utf-8 -*-
#
# Copyright © Spyder Project Contributors
# Licensed under the terms of the MIT License
# (see spyder/__init__.py for details)

"""Spyder run models."""

# Standard library imports
from collections import OrderedDict
from typing import Dict, Tuple, Optional, Set, Union, List

# Third-party imports
from qtpy.QtCore import (Qt, Signal, QAbstractListModel, QModelIndex,
                         QAbstractTableModel)

# Local imports
from spyder.api.translations import _
from spyder.plugins.run.api import (
    StoredRunExecutorParameters, RunContext, RunConfigurationMetadata,
    SupportedExecutionRunConfiguration, RunParameterFlags,
    StoredRunConfigurationExecutor, ExtendedRunExecutionParameters,
    RunExecutionParameters, WorkingDirOpts, WorkingDirSource)


class RunExecutorListModel(QAbstractListModel):
    sig_executor_widget_changed = Signal(object)

    def __init__(self, parent):
        super().__init__(parent)
        self.parent = parent
        self.uuid: Optional[str] = None
        self.current_input: Optional[Tuple[str, str]] = None
        self.executor_names: Dict[str, str] = {}
        self.executor_configurations: Dict[Tuple[str, str], Dict[
            str, SupportedExecutionRunConfiguration]] = {}
        self.executors_per_input: Dict[Tuple[str, str], Dict[str, int]] = {}
        self.inverted_pos: Dict[Tuple[str, str], Dict[int, str]] = {}
        self.executor_priority: Dict[Tuple[str, str], Dict[str, int]] = {}

    def set_executor_name(self, executor_id: str, executor_name: str):
        self.executor_names[executor_id] = executor_name

    def add_input_executor_configuration(
        self,
        ext: str,
        context_id: str,
        executor_id: str,
        config: SupportedExecutionRunConfiguration
    ):
        executor_conf = self.executor_configurations.get((ext, context_id), {})
        executor_conf[executor_id] = config
        self.executor_configurations[(ext, context_id)] = executor_conf

        input_executors = self.executors_per_input.get((ext, context_id), {})
        all_exec_prio = self.executor_priority.get((ext, context_id), {})
        priority = config['priority']

        # Remove if existing
        input_executors.pop(executor_id, None)
        all_exec_prio.pop(executor_id, None)
        all_exec_prio[executor_id] = priority

        input_values = list(all_exec_prio.items())
        input_values = sorted(input_values, key=lambda k: k[1])

        input_values = [(x, i) for i, (x, _) in enumerate(input_values)]
        self.executors_per_input[(ext, context_id)] = dict(input_values)

        self.inverted_pos[(ext, context_id)] = {
            v: k for (k, v) in input_values}
        self.executor_priority[(ext, context_id)] = all_exec_prio

    def remove_input_executor_configuration(
        self,
        ext: str,
        context_id: str,
        executor_id: str
    ):
        input_executors = self.executors_per_input[(ext, context_id)]
        pos = input_executors.pop(executor_id)

        inverted_executors = self.inverted_pos[(ext, context_id)]
        inverted_executors.pop(pos)

    def switch_input(self, uuid: str, run_input: Tuple[str, str]):
        if run_input in self.executors_per_input:
            self.beginResetModel()
            self.current_input = run_input
            self.uuid = uuid
            self.endResetModel()
        else:
            raise ValueError(
                f'The requested run input combination {run_input} is not '
                f'registered'
            )

    def get_selected_run_executor(
        self,
        index: int
    ) -> Tuple[str, SupportedExecutionRunConfiguration]:

        executor_indices = self.inverted_pos[self.current_input]
        executor_name = executor_indices[index]
        executors = self.executor_configurations[self.current_input]
        executor = executors[executor_name]

        return executor_name, executor

    def get_run_executor_index(self, executor_name: str) -> int:
        ordered_executors = self.executors_per_input[self.current_input]
        executor_index = ordered_executors[executor_name]
        return executor_index

    def get_initial_index(self) -> int:
        last_executor = self.parent.get_last_used_executor_parameters(
            self.uuid)
        executor_id = last_executor['executor']
        pos = 0

        if executor_id is not None:
            executors = self.executors_per_input[self.current_input]
            pos = executors[executor_id]

        return pos

    def get_default_executor(self, input: Tuple[str, str]) -> str:
        executors = self.inverted_pos[input]
        return executors[0]

    def executor_supports_configuration(
        self,
        executor: str,
        exec_input: Tuple[str, str]
    ) -> bool:

        input_executors = self.executor_configurations.get(exec_input, {})
        return executor in input_executors

    def data(self, index: QModelIndex, role: int = Qt.DisplayRole):
        if role == Qt.DisplayRole:
            executor_indices = self.inverted_pos[self.current_input]
            executor_id = executor_indices[index.row()]
            return self.executor_names[executor_id]

    def rowCount(self, parent: QModelIndex = None) -> int:
        executors = self.executors_per_input.get(self.current_input, {})
        return len(executors)

    def __contains__(self, exec_input: Tuple[str, str]) -> bool:
        return exec_input in self.executor_configurations

    def __getitem__(
            self, input_executor: tuple) -> SupportedExecutionRunConfiguration:
        (input, executor) = input_executor
        input_executors = self.executor_configurations[input]
        return input_executors[executor]


class RunConfigurationListModel(QAbstractListModel):
    def __init__(self, parent, executor_model):
        super().__init__(parent)
        self.parent = parent
        self.selected_metadata: Optional[RunConfigurationMetadata] = None
        self.current_configuration: Optional[str] = None
        self.metadata_index: Dict[int, str] = {}
        self.inverted_index: Dict[str, int] = {}
        self.run_configurations: OrderedDict[
            str, RunConfigurationMetadata] = OrderedDict()
        self.executor_model: RunExecutorListModel = executor_model

    def get_metadata_context_extension(self, uuid: str):
        run_conf = self.run_configurations[uuid]
        return run_conf['context'], run_conf['input_extension']

    def set_current_run_configuration(self, uuid: str):
        self.current_configuration = uuid

    def get_initial_index(self) -> int:
        return self.inverted_index[self.current_configuration]

    def get_selected_metadata(self) -> Optional[RunConfigurationMetadata]:
        return self.selected_metadata

    def get_metadata(self, index: int) -> RunConfigurationMetadata:
        uuid = self.metadata_index[index]
        metadata = self.run_configurations[uuid]
        return metadata

    def update_index(self, index: int):
        uuid = self.metadata_index[index]
        metadata = self.run_configurations[uuid]
        context_name = metadata['context']['name']
        context_id = getattr(RunContext, context_name)
        ext = metadata['input_extension']
        self.selected_metadata = metadata
        self.executor_model.switch_input(uuid, (ext, context_id))

    def data(self, index: QModelIndex, role: int = Qt.DisplayRole):
        if role == Qt.DisplayRole:
            uuid = self.metadata_index[index.row()]
            metadata = self.run_configurations[uuid]
            return metadata['name']

    def rowCount(self, parent: QModelIndex = None) -> int:
        return len(self.run_configurations)

    def get_run_configuration_parameters(
        self,
        uuid: str,
        executor: str
    ) -> Optional[StoredRunExecutorParameters]:

        context, ext = self.get_metadata_context_extension(uuid)
        context_name = context['name']
        context_id = getattr(RunContext, context_name)

        return self.parent.get_executor_configuration_parameters(
            executor, ext, context_id)

    def get_last_used_execution_params(
        self,
        uuid: str,
        executor: str
    ) -> Optional[str]:

        return self.parent.get_last_used_execution_params(uuid, executor)

    def get_last_used_executor_parameters(
        self,
        uuid: str
    ) -> StoredRunConfigurationExecutor:

        return self.parent.get_last_used_executor_parameters(uuid)

    def pop(self, uuid: str) -> RunConfigurationMetadata:
        item = self.run_configurations.pop(uuid)
        self.metadata_index = dict(enumerate(self.run_configurations))
        self.inverted_index = {v: k for k, v in self.metadata_index.items()}
        self.dataChanged.emit(self.createIndex(0, 0),
                              self.createIndex(len(self.metadata_index), 0))
        return item

    def __iter__(self):
        return iter(self.run_configurations)

    def __len__(self):
        return len(self.run_configurations)

    def __getitem__(self, uuid: str) -> RunConfigurationMetadata:
        return self.run_configurations[uuid]

    def __setitem__(self, uuid: str, metadata: RunConfigurationMetadata):
        self.run_configurations[uuid] = metadata
        self.metadata_index[len(self.metadata_index)] = uuid
        self.inverted_index[uuid] = len(self.inverted_index)
        self.dataChanged.emit(self.createIndex(0, 0),
                              self.createIndex(len(self.metadata_index), 0))

    def __contains__(self, uuid: str):
        return uuid in self.run_configurations


class RunExecutorParameters(QAbstractListModel):
    def __init__(self, parent):
        super().__init__(parent)
        self.parent = parent
        self.executor_conf_params: Dict[
            str, ExtendedRunExecutionParameters] = {}
        self.params_index: Dict[int, str] = {}
        self.inverse_index: Dict[str, int] = {}

    def data(self, index: QModelIndex, role: int = Qt.DisplayRole):
        pos = index.row()
        total_saved_params = len(self.executor_conf_params)
        if pos == total_saved_params:
            if role == Qt.DisplayRole:
                return _("Default/Transient")
            elif role == Qt.ToolTipRole:
                return _(
                    "This configuration will not be saved after execution"
                )
        else:
            params_id = self.params_index[pos]
            params = self.executor_conf_params[params_id]
            params_name = params['name']
            if role == Qt.DisplayRole:
                return params_name

    def rowCount(self, parent: QModelIndex = ...) -> int:
        return len(self.executor_conf_params) + 1

    def get_executor_parameters(
        self,
        index: int
    ) -> Tuple[RunParameterFlags, RunExecutionParameters]:

        if index == len(self) - 1:
            default_working_dir = WorkingDirOpts(
                source=WorkingDirSource.ConfigurationDirectory,
                path=None)
            default_params = RunExecutionParameters(
                working_dir=default_working_dir,
                executor_params={})

            return RunParameterFlags.SetDefaults, default_params
        else:
            params_id = self.params_index[index]
            params = self.executor_conf_params[params_id]
            actual_params = params['params']

            return RunParameterFlags.SwitchValues, actual_params

    def get_parameters_uuid_name(
        self,
        index: int
    ) -> Tuple[Optional[str], Optional[str]]:

        if index == len(self) - 1:
            return None, None

        params_id = self.params_index[index]
        params = self.executor_conf_params[params_id]
        return params['uuid'], params['name']

    def set_parameters(
        self,
        parameters: Dict[str, ExtendedRunExecutionParameters]
    ):
        self.beginResetModel()
        self.executor_conf_params = parameters
        self.params_index = dict(enumerate(self.executor_conf_params))
        self.inverse_index = {self.params_index[k]: k
                              for k in self.params_index}
        self.endResetModel()

    def get_parameters_index(self, parameters_name: Optional[str]) -> int:
        index = self.inverse_index.get(parameters_name,
                                       len(self.executor_conf_params))
        return index

    def __len__(self) -> int:
        return len(self.executor_conf_params) + 1


class RunExecutorNamesListModel(QAbstractListModel):
    def __init__(self, parent, executor_model: RunExecutorListModel):
        super().__init__(parent)
        self.executor_model = executor_model
        self.executor_configurations: OrderedDict[
            str, Set[Tuple[str, str]]] = OrderedDict({})

        for input_conf in self.executor_model.executors_per_input:
            executors_available = self.executor_model.executors_per_input[
                input_conf]
            for executor in executors_available:
                exec_input_conf = self.executor_model[(input_conf, executor)]
                ConfigWidget = exec_input_conf.get(
                    'configuration_widget', None)
                if ConfigWidget is not None:
                    input_set = self.executor_configurations.get(
                        executor,set({}))
                    input_set |= {input_conf}
                    self.executor_configurations[executor] = input_set

        self.executor_indexed_list: Dict[int, str] = dict(
            enumerate(self.executor_configurations))

        self.executor_list: Dict[str, int] = {
            v: k for k, v in self.executor_indexed_list.items()}

    def data(self, index: QModelIndex, role: int = Qt.DisplayRole) -> str:
        row = index.row()
        if role == Qt.DisplayRole:
            executor_id = self.executor_indexed_list[row]
            return self.executor_model.executor_names[executor_id]

    def rowCount(self, parent: QModelIndex = None) -> int:
        return len(self.executor_list)

    def selected_executor(
            self, index: int) -> Tuple[str, Set[Tuple[str, str]]]:
        executor_name = self.executor_indexed_list[index]
        return executor_name, self.executor_configurations[executor_name]


class ExecutorRunParametersTableModel(QAbstractTableModel):
    EXTENSION = 0
    CONTEXT = 1
    NAME = 2

    sig_data_changed = Signal()

    def __init__(self, parent):
        super().__init__(parent)
        self.executor_conf_params: Dict[
            Tuple[str, str, str], ExtendedRunExecutionParameters] = {}

        self.params_index: Dict[int, Tuple[str, str, str]] = {}
        self.inverse_index: Dict[Tuple[str, str, str], int] = {}

    def data(self, index: QModelIndex,
             role: int = Qt.DisplayRole) -> Union[str, int]:
        row = index.row()
        column = index.column()

        params_idx = self.params_index[row]
        (extension, context, __) = params_idx
        params = self.executor_conf_params[params_idx]

        if role == Qt.DisplayRole:
            if column == self.EXTENSION:
                return extension
            elif column == self.CONTEXT:
                return context
            elif column == self.NAME:
                return params['name']
        elif role == Qt.TextAlignmentRole:
            return int(Qt.AlignHCenter | Qt.AlignVCenter)
        elif role == Qt.ToolTipRole:
            return _("Double-click to view or edit")

    def headerData(
        self,
        section: int,
        orientation: Qt.Orientation,
        role: int = Qt.DisplayRole
    ) -> Union[str, int]:

        if role == Qt.TextAlignmentRole:
            if orientation == Qt.Horizontal:
                return int(Qt.AlignHCenter | Qt.AlignVCenter)
            return int(Qt.AlignRight | Qt.AlignVCenter)

        if role == Qt.DisplayRole:
            if orientation == Qt.Horizontal:
                if section == self.EXTENSION:
                    return _('File extension')
                elif section == self.CONTEXT:
                    return _('Context name')
                elif section == self.NAME:
                    return _('Run parameters name')

    def rowCount(self, parent: QModelIndex = None) -> int:
        return len(self.params_index)

    def columnCount(self, parent: QModelIndex = None) -> int:
        return 3

    def set_parameters(
        self,
        params: Dict[Tuple[str, str, str], ExtendedRunExecutionParameters]
    ):
        self.beginResetModel()
        self.executor_conf_params = params
        self.params_index = dict(enumerate(params))
        self.inverse_index = {v: k for k, v in self.params_index.items()}
        self.endResetModel()

    def get_current_view(
        self
    ) -> Dict[Tuple[str, str, str], ExtendedRunExecutionParameters]:

        return self.executor_conf_params

    def get_tuple_index(self, index: int) -> Tuple[str, str, str]:
        return self.params_index[index]

    def reset_model(self):
        self.beginResetModel()
        self.endResetModel()

    def get_parameters_name(self, idx: Tuple[str, str, str]) -> str:
        params = self.executor_conf_params[idx]
        return params['name']

    def sort(self, column: int, order: Qt.SortOrder = ...) -> None:
        self.beginResetModel()
        reverse = order == Qt.DescendingOrder

        if column == self.EXTENSION:
            sorted_keys = sorted(self.executor_conf_params,
                                 key=lambda x: x[0],
                                 reverse=reverse)
        elif column == self.CONTEXT:
            sorted_keys = sorted(self.executor_conf_params,
                                 key=lambda x: x[1],
                                 reverse=reverse)
        elif column == self.NAME:
            sorted_keys = sorted(
                self.executor_conf_params,
                key=lambda x: self.executor_conf_params[x]['name'],
                reverse=reverse)

        self.params_index = dict(enumerate(sorted_keys))
        self.inverse_index = {v: k for k, v in self.params_index.items()}
        self.endResetModel()

    def apply_changes(
        self,
        changes: List[Tuple[str, ExtendedRunExecutionParameters]],
        extension: str,
        context: str
    ):
        self.beginResetModel()

        for (operation, params) in changes:
            key = (extension, context, params['uuid'])
            if operation == 'deleted':
                self.executor_conf_params.pop(key)
            else:
                self.executor_conf_params[key] = params

        self.params_index = dict(enumerate(self.executor_conf_params))
        self.inverse_index = {v: k for k, v in self.params_index.items()}
        self.endResetModel()
        self.sig_data_changed.emit()

    def __len__(self):
        return len(self.inverse_index)

    def __iter__(self):
        return iter(self.executor_conf_params)

    def __getitem__(
        self,
        index: int
    ) -> Tuple[str, str, ExtendedRunExecutionParameters]:

        tuple_index = self.params_index[index]
        (ext, context, _) = tuple_index
        return (ext, context, self.executor_conf_params[tuple_index])