mirror of
https://git.hiabuto.net/C3MA/CCMA.git
synced 2026-09-06 03:20:49 +02:00
302 lines
10 KiB
Python
302 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import threading
|
|
import time
|
|
import tkinter as tk
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from queue import Empty, Queue
|
|
|
|
from ccma import __version__
|
|
from ccma.domain.models import HousekeeperFinding
|
|
from ccma.services.housekeeper import Housekeeper, HousekeeperSettings
|
|
from ccma.storage.repository import MemberRepository
|
|
from ccma.ui.monitors import MonitorBounds, preferred_monitor
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class StartupResult:
|
|
repository: MemberRepository
|
|
validation_errors: list[str]
|
|
findings: list[HousekeeperFinding]
|
|
|
|
|
|
class SplashScreen(tk.Toplevel):
|
|
def __init__(
|
|
self,
|
|
master: tk.Tk,
|
|
repository: MemberRepository,
|
|
on_complete: Callable[[StartupResult], None],
|
|
on_error: Callable[[Exception], None],
|
|
*,
|
|
run_housekeeper: bool = True,
|
|
monitor: MonitorBounds | None = None,
|
|
housekeeper_settings: HousekeeperSettings | None = None,
|
|
minimum_display_seconds: float = 5.0,
|
|
):
|
|
super().__init__(master)
|
|
self.repository = repository
|
|
self.on_complete = on_complete
|
|
self.on_error = on_error
|
|
self.run_housekeeper = run_housekeeper
|
|
self.monitor = monitor
|
|
self.housekeeper_settings = housekeeper_settings
|
|
self.minimum_display_seconds = (
|
|
max(0.0, float(minimum_display_seconds)) if math.isfinite(float(minimum_display_seconds)) else 5.0
|
|
)
|
|
self.shown_at = time.monotonic()
|
|
self.startup_finished = False
|
|
self._progress_job: str | None = None
|
|
self._messages: Queue[tuple[str, object]] = Queue()
|
|
self.overrideredirect(True)
|
|
self.resizable(False, False)
|
|
self.configure(background="#0b1117")
|
|
self.canvas: tk.Canvas
|
|
self.status_item: int
|
|
self._build_ui()
|
|
self._center()
|
|
self._update_progress()
|
|
self.after(120, self._start)
|
|
|
|
def _build_ui(self) -> None:
|
|
width, height = 768, 512
|
|
self.geometry(f"{width}x{height}")
|
|
self.canvas = tk.Canvas(
|
|
self,
|
|
width=width,
|
|
height=height,
|
|
highlightthickness=0,
|
|
background="#0b1117",
|
|
)
|
|
self.canvas.pack(fill="both", expand=True)
|
|
image_path = Path(__file__).resolve().parent.parent / "assets" / "splash.png"
|
|
self.background_image = tk.PhotoImage(master=self, file=str(image_path)).subsample(2, 2)
|
|
self.background_item = self.canvas.create_image(0, 0, anchor="nw", image=self.background_image)
|
|
self.canvas.create_text(
|
|
width - 24,
|
|
24,
|
|
anchor="ne",
|
|
text=f"v{__version__}",
|
|
fill="#b8c5d0",
|
|
font=("TkFixedFont", 10),
|
|
)
|
|
self.status_item = self.canvas.create_text(
|
|
30,
|
|
450,
|
|
anchor="sw",
|
|
text="Initialisiere …",
|
|
fill="#e8edf2",
|
|
font=("TkFixedFont", 10),
|
|
)
|
|
self.progress = SplashProgressBar(self, width=708, height=16)
|
|
self.canvas.create_window(30, 476, anchor="nw", width=708, height=16, window=self.progress)
|
|
|
|
def _center(self) -> None:
|
|
self.update_idletasks()
|
|
width, height = self.winfo_width(), self.winfo_height()
|
|
monitor = self.monitor or preferred_monitor(self)
|
|
pointer_x, pointer_y = monitor.center
|
|
x, y = centered_position(
|
|
width=width,
|
|
height=height,
|
|
pointer_x=pointer_x,
|
|
pointer_y=pointer_y,
|
|
screen_x=monitor.x,
|
|
screen_y=monitor.y,
|
|
screen_width=monitor.width,
|
|
screen_height=monitor.height,
|
|
)
|
|
self.geometry(f"{width}x{height}+{x}+{y}")
|
|
|
|
def _start(self) -> None:
|
|
def worker() -> None:
|
|
try:
|
|
self._messages.put(("status", "Öffne Mitglieder-Store …"))
|
|
self.repository.initialize()
|
|
self._messages.put(("status", "Validiere Mitgliederakten …"))
|
|
errors = self.repository.validate()
|
|
self._messages.put(("status", "Baue Suchindex …"))
|
|
self.repository.list_members()
|
|
member_count = len(self.repository.list_member_ids())
|
|
findings = []
|
|
if self.run_housekeeper:
|
|
self._messages.put(("status", "Starte Hausmeister …"))
|
|
findings = Housekeeper(self.repository, self.housekeeper_settings).run(
|
|
member_delay=_member_delay_for_splash(
|
|
self.minimum_display_seconds,
|
|
member_count,
|
|
)
|
|
)
|
|
result = StartupResult(self.repository, errors, findings)
|
|
self._messages.put(("result", result))
|
|
except Exception as exc:
|
|
self._messages.put(("error", exc))
|
|
|
|
threading.Thread(target=worker, name="ccma-startup", daemon=True).start()
|
|
self.after(30, self._poll_messages)
|
|
|
|
def _poll_messages(self) -> None:
|
|
keep_polling = True
|
|
try:
|
|
while True:
|
|
kind, payload = self._messages.get_nowait()
|
|
if kind == "status":
|
|
self._set_status(str(payload))
|
|
elif kind == "result":
|
|
keep_polling = False
|
|
self._finish(payload) # type: ignore[arg-type]
|
|
elif kind == "error":
|
|
keep_polling = False
|
|
self._fail(payload) # type: ignore[arg-type]
|
|
except Empty:
|
|
pass
|
|
if keep_polling and self.winfo_exists():
|
|
self.after(30, self._poll_messages)
|
|
|
|
def _finish(self, result: StartupResult) -> None:
|
|
self.startup_finished = True
|
|
self._update_progress()
|
|
self._set_status(f"Bereit · {len(result.findings)} Vorgänge benötigen Aufmerksamkeit")
|
|
self._after_minimum(lambda: self._complete(result))
|
|
|
|
def _after_minimum(self, callback: Callable[[], None]) -> None:
|
|
remaining_ms = _remaining_minimum_ms(
|
|
self.shown_at,
|
|
self.minimum_display_seconds,
|
|
time.monotonic(),
|
|
)
|
|
self.after(remaining_ms, lambda: self._show_completed_progress(callback))
|
|
|
|
def _show_completed_progress(self, callback: Callable[[], None]) -> None:
|
|
self.startup_finished = True
|
|
self.progress.set(100)
|
|
self.after_idle(callback)
|
|
|
|
def _set_status(self, text: str) -> None:
|
|
self.canvas.itemconfigure(self.status_item, text=text)
|
|
|
|
def _complete(self, result: StartupResult) -> None:
|
|
self._cancel_progress_job()
|
|
self.destroy()
|
|
self.on_complete(result)
|
|
|
|
def _fail(self, error: Exception) -> None:
|
|
self.startup_finished = True
|
|
self._update_progress()
|
|
self._set_status("Start fehlgeschlagen")
|
|
self._after_minimum(lambda: self._complete_error(error))
|
|
|
|
def _complete_error(self, error: Exception) -> None:
|
|
self._cancel_progress_job()
|
|
self.destroy()
|
|
self.on_error(error)
|
|
|
|
def _update_progress(self) -> None:
|
|
if not self.winfo_exists():
|
|
return
|
|
elapsed = max(0.0, time.monotonic() - self.shown_at)
|
|
value = _progress_value(elapsed, self.minimum_display_seconds, self.startup_finished)
|
|
self.progress.set(value)
|
|
if value < 100 or not self.startup_finished:
|
|
self._cancel_progress_job()
|
|
self._progress_job = self.after(40, self._update_progress)
|
|
|
|
def _cancel_progress_job(self) -> None:
|
|
if not self._progress_job:
|
|
return
|
|
try:
|
|
self.after_cancel(self._progress_job)
|
|
except tk.TclError:
|
|
pass
|
|
self._progress_job = None
|
|
|
|
|
|
class SplashProgressBar(tk.Canvas):
|
|
TRACK_COLOR = "#071a29"
|
|
FILL_COLOR = "#2389c9"
|
|
BORDER_COLOR = "#aeb8c2"
|
|
BORDER_WIDTH = 2
|
|
|
|
def __init__(self, master: tk.Misc, *, width: int, height: int):
|
|
super().__init__(
|
|
master,
|
|
width=width,
|
|
height=height,
|
|
background=self.BORDER_COLOR,
|
|
borderwidth=0,
|
|
highlightthickness=0,
|
|
)
|
|
inset = self.BORDER_WIDTH
|
|
self._fill_left = inset
|
|
self._fill_top = inset
|
|
self._fill_width = max(1, width - 2 * inset)
|
|
self._fill_bottom = max(inset + 1, height - inset)
|
|
self.create_rectangle(
|
|
inset,
|
|
inset,
|
|
width - inset,
|
|
height - inset,
|
|
fill=self.TRACK_COLOR,
|
|
outline="",
|
|
)
|
|
self.fill_item = self.create_rectangle(
|
|
self._fill_left,
|
|
self._fill_top,
|
|
self._fill_left,
|
|
self._fill_bottom,
|
|
fill=self.FILL_COLOR,
|
|
outline="",
|
|
)
|
|
|
|
def set(self, percent: float) -> None:
|
|
value = min(100.0, max(0.0, float(percent)))
|
|
fill_right = self._fill_left + self._fill_width * value / 100.0
|
|
self.coords(
|
|
self.fill_item,
|
|
self._fill_left,
|
|
self._fill_top,
|
|
fill_right,
|
|
self._fill_bottom,
|
|
)
|
|
|
|
|
|
def centered_position(
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
pointer_x: int,
|
|
pointer_y: int,
|
|
screen_x: int,
|
|
screen_y: int,
|
|
screen_width: int,
|
|
screen_height: int,
|
|
) -> tuple[int, int]:
|
|
"""Center around the pointer and keep the complete window on the virtual desktop."""
|
|
maximum_x = screen_x + max(0, screen_width - width)
|
|
maximum_y = screen_y + max(0, screen_height - height)
|
|
x = min(max(pointer_x - width // 2, screen_x), maximum_x)
|
|
y = min(max(pointer_y - height // 2, screen_y), maximum_y)
|
|
return x, y
|
|
|
|
|
|
def _remaining_minimum_ms(started_at: float, minimum_seconds: float, now: float) -> int:
|
|
remaining = max(0.0, minimum_seconds - max(0.0, now - started_at))
|
|
return int(remaining * 1000 + 0.999)
|
|
|
|
|
|
def _progress_value(elapsed: float, minimum_seconds: float, startup_finished: bool) -> float:
|
|
if minimum_seconds <= 0:
|
|
return 100.0 if startup_finished else 5.0
|
|
elapsed_ratio = min(1.0, max(0.0, elapsed) / minimum_seconds)
|
|
if startup_finished:
|
|
return elapsed_ratio * 100.0
|
|
return elapsed_ratio * 95.0
|
|
|
|
|
|
def _member_delay_for_splash(minimum_seconds: float, member_count: int) -> float:
|
|
if minimum_seconds <= 0 or member_count <= 0:
|
|
return 0.0
|
|
return minimum_seconds / member_count
|