Мобильная версияВходной файл cfg
[job1]
EXEC_PATH = /usr/path
RUNIT = echo job1
TYPES = ['schedule' , 'tumbling']
[job2]
EXEC_PATH = /usr/path
RUNIT = echo job2
PARENTS = job1
TYPES = [ 'tumbling']
[job3]
EXEC_PATH = /usr/path
RUNIT = echo job3
PARENTS = job2
tkinter показывает порядок заданий на основе параметра PARENTS, определенного в файле конфигурации.
Проблемы
- У меня есть меню опций, определенное для каждого узла/задания в графическом интерфейсе. Если я последовательно нажимаю «Показать сведения о задании» и «Редактировать сведения о задании» или иногда даже один раз, нажимая на меню с текстом ниже в правой части, tk. Поле ввода:
- После редактирование, в правой части поля tk. Если я нажму «Сохранить», отредактированная информация должна быть сохранена в файле cfg на диске, чего не происходит.
- Можно ли переместить вверх кнопку «Проверить» чуть ниже того места, где заканчиваются параметры задания/узла?
- Вместо этого диалоговое окно проверки входного файла, могу ли я использовать метку с зеленым и красным цветом текст об успехе и неудаче соответственно?
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from argparse import ArgumentParser
import configparser
import os
import subprocess
def parse():
parser = ArgumentParser()
parser.add_argument("-input_file", help="Input input file.", required=False)
parser.add_argument("-type", help="Filter sections by types.", choices=['schedule', 'tumbling'], required=False)
args = parser.parse_args()
return args
class JobLevels:
def __init__(self, input_file=None, type_filter=None):
self.input_file = input_file
self.type_filter = type_filter
self.deps_dict = {}
self.level_dict = {}
self.visited_files = set()
self.sections = {}
def parse_input_file(self, input_file):
if input_file in self.visited_files:
return
self.visited_files.add(input_file)
config = configparser.ConfigParser(allow_no_value=True)
config.optionxform = str # Preserve case sensitivity
config.read(input_file)
for section in config.sections():
if section == 'include':
for include_file in config.options(section):
include_file_path = os.path.join(os.path.dirname(input_file), include_file)
self.parse_input_file(include_file_path)
else:
self.sections[section] = {}
self.sections[section]['EXEC_PATH'] = config.get(section, 'EXEC_PATH', fallback=None)
self.sections[section]['RUNIT'] = config.get(section, 'RUNIT', fallback=None)
self.sections[section]['PARENTS'] = config.get(section, 'PARENTS', fallback=None)
self.sections[section]['TYPES'] = config.get(section, 'TYPES', fallback=None)
self.sections[section]['RATING'] = config.get(section, 'RATING', fallback='0')
def filter_sections_by_type(self):
if not self.type_filter:
return
filtered_sections = {}
for section, attributes in self.sections.items():
types = attributes['TYPES']
if types:
types_list = [type.strip() for type in types.strip('[]').split(',')]
if self.type_filter in types_list:
filtered_sections[section] = attributes
else:
# If TYPES key is not present
filtered_sections[section] = attributes
self.sections = filtered_sections
def build_deps_dict(self):
for section, attributes in self.sections.items():
deps = attributes['PARENTS']
if deps:
deps_list = [dep.strip() for dep in deps.split(',')]
self.deps_dict[section] = deps_list
def assign_level(self, section):
if section not in self.level_dict:
if section not in self.deps_dict or not self.deps_dict[section]:
self.level_dict[section] = 0
else:
levels = []
for dep_section in self.deps_dict[section]:
temp_level = self.assign_level(dep_section)
levels.append(temp_level)
self.level_dict[section] = max(levels) + 1
return self.level_dict[section]
def start_proc(self):
if self.input_file:
self.visited_files.clear()
self.sections.clear()
self.deps_dict.clear()
self.level_dict.clear()
self.parse_input_file(self.input_file)
self.filter_sections_by_type()
self.build_deps_dict()
for section in self.deps_dict:
self.assign_level(section)
class Viewer:
def __init__(self, root, node_order_finder):
self.root = root
self.node_order_finder = node_order_finder
self.main_paned_window = ttk.PanedWindow(root, orient=tk.HORIZONTAL)
self.main_paned_window.pack(fill=tk.BOTH, expand=True)
self.left_paned_window = ttk.PanedWindow(self.main_paned_window, orient=tk.VERTICAL)
self.main_paned_window.add(self.left_paned_window, weight=3)
self.right_panel = tk.Frame(self.main_paned_window, width=200, bg="lightgray")
self.main_paned_window.add(self.right_panel, weight=1)
self.top_frame = tk.Frame(self.left_paned_window)
self.left_paned_window.add(self.top_frame, weight=3)
self.bottom_panel = tk.Frame(self.left_paned_window, height=200, bg="lightgray")
self.left_paned_window.add(self.bottom_panel, weight=1)
self.canvas = tk.Canvas(self.top_frame, bg="white")
self.h_scrollbar = tk.Scrollbar(self.top_frame, orient=tk.HORIZONTAL, command=self.canvas.xview)
self.v_scrollbar = tk.Scrollbar(self.top_frame, orient=tk.VERTICAL, command=self.canvas.yview)
self.scrollable_frame = tk.Frame(self.canvas)
self.scrollable_frame.bind(
"",
lambda e: self.canvas.configure(
scrollregion=self.canvas.bbox("all")
)
)
self.canvas.create_window((0, 0), window=self.scrollable_frame, anchor="nw")
self.canvas.configure(xscrollcommand=self.h_scrollbar.set, yscrollcommand=self.v_scrollbar.set)
self.h_scrollbar.pack(side=tk.BOTTOM, fill=tk.X)
self.v_scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
self.type_var = tk.StringVar(value="All")
self.create_widgets()
self.draw_graph()
def create_widgets(self):
control_frame = tk.Frame(self.right_panel)
control_frame.pack(side=tk.TOP, fill=tk.X)
type_label = tk.Label(control_frame, text="Event:")
type_label.pack(side=tk.LEFT, padx=5, pady=5)
type_options = ["All", "schedule", "tumbling"]
type_menu = ttk.Combobox(control_frame, textvariable=self.type_var, values=type_options)
type_menu.pack(side=tk.LEFT, padx=5, pady=5)
type_menu.bind("", self.on_type_change)
browse_button = tk.Button(control_frame, text="Browse input File", command=self.browse_file)
browse_button.pack(side=tk.LEFT, padx=5, pady=5)
save_button = tk.Button(control_frame, text="Save", command=self.save_input_file)
save_button.pack(side=tk.LEFT, padx=5, pady=5)
self.stage_params_frame = tk.Frame(self.right_panel)
self.stage_params_frame.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
self.validate_button = tk.Button(self.right_panel, text="Validate", command=self.validate_input)
self.validate_button.pack(side=tk.TOP, padx=5, pady=5)
def browse_file(self):
input_file = filedialog.askopenfilename(filetypes=[("input files", "*.input")])
if input_file:
self.node_order_finder.input_file = input_file
self.node_order_finder.start_proc()
self.draw_graph()
def save_input_file(self):
save_file = filedialog.asksaveasfilename(defaultextension=".input", filetypes=[("input files", "*.input")])
if save_file:
config = configparser.ConfigParser()
for section, attributes in self.node_order_finder.sections.items():
config[section] = {k: v.get() if isinstance(v, tk.Entry) else v for k, v in attributes.items()}
with open(save_file, 'w') as configfile:
config.write(configfile)
self.node_order_finder.input_file = save_file
self.node_order_finder.start_proc()
self.draw_graph()
def validate_input(self):
try:
result = subprocess.run(['ls', self.node_order_finder.input_file], capture_output=True, text=True)
if result.returncode == 0:
messagebox.showinfo("Validation", "Validation successful")
else:
messagebox.showerror("Validation", f"Error: {result.stderr}")
except Exception as e:
messagebox.showerror("Validation", f"Error: {str(e)}")
def on_type_change(self, type):
self.node_order_finder.type_filter = self.type_var.get() if self.type_var.get() != "All" else None
self.node_order_finder.start_proc()
self.draw_graph()
def draw_graph(self):
for widget in self.scrollable_frame.winfo_children():
widget.destroy()
self.level_frames = {}
levels = {}
for section, level in self.node_order_finder.level_dict.items():
if level not in levels:
levels[level] = []
levels[level].append(section)
colors = ["lightblue", "lightgreen", "lightyellow", "lightpink", "lightgray"]
for level, nodes in sorted(levels.items()):
level_frame = tk.Frame(self.scrollable_frame, bg=colors[level % len(colors)], bd=2, relief=tk.SOLID)
level_frame.pack(fill=tk.X, padx=10, pady=5)
self.level_frames[level] = level_frame
level_label = tk.Label(level_frame, text=f"Level {level}", bg=colors[level % len(colors)], font=("Arial", 12, "bold"), anchor="w")
level_label.pack(side=tk.TOP, fill=tk.X)
for node in nodes:
self.draw_node(level_frame, node)
def draw_node(self, parent, node):
level = self.node_order_finder.level_dict.get(node, 0)
label = f'{node}({level})'
if node in self.node_order_finder.sections:
if self.node_order_finder.sections[node]['RATING'] == '1':
color = 'lightblue'
else:
color = 'skyblue'
fg_color = 'darkblue'
node_label = tk.Label(parent, text=label, bg=color, fg=fg_color, font=("Arial", 10), bd=1, relief=tk.SOLID, padx=5, pady=5)
node_label.pack(side=tk.LEFT, padx=5, pady=5)
node_label.bind("", lambda type, node=node: self.show_context_menu(type, node))
def show_context_menu(self, type, node):
context_menu = tk.Menu(self.root, tearoff=0)
context_menu.add_command(label="Show job details", command=lambda: self.show_stage_params(node, readonly=True))
context_menu.add_command(label="Edit job details", command=lambda: self.show_stage_params(node, readonly=False))
context_menu.add_command(label="Show Upward Dependency", command=lambda: self.show_upward_dependency(node))
context_menu.add_command(label="Show Downward Dependency", command=lambda: self.show_downward_dependency(node))
context_menu.post(type.x_root, type.y_root)
self.root.bind("", lambda e: context_menu.unpost())
def show_stage_params(self, node, readonly):
for widget in self.stage_params_frame.winfo_children():
widget.destroy()
selected_node_label = tk.Label(self.stage_params_frame, text=f"Stage: {node}", font=("Arial", 12, "bold"))
selected_node_label.pack(anchor="w")
params = self.node_order_finder.sections[node]
for param, value in params.items():
param_frame = tk.Frame(self.stage_params_frame)
param_frame.pack(fill=tk.X, pady=2)
param_label = tk.Label(param_frame, text=f"{param}:")
param_label.pack(side=tk.LEFT)
param_entry = tk.Entry(param_frame)
param_entry.insert(0, value)
param_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
if readonly:
param_entry.config(state='readonly')
params[param] = param_entry
if not readonly:
add_button = tk.Button(self.stage_params_frame, text="+", command=lambda: self.add_param(node))
add_button.pack(side=tk.TOP, padx=5, pady=5)
self.validate_button.pack(side=tk.TOP, padx=5, pady=5)
def add_param(self, node):
param_frame = tk.Frame(self.stage_params_frame)
param_frame.pack(fill=tk.X, pady=2)
param_label = tk.Entry(param_frame)
param_label.pack(side=tk.LEFT)
param_entry = tk.Entry(param_frame)
param_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
remove_button = tk.Button(param_frame, text="-", command=lambda: self.remove_param(param_frame))
remove_button.pack(side=tk.LEFT, padx=5, pady=5)
self.stage_params_frame.params[param_label] = param_entry
def remove_param(self, param_frame):
param_frame.destroy()
def show_upward_dependency(self, node):
for widget in self.bottom_panel.winfo_children():
widget.destroy()
selected_node_label = tk.Label(self.bottom_panel, text=f"Stage: {node}", font=("Arial", 12, "bold"))
selected_node_label.pack(anchor="w")
upward_deps = [dep for dep, deps in self.node_order_finder.deps_dict.items() if node in deps]
self.display_dependencies(upward_deps, "Upward Dependencies")
def show_downward_dependency(self, node):
for widget in self.bottom_panel.winfo_children():
widget.destroy()
selected_node_label = tk.Label(self.bottom_panel, text=f"Stage: {node}", font=("Arial", 12, "bold"))
selected_node_label.pack(anchor="w")
downward_deps = self.node_order_finder.deps_dict.get(node, [])
self.display_dependencies(downward_deps, "Downward Dependencies")
def display_dependencies(self, dependencies, title):
title_label = tk.Label(self.bottom_panel, text=title, font=("Arial", 12, "bold"))
title_label.pack(anchor="w")
for dep in dependencies:
dep_label = tk.Label(self.bottom_panel, text=dep, font=("Arial", 10))
dep_label.pack(anchor="w")
if __name__ == '__main__':
args = parse()
job_obj = JobLevels(args.input_file, args.type)
job_obj.start_proc()
root = tk.Tk()
root.title("Job Visualization")
app = Viewer(root, job_obj)
root.geometry("800x600")
root.mainloop()
Подробнее здесь: https://stackoverflow.com/questions/792 ... -and-savin
