timka123andClaude Opus 5
8e659cd315
The design of the full desktop app was still bad, and the reason it stayed bad is that it had no business existing. Five pages — Timer, Tasks, Blocker, Report, Settings — each a second view of the *server's* data: a second place to be wrong, a second place to style, and a second place to keep in step with a web app that already does lists, rules and reports properly in a browser, on any device. So it is gone, along with 2,100 lines of QML and the controller surface that fed it. What is left is what only a program on your machine can do, plus the two surfaces worth having in front of you: * **the tray menu — unchanged**, as asked. All fourteen entries, plus "Settings…". "Open Kadence" now brings the countdown up, because that is what it meant all along. * **the mini timer**, which is now the whole of this app's own interface and so finally carries the house style on its own: warm surface, hairline instead of shadow, a letterspaced eyebrow, the progress lane as a rule, and exactly one filled button — whichever action is the obvious next one. The countdown is set in the display serif on a **fixed advance**, because Fraunces has no tabular figures and ships no `tnum`: each glyph now sits in a box as wide as the widest digit, instead of the clock shifting sideways on every tick. The phone client documents the same trap; the desktop had it too and nobody had looked. It also says OFFLINE when the backend is away — the count and the clock are both the server's, and this is the only window left to say it in. * **a settings dialog**, because a fresh install has to be pointed at a backend somewhere. One column, hairline sections: the backend and its token, where the mini timer sits, what this machine enforces, and what it does to the session. It edits `desktop.json` and nothing else — with one honest exception, view-only mode, which is a server-side flag the other clients have to honour too. Structurally, `app.py` no longer loads a QML root at all: a plain `QQmlEngine` with the mini timer as a component, and the dialog built on demand. Everything that used to ask "is the main window visible?" is simpler for not having one — the mini timer's auto-show rule is now just "an interval is running", which is what the setting always claimed. The window-only controller surface goes with the pages: tasks and task writes, stats, the daily report, the settings document, the presets, the cycle and interruption counters, and their worker jobs and API calls. Two things were kept deliberately: the task list is still read once a minute for one fact — which task is pinned, so "Start focus" from the tray attaches what you pinned this morning — and `timerEnabled` now comes from the timer state, which already carries `timer_enabled`, rather than from a second fetch of a document this app can no longer edit. Also: `--settings` and a `Settings` D-Bus verb; toasts go to the tray's balloon now that the snackbar is gone; the launcher entry starts the agent and grew a Settings action; a systemd user unit is shipped rather than only written at runtime; and both READMEs say what this piece is. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
766 lines
30 KiB
Python
766 lines
30 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, placement
|
|
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()
|
|
blockerChanged = pyqtSignal()
|
|
tasksChanged = pyqtSignal()
|
|
connectionChanged = pyqtSignal()
|
|
configChanged = pyqtSignal()
|
|
miniChanged = pyqtSignal()
|
|
miniPlacementChanged = pyqtSignal()
|
|
toast = pyqtSignal(str)
|
|
showWindowRequested = pyqtSignal()
|
|
settingsRequested = 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._pinned_task_id: int | None = None
|
|
"""The task pinned in the web app, so "Start focus" can attach it.
|
|
|
|
All that is left of this app's interest in the task list: there is no
|
|
Tasks page any more, but starting an interval from the tray and having
|
|
it land on the thing you pinned this morning is worth one request a
|
|
minute. The list itself is never rendered here.
|
|
"""
|
|
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.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()
|
|
|
|
# A blocked app can be launched at any second, so the scan has its own
|
|
# clock rather than riding the policy poll — and the server is the one
|
|
# that says how fast (`blocker.app_poll_seconds`). It is cheap: a
|
|
# `/proc` walk and, at most, one signal per blocked process.
|
|
self._app_scan = QTimer(self)
|
|
self._app_scan.setInterval(self._app_scan_interval())
|
|
self._app_scan.timeout.connect(self._blocker.enforce_apps_now)
|
|
self._app_scan.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 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(bool, notify=stateChanged)
|
|
def timerEnabled(self) -> bool:
|
|
"""When False the app is in view-only mode — no timer controls.
|
|
|
|
Read from the timer *state*, which carries `timer_enabled`, rather than
|
|
from a second fetch of the settings document: this app no longer has a
|
|
page to edit that document in, and one less endpoint to poll is one less
|
|
thing to be out of date.
|
|
"""
|
|
if self._config.view_only:
|
|
return False
|
|
return bool(self._state.get("timer_enabled", True))
|
|
|
|
@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]:
|
|
"""Every hostname the policy wants blocked, server-expanded.
|
|
|
|
This is the *intent*: what the rules add up to. What is actually in this
|
|
machine's hosts file is [appliedDomains] — reporting this one as the
|
|
state was the page lying about a machine it had not touched.
|
|
"""
|
|
return [str(item) for item in self._policy.get("all_domains", [])]
|
|
|
|
@pyqtProperty("QVariantList", notify=blockerChanged)
|
|
def appliedDomains(self) -> list[str]:
|
|
"""What is really inside the managed region of `/etc/hosts` right now."""
|
|
return self._blocker.blocked_domains
|
|
|
|
@pyqtProperty("QVariantList", notify=blockerChanged)
|
|
def routedZones(self) -> list[str]:
|
|
"""DNS zones currently pointed at this machine's local resolver stub."""
|
|
return list(self._blocker.routed_zones)
|
|
|
|
@pyqtProperty("QVariantList", notify=blockerChanged)
|
|
def unroutableDomains(self) -> list[str]:
|
|
"""Masks no machine can enforce: no DNS zone can be routed for them."""
|
|
if not self._policy.get("active"):
|
|
return []
|
|
return list(self._blocker.unroutable)
|
|
|
|
@pyqtProperty("QVariantList", notify=blockerChanged)
|
|
def blockedApps(self) -> list[str]:
|
|
return [str(item) for item in self._policy.get("all_apps", [])]
|
|
|
|
@pyqtProperty("QVariantList", notify=blockerChanged)
|
|
def invalidExpansions(self) -> list[str]:
|
|
"""Subdomain expansions the server refused, e.g. `not a label`.
|
|
|
|
Worth a banner rather than a log line: a mistyped label silently reduces
|
|
every `*.site` mask to the bare domain, which looks like "blocking is on
|
|
and does nothing".
|
|
"""
|
|
return [str(item) for item in self._policy.get("invalid_expansions", [])]
|
|
|
|
@pyqtProperty(bool, notify=blockerChanged)
|
|
def helperInstalled(self) -> bool:
|
|
return self._blocker.helper_installed
|
|
|
|
@pyqtProperty(str, notify=blockerChanged)
|
|
def blockerError(self) -> str:
|
|
return self._blocker.error or ""
|
|
|
|
@pyqtProperty("QVariantList", notify=configChanged)
|
|
def miniCornerLabels(self) -> list[str]:
|
|
"""Where the mini timer may be put, as the settings dialog lists them.
|
|
|
|
From `placement.CORNER_LABELS` rather than written out again in QML:
|
|
the old settings page kept its own copy of this list, which is how a
|
|
corner gets added in one place and silently missing in the other.
|
|
"""
|
|
return list(placement.CORNER_LABELS.values())
|
|
|
|
@pyqtProperty(int, notify=configChanged)
|
|
def miniCornerIndex(self) -> int:
|
|
keys = list(placement.CORNER_LABELS)
|
|
try:
|
|
return keys.index(self._config.mini_timer_corner)
|
|
except ValueError:
|
|
return keys.index("bottom-right")
|
|
|
|
@pyqtSlot(int)
|
|
def setMiniCorner(self, index: int) -> None:
|
|
keys = list(placement.CORNER_LABELS)
|
|
if 0 <= index < len(keys):
|
|
self.setConfig("mini_timer_corner", keys[index])
|
|
|
|
@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") and self._pinned_task_id is not None:
|
|
args["task_id"] = self._pinned_task_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.stateChanged.emit()
|
|
self._patch_settings({"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()
|
|
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")
|
|
|
|
def _patch_settings(self, patch: dict[str, Any]) -> None:
|
|
"""The one write this app still makes to the server's settings document.
|
|
|
|
Only [setTimerEnabled] uses it — view-only mode is a server-side flag,
|
|
because the other clients have to honour it too. Everything else in that
|
|
document is edited in the web app.
|
|
"""
|
|
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:
|
|
# Takes the DNS route down with it, in the safe order.
|
|
self._blocker.shutdown()
|
|
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)
|
|
wanted = self._app_scan_interval()
|
|
if self._app_scan.interval() != wanted:
|
|
self._app_scan.setInterval(wanted)
|
|
self.blockerChanged.emit()
|
|
|
|
def _app_scan_interval(self) -> int:
|
|
"""Milliseconds between `/proc` scans, as the server asked.
|
|
|
|
Clamped because this comes from a settings document a human edits: a
|
|
zero would spin, and a minute would let a game run for a minute.
|
|
"""
|
|
seconds = self._policy.get("poll_seconds") or self._config.poll_seconds
|
|
try:
|
|
seconds = int(seconds)
|
|
except (TypeError, ValueError):
|
|
seconds = self._config.poll_seconds
|
|
return max(1, min(30, seconds)) * 1000
|
|
|
|
def _on_tasks(self, payload: dict[str, Any]) -> None:
|
|
pinned = next(
|
|
(task for task in payload.get("tasks", []) if task.get("pinned")),
|
|
None,
|
|
)
|
|
self._pinned_task_id = int(pinned["id"]) if pinned else None
|
|
self.tasksChanged.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, the hosts file and the DNS route."""
|
|
self._tick.stop()
|
|
self._idle_check.stop()
|
|
self._app_scan.stop()
|
|
self._screensaver.release()
|
|
self._dnd.release()
|
|
self._launcher.clear()
|
|
try:
|
|
self._blocker.shutdown()
|
|
except Exception: # pragma: no cover
|
|
log.exception("could not clear the block on exit")
|
|
self._thread.quit()
|
|
self._thread.wait(2000)
|