core-extra/daemon/core/gui/dialogs/serviceconfig.py

611 lines
25 KiB
Python

import logging
import tkinter as tk
from pathlib import Path
from tkinter import filedialog, ttk
from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple
import grpc
from PIL.ImageTk import PhotoImage
from core.api.grpc.wrappers import Node, NodeServiceData, ServiceValidationMode
from core.gui import images
from core.gui.dialogs.copyserviceconfig import CopyServiceConfigDialog
from core.gui.dialogs.dialog import Dialog
from core.gui.images import ImageEnum
from core.gui.themes import FRAME_PAD, PADX, PADY
from core.gui.widgets import CodeText, ListboxScroll
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from core.gui.app import Application
from core.gui.coreclient import CoreClient
ICON_SIZE: int = 16
class ServiceConfigDialog(Dialog):
def __init__(
self, master: tk.BaseWidget, app: "Application", service_name: str, node: Node
) -> None:
title = f"{service_name} Service"
super().__init__(app, title, master=master)
self.core: "CoreClient" = app.core
self.node: Node = node
self.service_name: str = service_name
self.radiovar: tk.IntVar = tk.IntVar(value=2)
self.metadata: str = ""
self.filenames: List[str] = []
self.dependencies: List[str] = []
self.executables: List[str] = []
self.startup_commands: List[str] = []
self.validation_commands: List[str] = []
self.shutdown_commands: List[str] = []
self.default_startup: List[str] = []
self.default_validate: List[str] = []
self.default_shutdown: List[str] = []
self.validation_mode: Optional[ServiceValidationMode] = None
self.validation_time: Optional[int] = None
self.validation_period: Optional[float] = None
self.directory_entry: Optional[ttk.Entry] = None
self.default_directories: List[str] = []
self.temp_directories: List[str] = []
self.documentnew_img: PhotoImage = self.app.get_enum_icon(
ImageEnum.DOCUMENTNEW, width=ICON_SIZE
)
self.editdelete_img: PhotoImage = self.app.get_enum_icon(
ImageEnum.EDITDELETE, width=ICON_SIZE
)
self.notebook: Optional[ttk.Notebook] = None
self.metadata_entry: Optional[ttk.Entry] = None
self.filename_combobox: Optional[ttk.Combobox] = None
self.dir_list: Optional[ListboxScroll] = None
self.startup_commands_listbox: Optional[tk.Listbox] = None
self.shutdown_commands_listbox: Optional[tk.Listbox] = None
self.validate_commands_listbox: Optional[tk.Listbox] = None
self.validation_time_entry: Optional[ttk.Entry] = None
self.validation_mode_entry: Optional[ttk.Entry] = None
self.service_file_data: Optional[CodeText] = None
self.validation_period_entry: Optional[ttk.Entry] = None
self.original_service_files: Dict[str, str] = {}
self.default_config: Optional[NodeServiceData] = None
self.temp_service_files: Dict[str, str] = {}
self.modified_files: Set[str] = set()
self.has_error: bool = False
self.load()
if not self.has_error:
self.draw()
def load(self) -> None:
try:
self.core.start_session(definition=True)
default_config = self.app.core.get_node_service(
self.node.id, self.service_name
)
self.default_startup = default_config.startup[:]
self.default_validate = default_config.validate[:]
self.default_shutdown = default_config.shutdown[:]
self.default_directories = default_config.dirs[:]
custom_service_config = self.node.service_configs.get(self.service_name)
self.default_config = default_config
service_config = (
custom_service_config if custom_service_config else default_config
)
self.dependencies = service_config.dependencies[:]
self.executables = service_config.executables[:]
self.metadata = service_config.meta
self.filenames = service_config.configs[:]
self.startup_commands = service_config.startup[:]
self.validation_commands = service_config.validate[:]
self.shutdown_commands = service_config.shutdown[:]
self.validation_mode = service_config.validation_mode
self.validation_time = service_config.validation_timer
self.temp_directories = service_config.dirs[:]
self.original_service_files = {
x: self.app.core.get_node_service_file(
self.node.id, self.service_name, x
)
for x in default_config.configs
}
self.temp_service_files = dict(self.original_service_files)
file_configs = self.node.service_file_configs.get(self.service_name, {})
for file, data in file_configs.items():
self.temp_service_files[file] = data
except grpc.RpcError as e:
self.app.show_grpc_exception("Get Node Service Error", e)
self.has_error = True
def draw(self) -> None:
self.top.columnconfigure(0, weight=1)
self.top.rowconfigure(1, weight=1)
# draw metadata
frame = ttk.Frame(self.top)
frame.grid(sticky=tk.EW, pady=PADY)
frame.columnconfigure(1, weight=1)
label = ttk.Label(frame, text="Meta-data")
label.grid(row=0, column=0, sticky=tk.W, padx=PADX)
self.metadata_entry = ttk.Entry(frame, textvariable=self.metadata)
self.metadata_entry.grid(row=0, column=1, sticky=tk.EW)
# draw notebook
self.notebook = ttk.Notebook(self.top)
self.notebook.grid(sticky=tk.NSEW, pady=PADY)
self.draw_tab_files()
self.draw_tab_directories()
self.draw_tab_startstop()
self.draw_tab_configuration()
self.draw_buttons()
def draw_tab_files(self) -> None:
tab = ttk.Frame(self.notebook, padding=FRAME_PAD)
tab.grid(sticky=tk.NSEW)
tab.columnconfigure(0, weight=1)
self.notebook.add(tab, text="Files")
label = ttk.Label(
tab, text="Config files and scripts that are generated for this service."
)
label.grid()
frame = ttk.Frame(tab)
frame.grid(sticky=tk.EW, pady=PADY)
frame.columnconfigure(1, weight=1)
label = ttk.Label(frame, text="File Name")
label.grid(row=0, column=0, padx=PADX, sticky=tk.W)
self.filename_combobox = ttk.Combobox(frame, values=self.filenames)
self.filename_combobox.bind(
"<<ComboboxSelected>>", self.display_service_file_data
)
self.filename_combobox.grid(row=0, column=1, sticky=tk.EW, padx=PADX)
button = ttk.Button(
frame, image=self.documentnew_img, command=self.add_filename
)
button.grid(row=0, column=2, padx=PADX)
button = ttk.Button(
frame, image=self.editdelete_img, command=self.delete_filename
)
button.grid(row=0, column=3)
frame = ttk.Frame(tab)
frame.grid(sticky=tk.EW, pady=PADY)
frame.columnconfigure(1, weight=1)
button = ttk.Radiobutton(
frame,
variable=self.radiovar,
text="Copy Source File",
value=1,
state=tk.DISABLED,
)
button.grid(row=0, column=0, sticky=tk.W, padx=PADX)
entry = ttk.Entry(frame, state=tk.DISABLED)
entry.grid(row=0, column=1, sticky=tk.EW, padx=PADX)
image = images.from_enum(ImageEnum.FILEOPEN, width=images.BUTTON_SIZE)
button = ttk.Button(frame, image=image)
button.image = image
button.grid(row=0, column=2)
frame = ttk.Frame(tab)
frame.grid(sticky=tk.EW, pady=PADY)
frame.columnconfigure(0, weight=1)
button = ttk.Radiobutton(
frame,
variable=self.radiovar,
text="Use text below for file contents",
value=2,
)
button.grid(row=0, column=0, sticky=tk.EW)
image = images.from_enum(ImageEnum.FILEOPEN, width=images.BUTTON_SIZE)
button = ttk.Button(frame, image=image)
button.image = image
button.grid(row=0, column=1)
image = images.from_enum(ImageEnum.DOCUMENTSAVE, width=images.BUTTON_SIZE)
button = ttk.Button(frame, image=image)
button.image = image
button.grid(row=0, column=2)
self.service_file_data = CodeText(tab)
self.service_file_data.grid(sticky=tk.NSEW)
tab.rowconfigure(self.service_file_data.grid_info()["row"], weight=1)
if len(self.filenames) > 0:
self.filename_combobox.current(0)
self.service_file_data.text.delete(1.0, "end")
self.service_file_data.text.insert(
"end", self.temp_service_files[self.filenames[0]]
)
self.service_file_data.text.bind(
"<FocusOut>", self.update_temp_service_file_data
)
def draw_tab_directories(self) -> None:
tab = ttk.Frame(self.notebook, padding=FRAME_PAD)
tab.grid(sticky=tk.NSEW)
tab.columnconfigure(0, weight=1)
tab.rowconfigure(2, weight=1)
self.notebook.add(tab, text="Directories")
label = ttk.Label(
tab,
text="Directories required by this service that are unique for each node.",
)
label.grid(row=0, column=0, sticky=tk.EW)
frame = ttk.Frame(tab, padding=FRAME_PAD)
frame.columnconfigure(0, weight=1)
frame.grid(row=1, column=0, sticky=tk.NSEW)
var = tk.StringVar(value="")
self.directory_entry = ttk.Entry(frame, textvariable=var)
self.directory_entry.grid(row=0, column=0, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, text="...", command=self.find_directory_button)
button.grid(row=0, column=1, sticky=tk.EW)
self.dir_list = ListboxScroll(tab)
self.dir_list.grid(row=2, column=0, sticky=tk.NSEW, pady=PADY)
self.dir_list.listbox.bind("<<ListboxSelect>>", self.directory_select)
for d in self.temp_directories:
self.dir_list.listbox.insert("end", d)
frame = ttk.Frame(tab)
frame.grid(row=3, column=0, sticky=tk.NSEW)
frame.columnconfigure(0, weight=1)
frame.columnconfigure(1, weight=1)
button = ttk.Button(frame, text="Add", command=self.add_directory)
button.grid(row=0, column=0, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, text="Remove", command=self.remove_directory)
button.grid(row=0, column=1, sticky=tk.EW)
def draw_tab_startstop(self) -> None:
tab = ttk.Frame(self.notebook, padding=FRAME_PAD)
tab.grid(sticky=tk.NSEW)
tab.columnconfigure(0, weight=1)
for i in range(3):
tab.rowconfigure(i, weight=1)
self.notebook.add(tab, text="Startup/Shutdown")
commands = []
# tab 3
for i in range(3):
label_frame = None
if i == 0:
label_frame = ttk.LabelFrame(
tab, text="Startup Commands", padding=FRAME_PAD
)
commands = self.startup_commands
elif i == 1:
label_frame = ttk.LabelFrame(
tab, text="Shutdown Commands", padding=FRAME_PAD
)
commands = self.shutdown_commands
elif i == 2:
label_frame = ttk.LabelFrame(
tab, text="Validation Commands", padding=FRAME_PAD
)
commands = self.validation_commands
label_frame.columnconfigure(0, weight=1)
label_frame.rowconfigure(1, weight=1)
label_frame.grid(row=i, column=0, sticky=tk.NSEW, pady=PADY)
frame = ttk.Frame(label_frame)
frame.grid(row=0, column=0, sticky=tk.NSEW, pady=PADY)
frame.columnconfigure(0, weight=1)
entry = ttk.Entry(frame, textvariable=tk.StringVar())
entry.grid(row=0, column=0, stick="ew", padx=PADX)
button = ttk.Button(frame, image=self.documentnew_img)
button.bind("<Button-1>", self.add_command)
button.grid(row=0, column=1, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, image=self.editdelete_img)
button.grid(row=0, column=2, sticky=tk.EW)
button.bind("<Button-1>", self.delete_command)
listbox_scroll = ListboxScroll(label_frame)
listbox_scroll.listbox.bind("<<ListboxSelect>>", self.update_entry)
for command in commands:
listbox_scroll.listbox.insert("end", command)
listbox_scroll.listbox.config(height=4)
listbox_scroll.grid(row=1, column=0, sticky=tk.NSEW)
if i == 0:
self.startup_commands_listbox = listbox_scroll.listbox
elif i == 1:
self.shutdown_commands_listbox = listbox_scroll.listbox
elif i == 2:
self.validate_commands_listbox = listbox_scroll.listbox
def draw_tab_configuration(self) -> None:
tab = ttk.Frame(self.notebook, padding=FRAME_PAD)
tab.grid(sticky=tk.NSEW)
tab.columnconfigure(0, weight=1)
self.notebook.add(tab, text="Configuration", sticky=tk.NSEW)
frame = ttk.Frame(tab)
frame.grid(sticky=tk.EW, pady=PADY)
frame.columnconfigure(1, weight=1)
label = ttk.Label(frame, text="Validation Time")
label.grid(row=0, column=0, sticky=tk.W, padx=PADX)
self.validation_time_entry = ttk.Entry(frame)
self.validation_time_entry.insert("end", self.validation_time)
self.validation_time_entry.config(state=tk.DISABLED)
self.validation_time_entry.grid(row=0, column=1, sticky=tk.EW, pady=PADY)
label = ttk.Label(frame, text="Validation Mode")
label.grid(row=1, column=0, sticky=tk.W, padx=PADX)
if self.validation_mode == ServiceValidationMode.BLOCKING:
mode = "BLOCKING"
elif self.validation_mode == ServiceValidationMode.NON_BLOCKING:
mode = "NON_BLOCKING"
else:
mode = "TIMER"
self.validation_mode_entry = ttk.Entry(
frame, textvariable=tk.StringVar(value=mode)
)
self.validation_mode_entry.insert("end", mode)
self.validation_mode_entry.config(state=tk.DISABLED)
self.validation_mode_entry.grid(row=1, column=1, sticky=tk.EW, pady=PADY)
label = ttk.Label(frame, text="Validation Period")
label.grid(row=2, column=0, sticky=tk.W, padx=PADX)
self.validation_period_entry = ttk.Entry(
frame, state=tk.DISABLED, textvariable=tk.StringVar()
)
self.validation_period_entry.grid(row=2, column=1, sticky=tk.EW, pady=PADY)
label_frame = ttk.LabelFrame(tab, text="Executables", padding=FRAME_PAD)
label_frame.grid(sticky=tk.NSEW, pady=PADY)
label_frame.columnconfigure(0, weight=1)
label_frame.rowconfigure(0, weight=1)
listbox_scroll = ListboxScroll(label_frame)
listbox_scroll.grid(sticky=tk.NSEW)
tab.rowconfigure(listbox_scroll.grid_info()["row"], weight=1)
for executable in self.executables:
listbox_scroll.listbox.insert("end", executable)
label_frame = ttk.LabelFrame(tab, text="Dependencies", padding=FRAME_PAD)
label_frame.grid(sticky=tk.NSEW, pady=PADY)
label_frame.columnconfigure(0, weight=1)
label_frame.rowconfigure(0, weight=1)
listbox_scroll = ListboxScroll(label_frame)
listbox_scroll.grid(sticky=tk.NSEW)
tab.rowconfigure(listbox_scroll.grid_info()["row"], weight=1)
for dependency in self.dependencies:
listbox_scroll.listbox.insert("end", dependency)
def draw_buttons(self) -> None:
frame = ttk.Frame(self.top)
frame.grid(sticky=tk.EW)
for i in range(4):
frame.columnconfigure(i, weight=1)
button = ttk.Button(frame, text="Apply", command=self.click_apply)
button.grid(row=0, column=0, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, text="Defaults", command=self.click_defaults)
button.grid(row=0, column=1, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, text="Copy...", command=self.click_copy)
button.grid(row=0, column=2, sticky=tk.EW, padx=PADX)
button = ttk.Button(frame, text="Cancel", command=self.destroy)
button.grid(row=0, column=3, sticky=tk.EW)
def add_filename(self) -> None:
filename = self.filename_combobox.get()
if filename not in self.filename_combobox["values"]:
self.filename_combobox["values"] += (filename,)
self.filename_combobox.set(filename)
self.temp_service_files[filename] = self.service_file_data.text.get(
1.0, "end"
)
else:
logger.debug("file already existed")
def delete_filename(self) -> None:
cbb = self.filename_combobox
filename = cbb.get()
if filename in cbb["values"]:
cbb["values"] = tuple([x for x in cbb["values"] if x != filename])
cbb.set("")
self.service_file_data.text.delete(1.0, "end")
self.temp_service_files.pop(filename, None)
if filename in self.modified_files:
self.modified_files.remove(filename)
@classmethod
def add_command(cls, event: tk.Event) -> None:
frame_contains_button = event.widget.master
listbox = frame_contains_button.master.grid_slaves(row=1, column=0)[0].listbox
command_to_add = frame_contains_button.grid_slaves(row=0, column=0)[0].get()
if command_to_add == "":
return
for cmd in listbox.get(0, tk.END):
if cmd == command_to_add:
return
listbox.insert(tk.END, command_to_add)
@classmethod
def update_entry(cls, event: tk.Event) -> None:
listbox = event.widget
current_selection = listbox.curselection()
if len(current_selection) > 0:
cmd = listbox.get(current_selection[0])
entry = listbox.master.master.grid_slaves(row=0, column=0)[0].grid_slaves(
row=0, column=0
)[0]
entry.delete(0, "end")
entry.insert(0, cmd)
@classmethod
def delete_command(cls, event: tk.Event) -> None:
button = event.widget
frame_contains_button = button.master
listbox = frame_contains_button.master.grid_slaves(row=1, column=0)[0].listbox
current_selection = listbox.curselection()
if len(current_selection) > 0:
listbox.delete(current_selection[0])
entry = frame_contains_button.grid_slaves(row=0, column=0)[0]
entry.delete(0, tk.END)
def click_apply(self) -> None:
if (
not self.is_custom_command()
and not self.is_custom_service_file()
and not self.has_new_files()
and not self.is_custom_directory()
):
self.node.service_configs.pop(self.service_name, None)
self.current_service_color("")
self.destroy()
return
files = set(self.filenames)
if (
self.is_custom_command()
or self.has_new_files()
or self.is_custom_directory()
):
startup, validate, shutdown = self.get_commands()
files = set(self.filename_combobox["values"])
service_data = NodeServiceData(
configs=list(files),
dirs=self.temp_directories,
startup=startup,
validate=validate,
shutdown=shutdown,
)
logger.info("setting service data: %s", service_data)
self.node.service_configs[self.service_name] = service_data
for file in self.modified_files:
if file not in files:
continue
file_configs = self.node.service_file_configs.setdefault(
self.service_name, {}
)
file_configs[file] = self.temp_service_files[file]
self.current_service_color("green")
self.destroy()
def display_service_file_data(self, event: tk.Event) -> None:
filename = self.filename_combobox.get()
self.service_file_data.text.delete(1.0, "end")
self.service_file_data.text.insert("end", self.temp_service_files[filename])
def update_temp_service_file_data(self, event: tk.Event) -> None:
filename = self.filename_combobox.get()
self.temp_service_files[filename] = self.service_file_data.text.get(1.0, "end")
if self.temp_service_files[filename] != self.original_service_files.get(
filename, ""
):
self.modified_files.add(filename)
else:
self.modified_files.discard(filename)
def is_custom_command(self) -> bool:
startup, validate, shutdown = self.get_commands()
return (
set(self.default_startup) != set(startup)
or set(self.default_validate) != set(validate)
or set(self.default_shutdown) != set(shutdown)
)
def has_new_files(self) -> bool:
return set(self.filenames) != set(self.filename_combobox["values"])
def is_custom_service_file(self) -> bool:
return len(self.modified_files) > 0
def is_custom_directory(self) -> bool:
return set(self.default_directories) != set(self.dir_list.listbox.get(0, "end"))
def click_defaults(self) -> None:
"""
clears out any custom configuration permanently
"""
# clear coreclient data
self.node.service_configs.pop(self.service_name, None)
file_configs = self.node.service_file_configs.pop(self.service_name, {})
file_configs.pop(self.service_name, None)
self.temp_service_files = dict(self.original_service_files)
self.modified_files.clear()
# reset files tab
files = list(self.default_config.configs[:])
self.filenames = files
self.filename_combobox.config(values=files)
self.service_file_data.text.delete(1.0, "end")
if len(files) > 0:
filename = files[0]
self.filename_combobox.set(filename)
self.service_file_data.text.insert("end", self.temp_service_files[filename])
# reset commands
self.startup_commands_listbox.delete(0, tk.END)
self.validate_commands_listbox.delete(0, tk.END)
self.shutdown_commands_listbox.delete(0, tk.END)
for cmd in self.default_startup:
self.startup_commands_listbox.insert(tk.END, cmd)
for cmd in self.default_validate:
self.validate_commands_listbox.insert(tk.END, cmd)
for cmd in self.default_shutdown:
self.shutdown_commands_listbox.insert(tk.END, cmd)
# reset directories
self.directory_entry.delete(0, "end")
self.dir_list.listbox.delete(0, "end")
self.temp_directories = list(self.default_directories)
for d in self.default_directories:
self.dir_list.listbox.insert("end", d)
self.current_service_color("")
def click_copy(self) -> None:
file_name = self.filename_combobox.get()
dialog = CopyServiceConfigDialog(
self.app, self, self.node.name, self.service_name, file_name
)
dialog.show()
@classmethod
def append_commands(
cls, commands: List[str], listbox: tk.Listbox, to_add: List[str]
) -> None:
for cmd in to_add:
commands.append(cmd)
listbox.insert(tk.END, cmd)
def get_commands(self) -> Tuple[List[str], List[str], List[str]]:
startup = self.startup_commands_listbox.get(0, "end")
shutdown = self.shutdown_commands_listbox.get(0, "end")
validate = self.validate_commands_listbox.get(0, "end")
return startup, validate, shutdown
def find_directory_button(self) -> None:
d = filedialog.askdirectory(initialdir="/")
self.directory_entry.delete(0, "end")
self.directory_entry.insert("end", d)
def add_directory(self) -> None:
directory = self.directory_entry.get()
directory = Path(directory)
if directory.is_dir():
if str(directory) not in self.temp_directories:
self.dir_list.listbox.insert("end", directory)
self.temp_directories.append(str(directory))
def remove_directory(self) -> None:
d = self.directory_entry.get()
dirs = self.dir_list.listbox.get(0, "end")
if d and d in self.temp_directories:
self.temp_directories.remove(d)
try:
i = dirs.index(d)
self.dir_list.listbox.delete(i)
except ValueError:
logger.debug("directory is not in the list")
self.directory_entry.delete(0, "end")
def directory_select(self, event) -> None:
i = self.dir_list.listbox.curselection()
if i:
d = self.dir_list.listbox.get(i)
self.directory_entry.delete(0, "end")
self.directory_entry.insert("end", d)
def current_service_color(self, color="") -> None:
"""
change the current service label color
"""
listbox = self.master.current.listbox
services = listbox.get(0, tk.END)
listbox.itemconfig(services.index(self.service_name), bg=color)