Files
Task-Pipeline/task-pipeline.py
T
2026-10-31 22:36:05 +01:00

1260 lines
40 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
### pip install doit tomli-w
# pip install tomli-w
import re
import os
import json
import tkinter as tk
from tkinter import ttk
from tkinter import filedialog
from tkinter import messagebox
from tkinter.scrolledtext import ScrolledText
import tomllib
import subprocess
import threading
from pathlib import Path
##############################################################################
# Pipeline
##############################################################################
from datetime import datetime
from typing import Any
def _flatten_sources(
sources: dict[str, Any],
prefix: str = "",
) -> dict[str, str | Path]:
"""Wandelt verschachtelte Source-Dictionaries in {flacher_name: pfad} um."""
flat: dict[str, str | Path] = {}
for name, value in sources.items():
key = f"{prefix}/{name}" if prefix else name
if isinstance(value, dict):
flat.update(_flatten_sources(value, prefix=key))
else:
flat[key] = value
return flat
def read_source_mtimes(sources_dict: dict[str, Any]) -> dict[str, float | None]:
"""
sources_dict: {name: pfad} oder verschachtelt {gruppe: {name: pfad}}
Returns: {flacher_name: mtime_float | None}
"""
mtimes: dict[str, float | None] = {}
for name, path_value in _flatten_sources(sources_dict).items():
path = Path(path_value)
if not path.is_file():
mtimes[name] = None
else:
mtimes[name] = path.stat().st_mtime
return mtimes
def sources_changed(
sources_dict: dict[str, Any],
old_mtimes: dict[str, float | None],
) -> tuple[bool, list[str]]:
"""
Prüft, ob sich mindestens eine Source gegenüber old_mtimes verändert hat.
Returns: (changed, changed_names)
"""
changed_names: list[str] = []
for name, path_value in _flatten_sources(sources_dict).items():
path = Path(path_value)
if not path.is_file():
# Datei ist verschwunden oder war vorher nicht vorhanden
if old_mtimes.get(name) is not None:
changed_names.append(name)
continue
new_mtime = path.stat().st_mtime
if old_mtimes.get(name) != new_mtime:
changed_names.append(name)
return bool(changed_names), changed_names
def max_zeitstempel(pfad: str | Path) -> datetime:
p = Path(pfad)
s = p.stat()
return datetime.fromtimestamp(max(s.st_birthtime, s.st_mtime))
def iter_values(obj):
if isinstance(obj, dict):
for value in obj.values():
yield from iter_values(value)
else:
yield obj
def isvalid(cfg):
min_target = None
if 'targets' in cfg:
targets = list(iter_values(cfg['targets']))
for target in targets:
try:
if min_target == None:
min_target = max_zeitstempel(target)
else:
min_target = min(max_zeitstempel(target), min_target)
except:
return False
max_source = None
if 'sources' in cfg:
sources = list(iter_values(cfg['sources']))
try:
for source in sources:
if max_source == None:
max_source = max_zeitstempel(source)
else:
max_source = min(max_zeitstempel(source), max_source)
except:
return False
if max_source == None or min_target == None:
return True
else:
return max_source < min_target
def resolve(obj, vars):
if isinstance(obj, dict):
return {
k: resolve(v, vars)
for k, v in obj.items()
}
elif isinstance(obj, list):
return [
resolve(v, vars)
for v in obj
]
elif isinstance(obj, str):
return obj.format(**vars)
else:
return obj
class Pipeline:
def __init__(self, filename):
self.filename = Path(filename)
self.load()
def load(self):
with open(self.filename, "rb") as f:
self.config = tomllib.load(f)
self.globals = self.config.get(
"globals",
{}
)
self.tasks = self.config.get(
"task",
{}
)
def topological_order(self):
visited = set()
result = []
def visit(task):
if task in visited:
return
visited.add(task)
for dep in self.tasks[task].get(
"depends_on",
[]
):
visit(dep)
result.append(task)
for task in self.tasks:
visit(task)
return result
##############################################################################
# GUI
##############################################################################
class ToolTip:
def __init__(self, widget, text, delay=500):
self.widget = widget
self.text = text
self.delay = delay
self.window = None
self.after_id = None
def schedule(self, event=None):
self.cancel()
self.after_id = self.widget.after(
self.delay,
lambda: self.show(event)
)
def cancel(self, event=None):
if self.after_id is not None:
self.widget.after_cancel(self.after_id)
self.after_id = None
self.hide()
def show(self, event=None):
self.after_id = None
if self.window is not None:
return
# Position des Mauszeigers in Bildschirmkoordinaten
x = self.widget.winfo_pointerx() + 12
y = self.widget.winfo_pointery() + 12
self.window = tk.Toplevel(self.widget)
self.window.wm_overrideredirect(True)
self.window.attributes("-topmost", True)
self.window.geometry(f"+{x}+{y}")
label = tk.Label(
self.window,
text=self.text,
justify=tk.LEFT,
background="#fff8c6",
foreground="#000000",
relief=tk.SOLID,
borderwidth=1,
padx=6,
pady=4
)
label.pack()
def hide(self):
if self.window is not None:
self.window.destroy()
self.window = None
class PipelineGUI:
COLORS = {
"waiting": "#6c757d",
"running": "#0d6efd",
"done": "#28a745",
"failed": "#dc3545",
"corrupt": "#fd7e14",
"missing": "#ffc107",
}
def __init__(self, root):
self.root = root
self.pipeline = None
self.pipeline_tasks_old = None
self.pipeline_file = None
self.status = {}
self.build_gui()
def _toml_changed_since_start(self):
if self.pipeline_file is None:
return False
if not self.pipeline_file.is_file():
# Datei verschwunden → behandeln wie Änderung
return True
if self._pipeline_toml_mtime_start is None:
return False
current_mtime = self.pipeline_file.stat().st_mtime
return current_mtime != self._pipeline_toml_mtime_start
def open_single_target(self, target_path, name=None):
p = Path(target_path)
if p.is_file():
if name:
self.write_log(f"Öffne Target '{name}': {p}")
else:
self.write_log(f"Öffne Target: {p}")
if os.name == "nt": # Windows
os.startfile(str(p))
else: # macOS / Linux
import platform
system = platform.system()
if system == "Darwin":
subprocess.Popen(["open", str(p)])
else:
subprocess.Popen(["xdg-open", str(p)])
else:
self.write_log(f"Target existiert nicht: {p}")
def _open_target_selection_dialog(self, task, targets_dict):
if not targets_dict:
self.write_log(f"Keine Targets für Task '{task}' (targets_dict ist leer).")
return
targets_dict = dict(targets_dict)
self.write_log(f"Targets für '{task}': {list(targets_dict.keys())}")
dlg = tk.Toplevel(self.root)
dlg.title(f"Targets auswählen – {task}")
tk.Label(
dlg,
text=f"Wähle ein oder mehrere Targets für Task '{task}':"
).pack(padx=10, pady=(10, 5))
listbox = tk.Listbox(
dlg,
width=80,
height=min(len(targets_dict), 12),
selectmode=tk.EXTENDED
)
listbox.pack(padx=10, pady=5, fill=tk.BOTH, expand=True)
items = list(targets_dict.items())
self.write_log(f"Listbox-Items: {len(items)}")
for name, path in items:
listbox.insert(tk.END, f"{name} → {path}")
def on_ok():
sel = listbox.curselection()
if not sel:
self.write_log("Kein Target in der Liste ausgewählt.")
return
to_open = [items[idx] for idx in sel]
dlg.destroy()
def open_selected():
for name, path in to_open:
self.open_single_target(path, name)
self.root.after_idle(open_selected)
def on_cancel():
dlg.destroy()
btn_frame = ttk.Frame(dlg)
btn_frame.pack(pady=10)
ttk.Button(btn_frame, text="Öffnen", command=on_ok).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Abbrechen", command=on_cancel).pack(side=tk.LEFT, padx=5)
dlg.bind("<Return>", lambda e: on_ok())
dlg.bind("<Escape>", lambda e: on_cancel())
dlg.update_idletasks()
dlg.transient(self.root)
dlg.grab_set()
w = dlg.winfo_width()
h = dlg.winfo_height()
x = self.root.winfo_x() + (self.root.winfo_width() - w) // 2
y = self.root.winfo_y() + (self.root.winfo_height() - h) // 2
dlg.geometry(f"+{x}+{y}")
dlg.wait_window()
def open_task_targets(self, task):
if self.pipeline is None:
return
cfg = self.pipeline.tasks.get(task, {})
targets = cfg.get("targets", {})
if not targets:
self.write_log(f"Keine Targets für Task '{task}' definiert.")
return
# Sicherstellen, dass targets ein dict ist
if not isinstance(targets, dict):
self.write_log(f"Targets für Task '{task}' haben unerwartetes Format.")
return
# Nur ein Target → direkt öffnen
if len(targets) == 1:
name, path = next(iter(targets.items()))
self.open_single_target(path, name)
return
# Mehrere Targets → Auswahl-Dialog
self._open_target_selection_dialog(task, targets)
def open_task_targets_old(self, task):
if self.pipeline is None:
return
cfg = self.pipeline.tasks.get(task, {})
targets = cfg.get("targets", {})
if not targets:
self.write_log(f"Keine Targets für Task '{task}' definiert.")
return
# Öffne alle Targets (als Dateien)
for name, target_path in targets.items():
p = Path(target_path)
if p.is_file():
self.write_log(f"Öffne Target '{name}': {p}")
# Plattformunabhängig öffnen:
if os.name == "nt": # Windows
os.startfile(str(p))
elif os.name == "posix": # macOS / Linux
# macOS: open, Linux: xdg-open
import platform
system = platform.system()
if system == "Darwin":
subprocess.Popen(["open", str(p)])
else:
subprocess.Popen(["xdg-open", str(p)])
else:
self.write_log(f"Target '{name}' existiert nicht: {p}")
##########################################################################
def build_gui(self):
style = ttk.Style()
style.theme_use("clam")
style.configure(
".",
background="#1e1e1e",
foreground="#d4d4d4"
)
style.configure(
"TFrame",
background="#525252"
)
style.configure(
"TLabel",
background="#1e1e1e",
foreground="#d4d4d4"
)
style.configure(
"TButton",
background="#2d2d30",
foreground="#ffffff",
borderwidth=1
)
style.map(
"TButton",
background=[
("active", "#3e3e42")
]
)
style.configure(
"TPanedwindow",
background="#a1a1a1"
)
style.configure(
"TScrollbar",
background="#707070",
troughcolor="#1e1e1e",
arrowcolor="white",
borderwidth=0
)
style.configure(
"TProgressbar",
troughcolor="#2d2d30",
background="#0e639c"
)
self.root.title(
"Pipeline GUI"
)
self.root.geometry(
"1600x900"
)
self.root.configure(
bg="#1e1e1e"
)
main = ttk.PanedWindow(
self.root,
orient=tk.HORIZONTAL
)
main.pack(
fill=tk.BOTH,
expand=True
)
######################################################################
# Links
######################################################################
left = ttk.Frame(main)
main.add(
left,
weight=1
)
toolbar = ttk.Frame(left)
toolbar.pack(
fill=tk.X
)
ttk.Button(
toolbar,
text="Öffnen",
command=self.open_pipeline
).pack(side=tk.LEFT)
ttk.Button(
toolbar,
text="Speichern",
command=self.save_toml
).pack(side=tk.LEFT)
editor_frame = ttk.Frame(left)
editor_frame.pack(
fill=tk.BOTH,
expand=True
)
editor_vscroll = ttk.Scrollbar(
editor_frame,
orient=tk.VERTICAL
)
editor_hscroll = ttk.Scrollbar(
editor_frame,
orient=tk.HORIZONTAL
)
self.editor = tk.Text(
editor_frame,
bg="#1e1e1e",
fg="#d4d4d4",
insertbackground="white",
font=("Consolas", 10),
undo=True,
wrap="none",
yscrollcommand=editor_vscroll.set,
xscrollcommand=editor_hscroll.set
)
self.editor.tag_configure(
"task_highlight",
background="#fff3b0" # helles Gelb
)
editor_vscroll.config(
command=self.editor.yview
)
editor_vscroll.pack(
side=tk.RIGHT,
fill=tk.Y
)
editor_hscroll.config(
command=self.editor.xview
)
editor_hscroll.pack(
side=tk.BOTTOM,
fill=tk.X
)
self.editor.pack(
side=tk.LEFT,
fill=tk.BOTH,
expand=True
)
self.editor.configure(
selectbackground="#264f78",
selectforeground="white",
insertwidth=2,
relief="flat",
borderwidth=0
)
self.editor.pack(
fill=tk.BOTH,
expand=True
)
self.editor.tag_config(
"section",
foreground="#569cd6"
)
self.editor.tag_config(
"key",
foreground="#9cdcfe"
)
self.editor.tag_config(
"string",
foreground="#ce9178"
)
self.editor.tag_config(
"comment",
foreground="#6a9955"
)
self.editor.tag_config(
"number",
foreground="#b5cea8"
)
self.editor.tag_config(
"boolean",
foreground="#c586c0"
)
self.editor.bind(
"<KeyRelease>",
self.on_editor_changed
)
self.editor.bind("<Control-s>", self.save_toml)
######################################################################
# Rechts
######################################################################
right = ttk.Frame(main)
main.add(
right,
weight=1
)
######################################################################
# Scrollbarer Graph
######################################################################
graph_frame = ttk.Frame(right)
graph_frame.pack(
fill=tk.BOTH,
expand=True
)
self.v_scroll = ttk.Scrollbar(
graph_frame,
orient=tk.VERTICAL
)
self.h_scroll = ttk.Scrollbar(
graph_frame,
orient=tk.HORIZONTAL
)
self.canvas = tk.Canvas(
graph_frame,
bg="#252526",
highlightthickness=0,
xscrollcommand=self.h_scroll.set,
yscrollcommand=self.v_scroll.set
)
self.v_scroll.config(
command=self.canvas.yview
)
self.h_scroll.config(
command=self.canvas.xview
)
self.v_scroll.pack(
side=tk.RIGHT,
fill=tk.Y
)
self.h_scroll.pack(
side=tk.BOTTOM,
fill=tk.X
)
self.canvas.pack(
side=tk.LEFT,
fill=tk.BOTH,
expand=True
)
self.canvas.bind(
"<MouseWheel>",
self.on_mousewheel
)
######################################################################
# Progress
######################################################################
self.progress = ttk.Progressbar(
right,
mode="determinate"
)
self.progress.pack(
fill=tk.X
)
######################################################################
# Buttons
######################################################################
buttons = ttk.Frame(right)
buttons.pack(
fill=tk.X
)
ttk.Button(
buttons,
text="Ausführen",
command=self.run_pipeline
).pack(side=tk.LEFT)
ttk.Button(
buttons,
text="Neu laden",
# command=self.reload_pipeline
command=self.save_toml
).pack(side=tk.LEFT)
######################################################################
# Log
######################################################################
self.log = ScrolledText(
right,
height=15,
bg="#1e1e1e",
fg="#d4d4d4",
insertbackground="white"
)
self.log.config(state=tk.DISABLED)
self.log.configure(
selectbackground="#264f78",
selectforeground="white",
relief="flat",
borderwidth=0
)
self.log.pack(
fill=tk.BOTH,
expand=True
)
##########################################################################
def write_log(self, text):
self.log.config(state=tk.NORMAL)
self.log.insert(
tk.END,
text + "\n"
)
self.log.config(state=tk.DISABLED)
self.log.see(
tk.END
)
##########################################################################
def on_editor_changed(self, event=None):
self.highlight_toml()
##########################################################################
def highlight_toml(self):
text = self.editor.get(
"1.0",
"end-1c"
)
tags = [
"section",
"key",
"string",
"comment",
"number",
"boolean"
]
for tag in tags:
self.editor.tag_remove(
tag,
"1.0",
tk.END
)
for m in re.finditer(
r"^\[.*?\]",
text,
re.MULTILINE
):
self.editor.tag_add(
"section",
f"1.0+{m.start()}c",
f"1.0+{m.end()}c"
)
for m in re.finditer(
r"^([A-Za-z_][A-Za-z0-9_]*)\s*=",
text,
re.MULTILINE
):
key = m.group(1)
self.editor.tag_add(
"key",
f"1.0+{m.start()}c",
f"1.0+{m.start() + len(key)}c"
)
for m in re.finditer(
r'"[^"]*"',
text
):
self.editor.tag_add(
"string",
f"1.0+{m.start()}c",
f"1.0+{m.end()}c"
)
for m in re.finditer(
r"#.*$",
text,
re.MULTILINE
):
self.editor.tag_add(
"comment",
f"1.0+{m.start()}c",
f"1.0+{m.end()}c"
)
for m in re.finditer(
r"\b\d+\b",
text
):
self.editor.tag_add(
"number",
f"1.0+{m.start()}c",
f"1.0+{m.end()}c"
)
for m in re.finditer(
r"\b(true|false)\b",
text,
re.IGNORECASE
):
self.editor.tag_add(
"boolean",
f"1.0+{m.start()}c",
f"1.0+{m.end()}c"
)
##########################################################################
def on_mousewheel(self, event):
self.canvas.yview_scroll(
int(-event.delta / 120),
"units"
)
##########################################################################
def open_pipeline(self):
filename = filedialog.askopenfilename(
filetypes=[
("TOML", "*.toml"),
("Alle Dateien", "*.*")
]
)
if not filename:
return
self.pipeline_file = Path(filename)
text = self.pipeline_file.read_text(
encoding="utf-8"
)
self.editor.delete(
"1.0",
tk.END
)
self.editor.insert(
"1.0",
text
)
self.highlight_toml()
self.reload_pipeline()
self.write_log(
f"Geladen: {filename}\n"
)
##########################################################################
def save_toml(self, event=None):
if self.pipeline_file is None:
messagebox.showwarning(
"Hinweis",
"Bitte zuerst eine TOML-Datei öffnen."
)
return
text = self.editor.get(
"1.0",
"end-1c"
)
self.pipeline_file.write_text(
text,
encoding="utf-8"
)
self.write_log(
"TOML gespeichert"
)
self.reload_pipeline()
##########################################################################
def reload_pipeline(self):
if self.pipeline_file is None:
return
try:
pipeline_temp = Pipeline(
self.pipeline_file
)
pipeline_temp.tasks = resolve(pipeline_temp.tasks, pipeline_temp.globals)
# gelöschte Task bereinigen: löschen der Targets
if self.pipeline:
targets = []
order = self.pipeline.topological_order()
for i, task in enumerate(order):
sources = []
if 'sources' in self.pipeline.tasks[task]:
sources = list(iter_values(self.pipeline.tasks[task]['sources']))
if 'depends_on' in self.pipeline.tasks[task]:
del self.pipeline.tasks[task]['depends_on']
if (task not in pipeline_temp.tasks or
self.pipeline.tasks[task] != pipeline_temp.tasks[task]):
if 'targets' in self.pipeline.tasks[task] and len(sources) > 0:
for target in self.pipeline.tasks[task]['targets'].values():
if Path(target).is_file():
os.remove(target)
# Target-liste mit Owner erstellen für Abhängigkeiten
self.pipeline = pipeline_temp
deps = {}
for task in self.pipeline.tasks:
if 'targets' in self.pipeline.tasks[task]:
for target in self.pipeline.tasks[task]['targets'].values():
deps[target] = task
# Korrupte Tasks identifieren: Tasks die eine Source verwenden
# die im TOML-File nicht deklariert ist
corrupt = []
for task in self.pipeline.tasks:
if 'sources' in self.pipeline.tasks[task]:
sources = list(iter_values(self.pipeline.tasks[task]['sources']))
self.pipeline.tasks[task]['depends_on'] = []
for source in sources:
if source in deps:
if deps[source] not in self.pipeline.tasks[task]['depends_on']:
self.pipeline.tasks[task]['depends_on'].append(deps[source])
elif task not in corrupt:
corrupt.append(task)
# Targets von korrupten Tasks löschen
for task in corrupt:
if 'targets' in self.pipeline.tasks[task]:
for target in self.pipeline.tasks[task]['targets'].values():
if Path(target).is_file():
os.remove(target)
# Status der Tasks aktualisieren und Targets löschen von Tasks, die nicht
# den Status "done" bekommen.
order = self.pipeline.topological_order()
for i, task in enumerate(order):
if 'sources' not in self.pipeline.tasks[task]:
self.status[task] = "done"
if 'targets' in self.pipeline.tasks[task]:
for target in self.pipeline.tasks[task]['targets'].values():
if not Path(target).is_file():
self.status[task] = "missing"
elif isvalid(self.pipeline.tasks[task]):
self.status[task] = "done"
else:
if task in corrupt:
self.status[task] = "corrupt"
else:
self.status[task] = "waiting"
if 'targets' in self.pipeline.tasks[task]:
for target in self.pipeline.tasks[task]['targets'].values():
if Path(target).is_file():
os.remove(target)
# Status von neuen Tasks auf "waiting setzen"
self.status = {
task: self.status[task] if task in self.status.keys() else "waiting"
for task in self.pipeline.tasks
}
self.log.config(state=tk.NORMAL)
self.log.delete("1.0", tk.END)
self.log.config(state=tk.DISABLED)
self.progress["value"] = 0
self.draw_graph()
except Exception as e:
self.write_log(
f"Fehler: {e}"
)
##########################################################################
def draw_graph(self):
self.canvas.delete(
"all"
)
if self.pipeline is None:
return
######################################################################
# Level bestimmen
######################################################################
levels = {}
def calculate_level(task):
if task in levels:
return levels[task]
deps = self.pipeline.tasks[task].get(
"depends_on",
[]
)
if not deps:
levels[task] = 0
else:
levels[task] = (
max(
calculate_level(dep)
for dep in deps
)
+ 1
)
return levels[task]
for task in self.pipeline.tasks:
calculate_level(task)
######################################################################
# Positionen
######################################################################
columns = {}
for task, lvl in levels.items():
columns.setdefault(
lvl,
[]
).append(task)
positions = {}
column_width = 300
row_height = 120
for lvl, tasks in columns.items():
for row, task in enumerate(tasks):
x = 150 + lvl * column_width
y = 100 + row * row_height
positions[task] = (
x,
y
)
######################################################################
# Kanten
######################################################################
for task, cfg in self.pipeline.tasks.items():
x1, y1 = positions[task]
for dep in cfg.get(
"depends_on",
[]
):
x2, y2 = positions[dep]
self.canvas.create_line(
x2 + 80,
y2,
x1 - 80,
y1,
arrow=tk.LAST,
width=2,
fill="#909090"
)
######################################################################
# Knoten
######################################################################
for task, (x, y) in positions.items():
color = self.COLORS.get(
self.status.get(
task,
"waiting"
),
"lightgray"
)
rect = self.canvas.create_rectangle(
x - 80,
y - 25,
x + 80,
y + 25,
fill=color,
outline="#707070",
tags=f"task_{task}"
)
label = task
if self.status.get(task) == "running":
label = f"⏳ {task}"
elif self.status.get(task) == "done":
label = f"✅ {task}"
elif self.status.get(task) == "failed":
label = f"❌ {task}"
elif self.status.get(task) == "corrupt":
label = f"🚫 {task}"
elif self.status.get(task) == "missing":
label = f"⚠️ {task}"
text = self.canvas.create_text(
x,
y,
text=label,
fill="white",
tags=f"task_{task}"
)
# Klick-Handler für diesen Task
def on_task_click(event, t=task):
# springe zu Position in der toml-Datei
self.scroll_toml_to_task(t)
# Doppelklick-Handler für diesen Task
def on_task_dblclick(event, t=task):
self.open_task_targets(t)
# Enter-Handler für diesen Task
def on_task_enter(event, t=task):
self.canvas.config(cursor="hand2")
cfg = self.pipeline.tasks.get(t, {})
description = cfg.get("description", {})
if hasattr(self, "tooltip"):
self.tooltip.cancel()
del self.tooltip
if description:
self.tooltip = ToolTip(
event.widget,
description
)
self.tooltip.schedule()
# Leave-Handler für diesen Task
def on_task_leave(event, t=task):
self.canvas.config(cursor="")
if hasattr(self, "tooltip"):
self.tooltip.cancel()
del self.tooltip
self.canvas.tag_bind(f"task_{task}", "<Button-1>", on_task_click)
self.canvas.tag_bind(f"task_{task}", "<Double-Button-1>", on_task_dblclick)
self.canvas.tag_bind(f"task_{task}", "<Enter>", on_task_enter)
self.canvas.tag_bind(f"task_{task}", "<Leave>", on_task_leave)
bbox = self.canvas.bbox(
"all"
)
if bbox:
self.canvas.configure(
scrollregion=(
bbox[0] - 100,
bbox[1] - 100,
bbox[2] + 100,
bbox[3] + 100,
)
)
def scroll_toml_to_task(self, task):
if not hasattr(self, "editor") or self.editor is None:
return
self.editor.tag_remove("task_highlight", "1.0", tk.END)
# Suche zuerst nach dem vollständigen TOML-Header
pattern = f"[task.{task}]"
start = self.editor.search(pattern, "1.0", stopindex=tk.END, exact=True)
# Fallback: Task-Name irgendwo im TOML suchen
if not start:
start = self.editor.search(task, "1.0", stopindex=tk.END, exact=True)
if not start:
self.write_log(f"Task '{task}' im TOML nicht gefunden.")
return
# Cursor setzen und hinscrollen
self.editor.mark_set("insert", start)
self.editor.see(start)
self.editor.focus_set()
# Optional: Zeile hervorheben
line_start = self.editor.index(f"{start} linestart")
line_end = self.editor.index(f"{start} lineend")
self.editor.tag_add("task_highlight", line_start, line_end)
##########################################################################
def execute_pipeline(self):
self.root.after(0, lambda: self.root.config(cursor="watch"))
try:
order = self.pipeline.topological_order()
total = len(order)
for i, task in enumerate(order):
# TOML-Änderung prüfen (vor jedem Task)
if self._toml_changed_since_start():
self.write_log(
"TOML-Datei hat sich während der Ausführung geändert. "
"Pipeline wird abgebrochen."
)
# Optional: aktuellen Task als failed markieren
self.status[task] = "failed"
self.draw_graph()
return
cfg = self.pipeline.tasks[task]
cfg["globals"] = self.pipeline.globals
# Vorab-Checks wie bisher
if self.status[task] == "corrupt":
rc = -1
self.write_log(f"{task}: source input is corrupt...")
elif self.status[task] == "missing":
rc = 0
elif self.status[task] != "done":
self.status[task] = "running"
self.draw_graph()
self.write_log(f"Starte {task}")
# Sources vorbereiten
sources_cfg = cfg.get("sources", {})
if sources_cfg:
sources_dict = dict(resolve(sources_cfg, self.pipeline.globals))
old_mtimes = read_source_mtimes(sources_dict)
else:
sources_dict = {}
old_mtimes = {}
rc = 0
if "action" in cfg:
env = os.environ.copy()
env["PIPELINE_CFG"] = json.dumps(cfg)
proc = subprocess.Popen(
cfg["action"],
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
encoding="utf-8",
env=env
)
current_line = ""
aborted = False
while True:
ch = proc.stdout.read(1)
if isinstance(ch, bytes):
ch = ch.decode("utf-8", errors="replace")
if not ch and proc.poll() is not None:
break
if ch == "\r":
start = self.log.index("end-1c linestart")
self.log.config(state=tk.NORMAL)
self.log.delete(start, "end-1c")
self.log.insert(start, current_line)
self.log.config(state=tk.DISABLED)
self.log.see(tk.END)
current_line = ""
elif ch == "\n":
self.log.config(state=tk.NORMAL)
self.log.insert(tk.END, current_line + "\n")
self.log.config(state=tk.DISABLED)
self.log.see(tk.END)
current_line = ""
else:
current_line += ch
# Prüfen, ob sich Sources geändert haben
if sources_dict:
changed, changed_names = sources_changed(sources_dict, old_mtimes)
if changed:
self.write_log(
f"{task}: Source-Änderung detektiert ({', '.join(changed_names)}). "
"Targets werden gelöscht."
)
aborted = True
if not aborted:
self.write_log("") # Zeile zwischen Tasks
rc = proc.wait()
else:
# Prozess wurde abgebrochen → rc != 0 setzen
rc = -1
else:
rc = 0
self.write_log(f"{task}: task is up to date...\n")
if rc != 0:
self.status[task] = "failed"
# Optional: Targets dieses Tasks löschen, da inkonsistent
if "targets" in cfg:
targets_cfg = cfg.get("targets", {})
if targets_cfg:
targets_dict = dict(resolve(targets_cfg, self.pipeline.globals))
for t_name, t_path in targets_dict.items():
p = Path(t_path)
if p.is_file():
self.write_log(f"Lösche inkonsistentes Target {t_name}: {p}")
p.unlink()
self.draw_graph()
self.write_log(f"Fehler in {task}")
return
if self.status[task] != "missing":
self.status[task] = "done"
self.progress["value"] = ((i + 1) * 100 / total)
self.draw_graph()
self.write_log("Pipeline beendet")
finally:
self.root.after(0, lambda: self.root.config(cursor=""))
##########################################################################
def run_pipeline(self):
if self.pipeline is None:
messagebox.showwarning(
"Hinweis",
"Bitte zuerst eine TOML-Datei öffnen."
)
return
# Zeitstempel der TOML vor Start merken
if self.pipeline_file is not None and self.pipeline_file.is_file():
self._pipeline_toml_mtime_start = self.pipeline_file.stat().st_mtime
else:
self._pipeline_toml_mtime_start = None
threading.Thread(
target=self.execute_pipeline,
daemon=True
).start()
##############################################################################
# MAIN
##############################################################################
root = tk.Tk()
PipelineGUI(root)
root.mainloop()