Ivan Juanes
247ce4c62e
Peak Monitor is a Linux system tray app that monitors two daily peak periods (09:00-12:00 and 14:00-18:00, Atlantic/Canary time) and sends desktop notifications when each period starts and ends. Files: - peak_monitor.py: Main application (GTK + AyatanaAppIndicator3) - peak-monitor.desktop: Desktop entry for autostart - ROADMAP.md: Development roadmap using Tau as the development framework - .env.example: Environment variable template (API keys) Tech stack: - Python 3 with GTK - Pillow for tray icons - libnotify for desktop notifications - Tau coding agent for development assistance
289 lines
10 KiB
Python
289 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Peak / Valley Time Monitor — system tray app for Linux.
|
||
Monitors two daily peak periods (Atlantic/Canary local time):
|
||
Morning: 09:00 – 12:00
|
||
Afternoon: 14:00 – 18:00
|
||
|
||
Sends a desktop notification when each period starts and ends.
|
||
Left-click the tray icon to see time remaining in the current block
|
||
(or time until the next peak block).
|
||
|
||
Requires: python3, python3-gi, gir1.2-gtk-3.0, gir1.2-ayatanaappindicator3-0.1,
|
||
python3-pil, libnotify-bin (notify-send)
|
||
"""
|
||
|
||
import datetime
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import gi
|
||
gi.require_version("Gtk", "3.0")
|
||
gi.require_version("AyatanaAppIndicator3", "0.1")
|
||
from gi.repository import Gtk, GLib, AyatanaAppIndicator3
|
||
|
||
from PIL import Image, ImageDraw
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Configuration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
TZ = ZoneInfo("Atlantic/Canary")
|
||
|
||
PEAKS = [
|
||
("Morning Peak", datetime.time(9, 0), datetime.time(12, 0)),
|
||
("Afternoon Peak", datetime.time(14, 0), datetime.time(18, 0)),
|
||
]
|
||
|
||
ICON_SIZE = 48 # px (system tray usually scales this down)
|
||
POLL_SECONDS = 15 # how often we re-check the clock
|
||
NOTIFY_WINDOW = datetime.timedelta(minutes=1) # grace window to send boundary notification
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _now() -> datetime.datetime:
|
||
return datetime.datetime.now(TZ)
|
||
|
||
|
||
def _make_icon_png(rgb: tuple[int, int, int]) -> str:
|
||
"""Create a plain coloured circle PNG in /tmp and return its path."""
|
||
path = f"/tmp/peak_monitor_{rgb[0]}_{rgb[1]}_{rgb[2]}.png"
|
||
if os.path.exists(path):
|
||
return path # reuse
|
||
|
||
img = Image.new("RGBA", (ICON_SIZE, ICON_SIZE), (0, 0, 0, 0))
|
||
draw = ImageDraw.Draw(img)
|
||
m = 4 # margin so the circle isn't clipped
|
||
draw.ellipse([m, m, ICON_SIZE - m, ICON_SIZE - m], fill=rgb)
|
||
img.save(path, "PNG")
|
||
return path
|
||
|
||
|
||
def _notify(title: str, body: str) -> None:
|
||
"""Fire a desktop notification via notify-send."""
|
||
subprocess.run(
|
||
["notify-send", title, body],
|
||
capture_output=True,
|
||
check=False,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Core logic
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _current_peak(t: datetime.time) -> tuple[bool, tuple | None]:
|
||
"""Return (is_peak, matching_peak_tuple_or_None)."""
|
||
for peak in PEAKS:
|
||
if peak[1] <= t < peak[2]:
|
||
return True, peak
|
||
return False, None
|
||
|
||
|
||
def _next_peak_start(now_dt: datetime.datetime) -> datetime.datetime | None:
|
||
"""Return the next peak start datetime (today or tomorrow), or None."""
|
||
today = now_dt.date()
|
||
candidates = []
|
||
for peak in PEAKS:
|
||
start_dt = datetime.datetime.combine(today, peak[1], tzinfo=TZ)
|
||
candidates.append(start_dt)
|
||
# also tomorrow's start (in case all today's starts have passed)
|
||
candidates.append(start_dt + datetime.timedelta(days=1))
|
||
future = [c for c in candidates if c > now_dt]
|
||
return min(future) if future else None
|
||
|
||
|
||
def _format_remaining(delta: datetime.timedelta) -> str:
|
||
"""Human-friendly timedelta string."""
|
||
total_min = int(delta.total_seconds() / 60)
|
||
if total_min <= 0:
|
||
return "now"
|
||
h, m = divmod(total_min, 60)
|
||
if h and m:
|
||
return f"{h}h {m}m"
|
||
if h:
|
||
return f"{h}h"
|
||
return f"{m}m"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GTK + AppIndicator application
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class PeakMonitor:
|
||
def __init__(self) -> None:
|
||
# -- icons ----------------------------------------------------------
|
||
self.icon_green = _make_icon_png((46, 204, 113)) # peak
|
||
self.icon_gray = _make_icon_png((149, 165, 166)) # valley
|
||
|
||
# -- notification dedup ---------------------------------------------
|
||
# Keys like "2025-06-15:morning_start"; we clear stale keys on date roll.
|
||
self._notified: set[str] = set()
|
||
self._last_date: datetime.date | None = None
|
||
|
||
# -- indicator (mute C-library deprecation warning) -----------------
|
||
_saved = os.dup(2)
|
||
os.close(2)
|
||
os.open(os.devnull, os.O_WRONLY)
|
||
try:
|
||
self.indicator = AyatanaAppIndicator3.Indicator.new(
|
||
"peak-monitor",
|
||
self.icon_gray,
|
||
AyatanaAppIndicator3.IndicatorCategory.APPLICATION_STATUS,
|
||
)
|
||
finally:
|
||
os.dup2(_saved, 2)
|
||
os.close(_saved)
|
||
|
||
self.indicator.set_status(AyatanaAppIndicator3.IndicatorStatus.ACTIVE)
|
||
|
||
# -- menu -----------------------------------------------------------
|
||
self.menu = Gtk.Menu()
|
||
|
||
self.status_item = Gtk.MenuItem(label="Starting…")
|
||
self.status_item.connect("activate", self._on_status_click)
|
||
self.menu.append(self.status_item)
|
||
|
||
self.menu.append(Gtk.SeparatorMenuItem())
|
||
|
||
quit_item = Gtk.MenuItem(label="Quit")
|
||
quit_item.connect("activate", self._on_quit)
|
||
self.menu.append(quit_item)
|
||
|
||
self.menu.show_all()
|
||
self.indicator.set_menu(self.menu)
|
||
|
||
# -- periodic timer -------------------------------------------------
|
||
GLib.idle_add(self._tick) # first tick immediately
|
||
GLib.timeout_add_seconds(POLL_SECONDS, self._tick)
|
||
|
||
# ------------------------------------------------------------------ update
|
||
|
||
def _tick(self) -> bool:
|
||
current = _now()
|
||
cur_time = current.time()
|
||
cur_date = current.date()
|
||
|
||
# --- roll daily notification dedup set ---------------------------
|
||
if self._last_date != cur_date:
|
||
self._notified.clear()
|
||
self._last_date = cur_date
|
||
|
||
in_peak, peak = _current_peak(cur_time)
|
||
|
||
# --- icon & tooltip ----------------------------------------------
|
||
if in_peak:
|
||
name, start, end = peak
|
||
end_dt = datetime.datetime.combine(cur_date, end, tzinfo=TZ)
|
||
remaining = end_dt - current
|
||
tooltip = f"☀ {name} — {_format_remaining(remaining)} left"
|
||
self.indicator.set_icon_full(self.icon_green, tooltip)
|
||
self.status_item.set_label(f"{name}: {_format_remaining(remaining)} left")
|
||
else:
|
||
next_start = _next_peak_start(current)
|
||
if next_start:
|
||
till = _format_remaining(next_start - current)
|
||
tooltip = f"🌙 Valley — next peak in {till}"
|
||
self.status_item.set_label(f"Valley — next peak in {till}")
|
||
else:
|
||
tooltip = "🌙 Valley"
|
||
self.status_item.set_label("Valley")
|
||
self.indicator.set_icon_full(self.icon_gray, tooltip)
|
||
|
||
# --- boundary notifications --------------------------------------
|
||
self._check_boundaries(current, cur_time, cur_date, in_peak, peak)
|
||
|
||
return True # keep GLib timeout alive
|
||
|
||
# ------------------------------------------------------------------ boundaries
|
||
|
||
def _check_boundaries(
|
||
self,
|
||
now_dt: datetime.datetime,
|
||
now_t: datetime.time,
|
||
today: datetime.date,
|
||
in_peak: bool,
|
||
peak: tuple | None,
|
||
) -> None:
|
||
"""Fire notify-send at period start/end if within the grace window."""
|
||
for name, start_t, end_t in PEAKS:
|
||
# --- period start ----------------------------------------------
|
||
key_start = f"{today.isoformat()}:{name}:start"
|
||
start_dt = datetime.datetime.combine(today, start_t, tzinfo=TZ)
|
||
if (
|
||
abs(now_dt - start_dt) <= NOTIFY_WINDOW
|
||
and key_start not in self._notified
|
||
):
|
||
_notify(
|
||
f"⏰ {name} started",
|
||
f"{start_t.strftime('%H:%M')} – {end_t.strftime('%H:%M')} "
|
||
f"(Atlantic/Canary)",
|
||
)
|
||
self._notified.add(key_start)
|
||
|
||
# --- period end ------------------------------------------------
|
||
key_end = f"{today.isoformat()}:{name}:end"
|
||
end_dt = datetime.datetime.combine(today, end_t, tzinfo=TZ)
|
||
if (
|
||
abs(now_dt - end_dt) <= NOTIFY_WINDOW
|
||
and key_end not in self._notified
|
||
):
|
||
_notify(
|
||
f"⏰ {name} ended",
|
||
f"Peak period {start_t.strftime('%H:%M')} – "
|
||
f"{end_t.strftime('%H:%M')} is over.",
|
||
)
|
||
self._notified.add(key_end)
|
||
|
||
# ------------------------------------------------------------------ actions
|
||
|
||
def _on_status_click(self, _widget) -> None:
|
||
"""Left-click on the status menu item: show a detailed notification."""
|
||
current = _now()
|
||
cur_time = current.time()
|
||
in_peak, peak = _current_peak(cur_time)
|
||
|
||
if in_peak:
|
||
name, start, end = peak
|
||
end_dt = datetime.datetime.combine(current.date(), end, tzinfo=TZ)
|
||
left = _format_remaining(end_dt - current)
|
||
_notify(
|
||
f"☀ Currently in {name}",
|
||
f"Started {start.strftime('%H:%M')}, "
|
||
f"ends {end.strftime('%H:%M')}\n"
|
||
f"⏳ {left} remaining",
|
||
)
|
||
else:
|
||
next_start = _next_peak_start(current)
|
||
if next_start:
|
||
till = _format_remaining(next_start - current)
|
||
ns_name = next_start.strftime("%A %H:%M")
|
||
_notify(
|
||
"🌙 Valley (off-peak)",
|
||
f"Next peak starts {ns_name}\n⏳ {till} until then",
|
||
)
|
||
else:
|
||
_notify("🌙 Valley", "No upcoming peak periods.")
|
||
|
||
def _on_quit(self, _widget) -> None:
|
||
Gtk.main_quit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Entry point
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def main() -> None:
|
||
PeakMonitor()
|
||
try:
|
||
Gtk.main()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|