timka123/kadence
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
77b26757bf29a8a1497933708767526ba02f1a68
kadence/desktop/kadence_desktop/controller.py
T
timka123 77b26757bf refactor: enhance TimerPage layout and functionality; update theme and tokens for improved design consistency 
- Refactored TimerPage component to improve layout and user experience.
- Updated theme configuration to utilize design tokens for better maintainability.
- Introduced new tokens for colors and typography to align with design principles.
- Simplified state management and improved timer functionality.
- Enhanced accessibility and responsiveness of UI elements.
2026-10-18 23:57:36 +03:00

737 lines
28 KiB
Python
Raw Blame History

"""The object QML talks to, and the place the desktop integrations are wired up.
It holds the last known state from the server, ticks the countdown locally
between polls, and reacts to the *transitions* — a focus interval starting is
what inhibits the screensaver, silences notifications, pauses your music and
turns the blocker on; it ending is what gives all of that back.
"""
from __future__ import annotations
import logging
from typing import Any
from PyQt6.QtCore import (
QObject,
QThread,
QTimer,
QUrl,
pyqtProperty,
pyqtSignal,
pyqtSlot,
)
from PyQt6.QtGui import QDesktopServices
from kadence_desktop.api import Api
from kadence_desktop.blocking.manager import BlockerManager
from kadence_desktop.config import DesktopConfig
from kadence_desktop.integrations import autostart, keep_above
from kadence_desktop.integrations.inhibit import DoNotDisturb, ScreensaverInhibitor
from kadence_desktop.integrations.launcher import LauncherEntry
from kadence_desktop.integrations.mpris import MediaController
from kadence_desktop.integrations.notifications import URGENCY_CRITICAL, Notifier
from kadence_desktop.integrations.session import SessionWatcher
from kadence_desktop.worker import ApiWorker
log = logging.getLogger(__name__)
KIND_LABELS = {"work": "Focus", "short_break": "Short break", "long_break": "Long break"}
def clock(seconds: int) -> str:
seconds = max(0, int(seconds))
return f"{seconds // 60:02d}:{seconds % 60:02d}"
class Controller(QObject):
# ---- notifications to QML ---------------------------------------------
stateChanged = pyqtSignal()
tasksChanged = pyqtSignal()
statsChanged = pyqtSignal()
reportChanged = pyqtSignal()
settingsChanged = pyqtSignal()
blockerChanged = pyqtSignal()
connectionChanged = pyqtSignal()
configChanged = pyqtSignal()
miniChanged = pyqtSignal()
miniPlacementChanged = pyqtSignal()
toast = pyqtSignal(str)
showWindowRequested = pyqtSignal()
def __init__(self, config: DesktopConfig, parent: QObject | None = None) -> None:
super().__init__(parent)
self._config = config
self._api = Api(config.server_url, config.api_token)
# ---- state mirrors --------------------------------------------------
self._state: dict[str, Any] = {}
self._policy: dict[str, Any] = {}
self._tasks: list[dict[str, Any]] = []
self._task_counts: dict[str, Any] = {}
self._stats: dict[str, Any] = {}
self._report: dict[str, Any] = {}
self._settings: dict[str, Any] = {}
self._connected = False
self._connection_error = ""
self._remaining = 0
self._last_session_id: int | None = None
self._last_kind = ""
self._last_status = ""
self._completing = False
self._paused_by_lock = False
self._mini_visible = False
# ---- system integrations -------------------------------------------
self._notifier = Notifier()
self._screensaver = ScreensaverInhibitor()
self._dnd = DoNotDisturb()
self._launcher = LauncherEntry()
self._media = MediaController()
self._session = SessionWatcher(self)
self._session.locked.connect(self._on_session_locked)
self._session.unlocked.connect(self._on_session_unlocked)
self._session.sleeping.connect(self._on_session_locked)
self._blocker = BlockerManager(
helper_path=config.helper_path,
report=self._report_block,
notify=self._blocker_notice,
)
self._blocker.enforce_hosts = config.block_hosts_file
self._blocker.enforce_apps = config.block_apps
# ---- network thread -------------------------------------------------
self._thread = QThread(self)
self._worker = ApiWorker(self._api, config.poll_seconds)
self._worker.moveToThread(self._thread)
self._thread.started.connect(self._worker.start)
self._worker.stateChanged.connect(self._on_state)
self._worker.policyChanged.connect(self._on_policy)
self._worker.tasksChanged.connect(self._on_tasks)
self._worker.statsChanged.connect(self._on_stats)
self._worker.reportChanged.connect(self._on_report)
self._worker.settingsChanged.connect(self._on_settings)
self._worker.connectionChanged.connect(self._on_connection)
self._worker.commandFailed.connect(self._on_command_failed)
self._worker.commandDone.connect(self._on_command_done)
self._thread.start()
# ---- local clock ----------------------------------------------------
self._tick = QTimer(self)
self._tick.setInterval(1000)
self._tick.timeout.connect(self._on_tick)
self._tick.start()
# A Wayland compositor ignores "stay on top", so ask KWin properly.
self.ensure_keep_above()
self._idle_check = QTimer(self)
self._idle_check.setInterval(30_000)
self._idle_check.timeout.connect(self._check_idle)
self._idle_check.start()
# ======================================================================
# Properties for QML
# ======================================================================
@pyqtProperty(bool, notify=connectionChanged)
def connected(self) -> bool:
return self._connected
@pyqtProperty(str, notify=connectionChanged)
def connectionError(self) -> str:
return self._connection_error
@pyqtProperty(bool, notify=stateChanged)
def running(self) -> bool:
return bool(self._state.get("current"))
@pyqtProperty(str, notify=stateChanged)
def kind(self) -> str:
current = self._state.get("current")
return str(current["kind"]) if current else str(self._state.get("next_kind", "work"))
@pyqtProperty(str, notify=stateChanged)
def kindLabel(self) -> str:
return KIND_LABELS.get(self.kind, self.kind)
@pyqtProperty(str, notify=stateChanged)
def status(self) -> str:
current = self._state.get("current")
return str(current["status"]) if current else "idle"
@pyqtProperty(int, notify=stateChanged)
def remainingSeconds(self) -> int:
if self._state.get("current"):
return int(self._remaining)
durations = self._state.get("durations") or {}
return int(durations.get(self.kind, 25)) * 60
@pyqtProperty(int, notify=stateChanged)
def plannedSeconds(self) -> int:
current = self._state.get("current")
if current:
return int(current["planned_seconds"])
durations = self._state.get("durations") or {}
return int(durations.get(self.kind, 25)) * 60
@pyqtProperty(float, notify=stateChanged)
def progress(self) -> float:
planned = self.plannedSeconds
if planned <= 0:
return 0.0
return max(0.0, min(1.0, 1 - self.remainingSeconds / planned))
@pyqtProperty(str, notify=stateChanged)
def clockText(self) -> str:
return clock(self.remainingSeconds)
@pyqtProperty(str, notify=stateChanged)
def taskTitle(self) -> str:
current = self._state.get("current") or {}
task = current.get("task") or {}
return str(task.get("title", ""))
@pyqtProperty(int, notify=stateChanged)
def taskId(self) -> int:
for task in self._tasks:
current = self._state.get("current") or {}
attached = current.get("task") or {}
if attached and task.get("external_id") == attached.get("external_id"):
return int(task.get("id", -1))
return -1
@pyqtProperty(int, notify=stateChanged)
def interruptions(self) -> int:
current = self._state.get("current") or {}
return int(current.get("interruptions", 0))
@pyqtProperty(int, notify=stateChanged)
def completedToday(self) -> int:
return int(self._state.get("completed_today", 0))
@pyqtProperty(int, notify=stateChanged)
def goal(self) -> int:
return int(self._state.get("goal", 8))
@pyqtProperty(int, notify=stateChanged)
def cyclePosition(self) -> int:
return int(self._state.get("cycle_position", 1))
@pyqtProperty(int, notify=stateChanged)
def cycleLength(self) -> int:
return int(self._state.get("cycles_before_long_break", 4))
@pyqtProperty(str, notify=stateChanged)
def nextKindLabel(self) -> str:
return KIND_LABELS.get(str(self._state.get("next_kind", "work")), "Focus")
@pyqtProperty(bool, notify=settingsChanged)
def timerEnabled(self) -> bool:
"""When False the app is in view-only mode — no timer controls."""
if self._config.view_only:
return False
pomodoro = self._settings.get("pomodoro", {})
return bool(pomodoro.get("timer_enabled", True))
@pyqtProperty("QVariantList", notify=tasksChanged)
def tasks(self) -> list[dict[str, Any]]:
return self._tasks
@pyqtProperty("QVariantMap", notify=tasksChanged)
def taskCounts(self) -> dict[str, Any]:
return self._task_counts
@pyqtProperty("QVariantMap", notify=statsChanged)
def stats(self) -> dict[str, Any]:
return self._stats
@pyqtProperty("QVariantMap", notify=reportChanged)
def report(self) -> dict[str, Any]:
return self._report
@pyqtProperty("QVariantMap", notify=settingsChanged)
def settings(self) -> dict[str, Any]:
return self._settings
@pyqtProperty(bool, notify=blockerChanged)
def blockerActive(self) -> bool:
return bool(self._policy.get("active"))
@pyqtProperty(str, notify=blockerChanged)
def blockerReason(self) -> str:
return str(self._policy.get("reason", ""))
@pyqtProperty("QVariantList", notify=blockerChanged)
def blockedDomains(self) -> list[str]:
return [str(item) for item in self._policy.get("all_domains", [])]
@pyqtProperty("QVariantList", notify=blockerChanged)
def blockedApps(self) -> list[str]:
return [str(item) for item in self._policy.get("all_apps", [])]
@pyqtProperty(bool, notify=blockerChanged)
def helperInstalled(self) -> bool:
return self._blocker.helper_installed
@pyqtProperty(str, notify=blockerChanged)
def blockerError(self) -> str:
return self._blocker.last_error or ""
@pyqtProperty(bool, notify=miniChanged)
def miniVisible(self) -> bool:
"""Whether the floating mini timer is pinned open by the user."""
return self._mini_visible
@pyqtProperty("QVariantMap", notify=configChanged)
def config(self) -> dict[str, Any]:
from dataclasses import asdict
data = asdict(self._config)
data["autostart"] = autostart.autostart_enabled()
data["service_enabled"] = autostart.service_enabled()
data["keep_above_supported"] = keep_above.supported()
data["keep_above_installed"] = keep_above.installed()
return data
# ======================================================================
# Commands from QML (and from the tray, and from D-Bus)
# ======================================================================
def _send(self, name: str, args: dict[str, Any] | None = None) -> None:
from PyQt6.QtCore import Q_ARG, QMetaObject, Qt
QMetaObject.invokeMethod(
self._worker,
"command",
Qt.ConnectionType.QueuedConnection,
Q_ARG(str, name),
Q_ARG(dict, args or {}),
)
@pyqtSlot()
@pyqtSlot(int)
def startWork(self, minutes: int = 0) -> None:
if not self.timerEnabled:
return
args: dict[str, Any] = {"kind": "work"}
if minutes:
args["minutes"] = minutes
current_task = self._state.get("current") or {}
if not current_task.get("task"):
pinned = next((task for task in self._tasks if task.get("pinned")), None)
if pinned:
args["task_id"] = pinned["id"]
self._send("start", args)
@pyqtSlot()
def startBreak(self) -> None:
if not self.timerEnabled:
return
kind = str(self._state.get("next_kind", "short_break"))
self._send("start", {"kind": "long_break" if kind == "long_break" else "short_break"})
@pyqtSlot()
def toggle(self) -> None:
"""One verb for a global hotkey: start → pause → resume."""
if not self.timerEnabled:
return
if not self.running:
self.startWork()
elif self.status == "paused":
self.resume()
else:
self.pause()
@pyqtSlot(bool)
def setTimerEnabled(self, enabled: bool) -> None:
"""Toggle view-only mode on or off."""
self._config.update(view_only=not enabled)
self.configChanged.emit()
self.settingsChanged.emit()
self.patchSettings({"pomodoro": {"timer_enabled": enabled}})
@pyqtSlot()
def pause(self) -> None:
self._send("pause")
@pyqtSlot()
def resume(self) -> None:
self._send("resume")
@pyqtSlot()
def complete(self) -> None:
self._send("complete", {"auto_next": True})
@pyqtSlot()
@pyqtSlot(str)
def abandon(self, reason: str = "") -> None:
self._send("abandon", {"reason": reason or None})
@pyqtSlot()
@pyqtSlot(int)
def extend(self, minutes: int = 5) -> None:
self._send("extend", {"minutes": minutes})
@pyqtSlot()
def interrupt(self) -> None:
self._send("interrupt")
@pyqtSlot(int)
def attachTask(self, task_id: int) -> None:
self._send("attach_task", {"task_id": task_id if task_id >= 0 else None})
@pyqtSlot(int, bool)
def setTaskPinned(self, task_id: int, pinned: bool) -> None:
"""Pinning is the only task edit this app makes — closing one is not."""
self._send("patch_task", {"task_id": task_id, "patch": {"pinned": pinned}})
@pyqtSlot(str, str)
def createTask(self, title: str, due_date: str) -> None:
self._send("create_task", {"title": title, "due_date": due_date or None})
@pyqtSlot()
def refreshSources(self) -> None:
self.toast.emit("Pulling Jira, Todoist, Google Tasks…")
self._send("refresh_sources")
@pyqtSlot()
def sendReport(self) -> None:
self._send("send_report")
@pyqtSlot()
def loadReport(self) -> None:
from PyQt6.QtCore import QMetaObject, Qt
QMetaObject.invokeMethod(self._worker, "refresh_report", Qt.ConnectionType.QueuedConnection)
@pyqtSlot()
def loadSettings(self) -> None:
from PyQt6.QtCore import QMetaObject, Qt
QMetaObject.invokeMethod(self._worker, "refresh_settings", Qt.ConnectionType.QueuedConnection)
@pyqtSlot("QVariantMap")
def patchSettings(self, patch: dict[str, Any]) -> None:
self._send("patch_settings", {"patch": patch})
@pyqtSlot(str, "QVariant")
def setConfig(self, key: str, value: Any) -> None:
"""Change a desktop-local setting and apply it immediately."""
if key == "autostart":
autostart.set_autostart(bool(value), minimized=self._config.start_minimized)
elif key == "service_enabled":
if value:
self.toast.emit(autostart.install_service(enable=True))
else:
autostart.remove_service()
self.toast.emit("systemd user service removed")
else:
self._config.update(**{key: value})
if key in ("server_url", "api_token"):
from PyQt6.QtCore import Q_ARG, QMetaObject, Qt
QMetaObject.invokeMethod(
self._worker,
"configure",
Qt.ConnectionType.QueuedConnection,
Q_ARG(str, self._config.server_url),
Q_ARG(str, self._config.api_token),
)
elif key == "poll_seconds":
from PyQt6.QtCore import Q_ARG, QMetaObject, Qt
QMetaObject.invokeMethod(
self._worker,
"set_interval",
Qt.ConnectionType.QueuedConnection,
Q_ARG(int, int(value)),
)
elif key == "block_hosts_file":
self._blocker.enforce_hosts = bool(value)
elif key == "block_apps":
self._blocker.enforce_apps = bool(value)
elif key == "mini_timer_keep_above":
ok, message = keep_above.apply() if value else keep_above.remove()
self.toast.emit(message)
if not ok:
self._config.update(mini_timer_keep_above=not value)
elif key in ("mini_timer_corner", "mini_timer_margin"):
self.miniPlacementChanged.emit()
elif key == "enforce_blocking" and not value:
self._blocker.release()
self.configChanged.emit()
@pyqtSlot()
@pyqtSlot(bool)
def setMiniVisible(self, visible: bool = True) -> None:
if visible == self._mini_visible:
return
self._mini_visible = visible
self.miniChanged.emit()
@pyqtSlot(result=str)
def toggleMini(self) -> str:
"""Tray menu, Ctrl+M and the D-Bus verb all land here."""
if not self._config.mini_timer:
self._config.update(mini_timer=True)
self.configChanged.emit()
self.setMiniVisible(not self._mini_visible)
if self._mini_visible:
self.ensure_keep_above()
return "shown" if self._mini_visible else "hidden"
def ensure_keep_above(
self, position: tuple[int, int] | None = None, *, announce: bool = False
) -> None:
"""Install or refresh the KWin rule, including where to place the window.
A no-op when the rule already says exactly this, so it can be called
freely — it only ever touches KWin when something really changed.
"""
config = self._config
if not (config.mini_timer and config.mini_timer_keep_above):
return
if not keep_above.supported() or keep_above.up_to_date(position):
return
ok, message = keep_above.apply(position)
log.info("keep-above rule: %s", message)
if announce or not ok:
self.toast.emit(message)
self.configChanged.emit()
@pyqtSlot(int, int)
def saveMiniPosition(self, x: int, y: int) -> None:
"""Remember where it was dragged to — only in "remember" mode.
In corner mode the app places the window itself, and recording that
placement would quietly overwrite the position you had dragged it to.
"""
if self._config.mini_timer_corner != "remember":
return
self._config.update(mini_timer_x=int(x), mini_timer_y=int(y))
@pyqtSlot()
def openWebUi(self) -> None:
QDesktopServices.openUrl(QUrl(self._config.server_url))
@pyqtSlot(str)
def openUrl(self, url: str) -> None:
if url:
QDesktopServices.openUrl(QUrl(url))
@pyqtSlot(result=str)
def statusLine(self) -> str:
"""One line for the tray tooltip and the D-Bus `Status` method."""
if not self._connected:
return "Kadence — backend unreachable"
if not self.timerEnabled:
return f"Kadence — view mode · {self.completedToday}/{self.goal} today"
if not self.running:
return f"Kadence — idle · {self.completedToday}/{self.goal} today"
return (
f"{self.kindLabel} {clock(self.remainingSeconds)}"
+ (f" · {self.taskTitle}" if self.taskTitle else "")
+ f" · {self.completedToday}/{self.goal} today"
)
# ======================================================================
# Worker signals
# ======================================================================
def _on_state(self, state: dict[str, Any]) -> None:
previous_current = self._state.get("current") or {}
self._state = state
current = state.get("current") or {}
self._remaining = int(current.get("remaining_seconds", 0))
session_id = current.get("id")
kind = str(current.get("kind", ""))
status = str(current.get("status", "idle"))
if session_id != self._last_session_id or kind != self._last_kind or status != self._last_status:
self._on_transition(previous_current, current)
self._last_session_id = session_id
self._last_kind = kind
self._last_status = status
self._completing = False
self.stateChanged.emit()
self._update_launcher()
def _on_policy(self, policy: dict[str, Any]) -> None:
self._policy = policy
if self._config.enforce_blocking:
self._blocker.sync(policy)
self.blockerChanged.emit()
def _on_tasks(self, payload: dict[str, Any]) -> None:
self._tasks = list(payload.get("tasks", []))
self._task_counts = dict(payload.get("counts", {}))
self.tasksChanged.emit()
def _on_stats(self, payload: dict[str, Any]) -> None:
self._stats = payload
self.statsChanged.emit()
def _on_report(self, payload: dict[str, Any]) -> None:
self._report = payload
self.reportChanged.emit()
def _on_settings(self, payload: dict[str, Any]) -> None:
self._settings = payload.get("settings", payload)
self.settingsChanged.emit()
def _on_connection(self, connected: bool, detail: str) -> None:
self._connected = connected
self._connection_error = detail
self.connectionChanged.emit()
if not connected:
self.toast.emit(detail or "Backend unreachable")
def _on_command_failed(self, name: str, detail: str) -> None:
log.warning("%s failed: %s", name, detail)
self.toast.emit(detail)
def _on_command_done(self, name: str, result: dict[str, Any]) -> None:
if name == "complete":
finished = result.get("finished") or {}
self._announce_finished(str(finished.get("kind", "work")), result)
elif name == "send_report":
self.toast.emit("Daily report sent" if result.get("sent") else "Report not sent — check Telegram settings")
elif name == "refresh_sources":
failed = [item for item in result.get("sources", []) if not item.get("ok")]
closed = result.get("closed", 0)
self.toast.emit(
f"{len(failed)} source(s) failed"
if failed
else f"{result.get('total', 0)} open · {closed} closed since last refresh"
)
elif name == "refresh_sources" and result.get("closed"):
self.toast.emit(f"{result['closed']} task(s) closed in your tools")
# ======================================================================
# Transitions → the system integrations
# ======================================================================
def _on_transition(self, previous: dict[str, Any], current: dict[str, Any]) -> None:
focusing = current.get("kind") == "work" and current.get("status") == "running"
if focusing:
if self._config.inhibit_screensaver:
self._screensaver.hold()
if self._config.do_not_disturb:
self._dnd.hold()
if self._config.pause_media and previous.get("kind") != "work":
self._media.pause_all()
else:
self._screensaver.release()
self._dnd.release()
if self._config.pause_media and previous.get("kind") == "work":
self._media.resume_paused()
def _announce_finished(self, kind: str, result: dict[str, Any]) -> None:
auto = result.get("auto_started") or {}
done = int(result.get("completed_today", self.completedToday))
goal = int(result.get("goal", self.goal))
if kind == "work":
summary = "Pomodoro done 🍅"
body = f"{done}/{goal} today."
if auto:
body += f" {KIND_LABELS.get(str(auto.get('kind')), 'Break')} started automatically."
actions = (
{
"break": ("Start break", self.startBreak),
"again": ("Another pomodoro", lambda: self.startWork()),
}
if not auto
else {"skip": ("Skip the break", lambda: self.startWork())}
)
else:
summary = "Break over ☕"
body = "Back to it — one more interval."
actions = {"work": ("Start focus", lambda: self.startWork()), "five": ("+5 min break", lambda: self.extend(5))}
if self._config.notifications:
self._notifier.notify(
summary,
body,
actions=actions if self._config.notification_actions else None,
urgency=URGENCY_CRITICAL if kind == "work" else 1,
timeout_ms=0 if kind != "work" else 12000,
)
if result.get("celebration"):
self.toast.emit("Milestone reached — Telegram got a cheer 🎉")
def _update_launcher(self) -> None:
if not self._config.taskbar_progress:
return
if self.running:
self._launcher.update(progress=self.progress, count=self.completedToday or None)
else:
self._launcher.update(progress=None, count=self.completedToday or None)
# ---- local clock -------------------------------------------------------
def _on_tick(self) -> None:
if not self.running or self.status == "paused":
return
if self._remaining > 0:
self._remaining -= 1
self.stateChanged.emit()
if self._remaining % 5 == 0:
self._update_launcher()
return
if not self._completing:
# The countdown reached zero: the server books the interval, and its
# reply decides what happens next (auto-break, or nothing).
self._completing = True
self.complete()
def _check_idle(self) -> None:
limit = self._config.pause_on_idle_minutes
if not limit or not self.running or self.status != "running" or self.kind != "work":
return
idle = self._session.idle_seconds()
if idle >= 0 and idle >= limit * 60:
self.pause()
self.toast.emit(f"Paused — no input for {limit} minutes")
if self._config.notifications:
self._notifier.notify("Timer paused", f"You have been away for {limit} minutes.")
def _on_session_locked(self) -> None:
if not self._config.pause_on_lock:
return
if self.running and self.status == "running":
self._paused_by_lock = True
self.pause()
def _on_session_unlocked(self) -> None:
if self._paused_by_lock and self.running and self.status == "paused":
self.resume()
if self._config.notifications:
self._notifier.notify("Welcome back", "Your interval is running again.")
self._paused_by_lock = False
# ---- blocker callbacks -------------------------------------------------
def _report_block(self, kind: str, target: str, action: str, details: str | None) -> None:
self._send("report_block", {"kind": kind, "target": target, "action": action, "details": details})
def _blocker_notice(self, summary: str, body: str) -> None:
if self._config.notifications:
self._notifier.notify(summary, body, urgency=URGENCY_CRITICAL, timeout_ms=6000)
self.toast.emit(body)
# ---- shutdown ----------------------------------------------------------
def shutdown(self) -> None:
"""Hand everything back: inhibits, DND, and the hosts file."""
self._tick.stop()
self._idle_check.stop()
self._screensaver.release()
self._dnd.release()
self._launcher.clear()
try:
self._blocker.release()
except Exception: # pragma: no cover
log.exception("could not clear the block on exit")
self._thread.quit()
self._thread.wait(2000)
Reference in new issue View Git Blame Copy Permalink