756 lines
29 KiB
Python
756 lines
29 KiB
Python
"""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
|
|
self._mini_dismissed = False
|
|
"""Set when the ✕ on the mini timer was clicked during a running interval.
|
|
|
|
`app._update_mini()` would otherwise bring it straight back up through
|
|
its "auto-show while running and the window is away" rule, making the
|
|
close button look broken. Cleared on the next explicit show and on the
|
|
next interval, so a dismissal only lasts for the one you closed it on.
|
|
"""
|
|
|
|
# ---- 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
|
|
|
|
@property
|
|
def mini_dismissed(self) -> bool:
|
|
"""For `app._update_mini()`, not QML: see `_mini_dismissed` above."""
|
|
return self._mini_dismissed
|
|
|
|
@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:
|
|
self._mini_dismissed = not visible
|
|
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:
|
|
if current.get("id") and current.get("id") != previous.get("id"):
|
|
# A new interval — pause/resume of the same one must not do this —
|
|
# gets its own chance to auto-show, even if the last one was closed.
|
|
self._mini_dismissed = False
|
|
|
|
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)
|