Initial commit: Peak Monitor system tray app with Tau development framework
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
This commit is contained in:
commit
247ce4c62e
6 changed files with 753 additions and 0 deletions
289
peak_monitor.py
Normal file
289
peak_monitor.py
Normal file
|
|
@ -0,0 +1,289 @@
|
|||
#!/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()
|
||||
Reference in a new issue