workdlg.py

import os
import queue
import signal
import subprocess
import threading
import tkinter as tk
from logging import getLogger
from tkinter import messagebox, ttk
from typing import Optional

from pystart import tktextext
from pystart.languages import tr
from pystart.misc_utils import get_menu_char, running_on_mac_os, running_on_windows
from pystart.ui_utils import (
    CommonDialog,
    create_action_label,
    ems_to_pixels,
    get_style_configuration,
    set_text_if_different,
)

logger = getLogger(__name__)


class WorkDialog(CommonDialog):
    def __init__(self, master, autostart=False):
        super(WorkDialog, self).__init__(master)

        self._autostart = autostart
        self._has_been_started = False
        self._state = "idle"
        self.success = False
        self._work_events_queue = queue.Queue()
        self.init_instructions_frame()
        self.init_main_frame()
        self.init_action_frame()
        self.init_log_frame()
        self.populate_main_frame()
        self.rowconfigure(4, weight=1)  # log frame
        self.columnconfigure(0, weight=1)
        self.title(self.get_title())
        self.stdout = ""
        self.stderr = ""

        self._update_scheduler = None
        self._keep_updating_ui()

        self.bind("<Escape>", self.on_cancel, True)
        self.protocol("WM_DELETE_WINDOW", self.on_cancel)

        if self._autostart:
            self.bind("<Map>", self._start_on_map, True)

    def _start_on_map(self, event) -> None:
        if self._has_been_started:
            return

        self._has_been_started = True
        self.start_work_and_update_ui()

    def populate_main_frame(self):
        pass

    def is_ready_for_work(self):
        return True

    def init_instructions_frame(self):
        instructions = self.get_instructions()
        # Aqua doesn't allow changing ttk.Frame background via theming
        tip_background = get_style_configuration("Tip.TFrame")["background"]
        tip_foreground = get_style_configuration("Tip.TLabel")["foreground"]

        self.instructions_frame = tk.Frame(self, background=tip_background)
        self.instructions_frame.grid(row=0, column=0, sticky="nsew")
        self.instructions_frame.rowconfigure(0, weight=1)
        self.instructions_frame.columnconfigure(0, weight=1)

        pad = self.get_large_padding()
        self.instructions_label = tk.Label(
            self,
            background=tip_background,
            text=instructions,
            justify="left",
            foreground=tip_foreground,
        )
        self.instructions_label.grid(row=0, column=0, sticky="w", padx=pad, pady=pad)

    def get_instructions(self) -> Optional[str]:
        return None

    def init_main_frame(self):
        self.main_frame = ttk.Frame(self)
        self.main_frame.grid(row=1, column=0, sticky="nsew")

    def init_action_frame(self):
        padding = self.get_large_padding()
        intpad = self.get_small_padding()

        self.action_frame = ttk.Frame(self)
        self.action_frame.grid(row=2, column=0, sticky="nsew")

        self._progress_bar = ttk.Progressbar(
            self.action_frame, length=ems_to_pixels(4), mode="indeterminate"
        )

        self._current_action_label = create_action_label(
            self.action_frame,
            text="",
            width=round(self.get_action_text_max_length() * 1.1),
            click_handler=self.toggle_log_frame,
        )
        self._current_action_label.grid(
            row=1, column=2, sticky="we", pady=padding, padx=(0, intpad)
        )

        self._menu_button = ttk.Button(
            self.action_frame,
            text=get_menu_char(),
            command=self.post_action_menu,
            # style="Toolbutton"
            width=3,
        )
        if self.has_action_menu():
            self._menu_button.grid(column=3, row=1, pady=padding, padx=(0, intpad))

        if running_on_mac_os():
            menu_conf = {}
        else:
            menu_conf = get_style_configuration("Menu")
        self._action_menu = tk.Menu(self, tearoff=False, **menu_conf)

        self._ok_button = ttk.Button(
            self.action_frame,
            text=self.get_ok_text(),
            command=self.on_click_ok_button,
            state="disabled",
            default="active",
        )
        if not self._autostart:
            self._ok_button.grid(column=4, row=1, pady=padding, padx=(0, intpad))

        self._cancel_button = ttk.Button(
            self.action_frame,
            text=self.get_cancel_text(),
            command=self.on_cancel,
        )
        self._cancel_button.grid(column=5, row=1, padx=(0, padding), pady=padding)

        self.action_frame.columnconfigure(2, weight=1)

    def on_click_ok_button(self):
        self.start_work_and_update_ui()

    def has_action_menu(self) -> bool:
        return False

    def populate_action_menu(self, action_menu: tk.Menu) -> None:
        pass

    def post_action_menu(self) -> None:
        self._action_menu.delete(0, "end")
        post_x = self._menu_button.winfo_rootx()
        post_y = self._menu_button.winfo_rooty() + self._menu_button.winfo_height()
        self.populate_action_menu(self._action_menu)
        self._action_menu.tk_popup(post_x, post_y)

    def get_action_text_max_length(self):
        return 35

    def get_initial_log_line_count(self):
        return 5

    def init_log_frame(self):
        self.log_frame = ttk.Frame(self)
        self.log_frame.columnconfigure(1, weight=1)
        self.log_frame.rowconfigure(1, weight=1)
        fixed_font = tk.font.nametofont("TkFixedFont")
        font = fixed_font.copy()
        font.configure(size=round(fixed_font.cget("size") * 0.8))
        self.log_text = tktextext.TextFrame(
            self.log_frame,
            horizontal_scrollbar=False,
            wrap="word",
            borderwidth=1,
            height=self.get_initial_log_line_count(),
            width=20,
            font=font,
            read_only=True,
        )

        padding = self.get_large_padding()
        self.log_text.grid(row=1, column=1, sticky="nsew", padx=padding, pady=(0, padding))

    def update_ui(self):
        if self._state == "closed":
            return

        while not self._work_events_queue.empty():
            self.handle_work_event(*self._work_events_queue.get())
            if self._state == "closed":
                return

        if self._state == "idle":
            if self.is_ready_for_work():
                self._ok_button.configure(state="normal")
            else:
                self._ok_button.configure(state="disabled")
        else:
            self._ok_button.configure(state="disabled")

        if self._state == "done":
            set_text_if_different(self._cancel_button, tr("Close"))
        else:
            set_text_if_different(self._cancel_button, tr("Cancel"))

    def start_work(self):
        pass

    def get_title(self):
        return "Work dialog"

    def _keep_updating_ui(self):
        if self._state != "closed":
            if self.winfo_ismapped():
                self.update_ui()
            self._update_scheduler = self.after(200, self._keep_updating_ui)
        else:
            self._update_scheduler = None

    def close(self):
        self._state = "closed"
        if self._update_scheduler is not None:
            try:
                self.after_cancel(self._update_scheduler)
            except tk.TclError:
                pass

        self.destroy()

    def cancel_work(self):
        # worker should periodically check this value
        self._state = "cancelling"
        self.set_action_text(tr("Cancelling"))

    def toggle_log_frame(self, event=None):
        if self.log_frame.winfo_ismapped():
            self.hide_log_frame()
        else:
            self.show_log_frame()

    def show_log_frame(self):
        self.log_frame.grid(row=4, column=0, sticky="nsew")
        self.rowconfigure(2, weight=0)
        self.rowconfigure(4, weight=1)

    def hide_log_frame(self):
        self.log_frame.grid_forget()
        self.rowconfigure(2, weight=1)
        self.rowconfigure(4, weight=0)

    def clear_log(self) -> None:
        self.log_text.text.direct_delete("1.0", "end")

    def get_ok_text(self):
        return tr("OK")

    def get_cancel_text(self):
        return tr("Cancel")

    def start_work_and_update_ui(self, event=None):
        assert self._state == "idle"
        if self.start_work() is not False:
            self._state = "working"
            self.success = False
            self.grid_progress_widgets()
            self._progress_bar["mode"] = "indeterminate"
            self._progress_bar.start()
            if not self._current_action_label["text"]:
                self._current_action_label["text"] = tr("Starting") + "..."

    def grid_progress_widgets(self):
        padding = self.get_large_padding()
        intpad = self.get_small_padding()
        self._progress_bar.grid(row=1, column=1, sticky="w", padx=(padding, intpad), pady=padding)

    def on_cancel(self, event=None):
        if self._state in ("idle", "done"):
            self.close()
        elif self._state == "cancelling" and self.confirm_leaving_while_cancelling():
            self.close()
        elif self.confirm_cancel():
            self.cancel_work()

    def confirm_leaving_while_cancelling(self):
        return messagebox.askyesno(
            "Close dialog?",
            "Cancelling is in progress.\nDo you still want to close the dialog?",
            parent=self,
        )

    def confirm_cancel(self):
        return messagebox.askyesno(
            "Cancel work?",
            "Are you sure you want to cancel?",
            parent=self,
        )

    def append_text(self, text: str, stream_name="stdout") -> None:
        """Appends text to the details box. May be called from another thread."""
        self._work_events_queue.put(("append", (text, stream_name)))
        setattr(self, stream_name, getattr(self, stream_name) + text)

    def replace_last_line(self, text: str, stream_name="stdout") -> None:
        """Replaces last line in the details box. May be called from another thread."""
        self._work_events_queue.put(("replace", (text, stream_name)))
        setattr(self, stream_name, getattr(self, stream_name) + text)

    def report_progress(self, value: Optional[float], maximum: Optional[float]) -> None:
        """Updates progress bar. May be called from another thread."""
        self._work_events_queue.put(("progress", (value, maximum)))

    def set_action_text(self, text: str) -> None:
        """Updates text above the progress bar. May be called from another thread."""
        self._work_events_queue.put(("action", (text,)))

    def set_action_text_smart(self, text: str) -> None:
        """Updates text above the progress bar. May be called from another thread."""
        text = text.strip()
        if not text:
            return
        if len(text) > self.get_action_text_max_length():
            text = text[: self.get_action_text_max_length() - 3] + "..."
        self.set_action_text(text)

    def report_done(self, success):
        """May be called from another thread."""
        self._work_events_queue.put(("done", (success,)))

    def handle_work_event(self, type, args):
        if type in ("append", "replace"):
            text, stream_name = args
            if type == "replace":
                self.log_text.text.direct_delete("end-1c linestart", "end-1c")
            self.log_text.text.direct_insert("end", text, (stream_name,))
            self.log_text.text.see("end")
        elif type == "action":
            set_text_if_different(self._current_action_label, args[0])
        elif type == "progress":
            value, maximum = args
            if value is None or maximum is None:
                if self._progress_bar["mode"] != "indeterminate":
                    self._progress_bar["mode"] = "indeterminate"
                    self._progress_bar.start()
            else:
                if self._progress_bar["mode"] != "determinate":
                    self._progress_bar["mode"] = "determinate"
                    self._progress_bar.stop()
                self._progress_bar.configure(value=value, maximum=maximum)
        elif type == "done":
            self.on_done(args[0])

    def allow_single_success(self) -> bool:
        return True

    def on_done(self, success):
        """NB! Don't call from non-ui thread!"""
        self.success = success
        if self.success and self.allow_single_success() or self._autostart:
            self._state = "done"
            self._cancel_button.focus_set()
            self._cancel_button["default"] = "active"
            self._ok_button["default"] = "normal"
        else:
            # allows trying again when failed or wasn't final action
            self.allow_new_work()

        self._progress_bar.stop()
        # need to put to determinate mode, otherwise it looks half done
        self._progress_bar["mode"] = "determinate"
        if self.success and self._autostart and not self.log_frame.winfo_ismapped():
            self.close()

        if not self.success:
            self.show_log_frame()

    def allow_new_work(self):
        self._state = "idle"
        self._ok_button.focus_set()
        self._ok_button["default"] = "active"
        self._cancel_button["default"] = "normal"


class SubprocessDialog(WorkDialog):
    """Shows incrementally the output of given subprocess.
    Allows cancelling"""

    def __init__(
        self, master, prepared_proc=None, title="Subprocess", long_description=None, autostart=True
    ):
        self._proc = None
        self._prepared_proc = prepared_proc
        self.stdout = ""
        self.stderr = ""
        self._stdout_thread = None
        self._stderr_thread = None
        self._title = title
        self._long_description = long_description
        self.returncode = None

        super().__init__(master, autostart=autostart)

    def is_ready_for_work(self):
        return True

    def get_title(self):
        return self._title

    def get_instructions(self) -> Optional[str]:
        return self._long_description

    def start_subprocess(self):
        if self._prepared_proc:
            return self._prepared_proc
        else:
            raise RuntimeError("No process provided")

    def start_work(self):
        self._proc = self.start_subprocess()
        if hasattr(self._proc, "cmd"):
            try:
                self.append_text(subprocess.list2cmdline(self._proc.cmd) + "\n")
            except:
                logger.warning("Could not extract cmd (%s)", self._proc.cmd)
        self._start_listening_current_proc()

    def _start_listening_current_proc(self):
        def listen_stream(stream_name):
            stream = getattr(self._proc, stream_name)
            while True:
                data = stream.readline()
                self.append_text(data, stream_name)
                self._check_set_action_text_from_output_line(data)
                setattr(self, stream_name, getattr(self, stream_name) + data)
                if data == "":
                    logger.debug("Finished reading %s", stream_name)
                    break

            if stream_name == "stdout":
                self._finish_process()

            logger.debug("Returning from reading %s", stream_name)

        self._stdout_thread = threading.Thread(target=listen_stream, args=["stdout"], daemon=True)
        self._stdout_thread.start()
        if self._proc.stderr is not None:
            self._stderr_thread = threading.Thread(
                target=listen_stream, args=["stderr"], daemon=True
            )
            self._stderr_thread.start()

    def _finish_process(self):
        self.returncode = self._proc.wait()
        logger.debug("Process ended with returncode %s", self.returncode)
        if self.returncode:
            self.set_action_text("Error")
            self.append_text("Error: process returned with code %s\n" % self.returncode)
        else:
            self.set_action_text("Done!")
            self.append_text("Done!")

        self.report_done(self.returncode == 0)

    def get_action_text_max_length(self):
        return 35

    def _check_set_action_text_from_output_line(self, line):
        if len(line) > self.get_action_text_max_length():
            line = line[: self.get_action_text_max_length() - 3].strip() + "..."
        if line:
            self.set_action_text(line.strip())

    def cancel_work(self):
        super().cancel_work()
        # try gently first
        try:
            try:
                if running_on_windows():
                    os.kill(self._proc.pid, signal.CTRL_BREAK_EVENT)  # pylint: disable=no-member
                else:
                    os.kill(self._proc.pid, signal.SIGINT)

                self._proc.wait(2)
            except subprocess.TimeoutExpired:
                if self._proc.poll() is None:
                    # now let's be more concrete
                    self._proc.kill()
        except OSError as e:
            messagebox.showerror("Error", "Could not kill subprocess: " + str(e), master=self)
            logger.error("Could not kill subprocess", exc_info=e)