diff --git a/plugins.json b/plugins.json index 37aed42..3edc7d3 100644 --- a/plugins.json +++ b/plugins.json @@ -1,6 +1,6 @@ { "version": "1.0.0", - "last_updated": "2026-07-28", + "last_updated": "2026-07-29", "plugins": [ { "id": "cricket-scoreboard", @@ -1120,6 +1120,31 @@ "screenshot": "https://raw.githubusercontent.com/ChuckBuilds/ledmatrix-plugins/main/plugins/incoming-packages/assets/screenshot.png", "latest_version": "1.0.0", "icon": "fas fa-box" + }, + { + "id": "pomodoro-timer", + "name": "Pomodoro Timer", + "description": "A configurable Pomodoro focus/break timer for your matrix. Set the work and break lengths, then start, pause, skip, or reset it over MQTT — with Home Assistant auto-discovery so the whole timer shows up as a device with no YAML.", + "author": "ChuckBuilds", + "category": "productivity", + "tags": [ + "pomodoro", + "timer", + "focus", + "productivity", + "mqtt", + "home-assistant" + ], + "repo": "https://github.com/ChuckBuilds/ledmatrix-plugins", + "branch": "main", + "plugin_path": "plugins/pomodoro-timer", + "stars": 0, + "downloads": 0, + "last_updated": "2026-07-29", + "verified": true, + "screenshot": "", + "latest_version": "1.3.3", + "icon": "fa-hourglass-half" } ] } diff --git a/plugins/pomodoro-timer/LICENSE b/plugins/pomodoro-timer/LICENSE new file mode 100644 index 0000000..e653a0c --- /dev/null +++ b/plugins/pomodoro-timer/LICENSE @@ -0,0 +1,17 @@ +GNU GENERAL PUBLIC LICENSE +Version 3, 29 June 2007 + +Copyright (C) 2025 LEDMatrix Team + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . diff --git a/plugins/pomodoro-timer/README.md b/plugins/pomodoro-timer/README.md new file mode 100644 index 0000000..9d5a0a1 --- /dev/null +++ b/plugins/pomodoro-timer/README.md @@ -0,0 +1,496 @@ +# Pomodoro Timer + +A focus/break timer for your LED matrix, driven over MQTT. Set the work and +break lengths however you like, then start, pause, skip, or reset the timer from +Home Assistant, a dashboard button, a voice assistant, or a one-line +`mosquitto_pub`. The matrix shows the phase you're in, the countdown, how far +through the phase you are, and how many sessions you've banked toward the long +break. + +Home Assistant MQTT auto-discovery is on by default, so the whole timer — +switch, buttons, duration boxes, and sensors — appears as a device with no YAML. + +![A work session on a 128×64 panel: the task name across the top, a large +seven-segment countdown, session pips along the bottom, and the burndown ring +part-drained around the edge](assets/hero.png) + +*A work session on a 128×64 panel. Every image in this README is real plugin +output, rendered at the true panel size and then scaled up so the pixels stay +pixels — nothing is a mockup.* + +--- + +## Table of Contents + +1. [How It Works](#how-it-works) +2. [What's On Screen](#whats-on-screen) +3. [Quick Start](#quick-start) +4. [Plugin Configuration](#plugin-configuration) +5. [MQTT Reference](#mqtt-reference) + - [Commands](#commands) + - [JSON Payloads](#json-payloads) + - [Published Topics](#published-topics) +6. [Home Assistant Setup](#home-assistant-setup) + - [Auto-Discovery](#auto-discovery-recommended) + - [What You Get](#what-you-get) + - [Automation Examples](#automation-examples) + - [Manual YAML](#manual-yaml-if-you-turn-discovery-off) +7. [Testing Without Home Assistant](#testing-without-home-assistant) +8. [Troubleshooting](#troubleshooting) + +--- + +## How It Works + +```text +Home Assistant ──MQTT──► ledmatrix/pomodoro/set ──► timer starts / pauses / skips +LED matrix ──MQTT──► ledmatrix/pomodoro/state ──► HA switch stays in sync +LED matrix ──MQTT──► ledmatrix/pomodoro/event ──► "work_complete" fires your automations +``` + +The timer is a classic Pomodoro cycle: + +```text +work ─► short break ─► work ─► short break ─► work ─► short break ─► work ─► LONG break ─► … + (4 sessions per set, configurable) +``` + +It runs in its own thread, so it keeps counting and keeps publishing state +whether or not it's the screen currently on the panel. By default it **holds the +display** for as long as a session is active, then hands the panel back to your +normal rotation when the timer goes idle. + +--- + +## What's On Screen + +```text +▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▔▔▔▔▔▔▔▔ ← burndown ring: lit = time left in the phase +▏ Editing specs ●●◉○ ▕ ← task (or phase label), and one pip per session +▏ ┌─┐ ┌─┐ ┌─┐ ┌─┐ ▕ +▏ │17│:│42│ ▕ ← countdown in seven-segment digits +▏ └─┘ └─┘ ▕ +▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ +``` + +Each state has its own colour, so you can read the timer from across the room +without reading a single word: + +![The six on-screen states on a 128×32 panel — idle in grey, work in red, +paused in amber, short break in green, long break in blue, and a work session +showing a task name instead of the phase label](assets/phases.png) + +The countdown is drawn as **seven-segment digits**, sized to whatever space the +panel has left rather than picked from a font, with a heavy stroke so it stays +readable from across a room. The phase label is held back a little so the eye +lands on the time first. + +If you want the full clock-radio look, **Show Unlit Segments** faintly lights the +dark segments the way real hardware does. It's off by default because it costs +real legibility — every digit gains a faint `8` behind it, and a `15` starts to +read as an `85`. + +The **task label** is whatever you're working on. Set it from Home Assistant (or +any MQTT client) and it takes the label row — the phase is already carried by the +colour, so it doesn't need the words too. Leave it empty and the phase name shows +instead. + +The **burndown ring** is the default indicator: the whole border is lit when a +phase starts and drains clockwise from the top-left, with a brighter head pixel +leading the way. It costs no interior space, which is what lets the digits be as +large as they are. If you'd rather have the border back, `bar` and `segments` put +the indicator along the bottom edge instead, and `none` hides it. + +Each element can be turned off, and each colour can be changed: + +![The appearance options side by side on a 128×32 panel — the four burndown +styles, unlit segments on, pixel digits, the calm colour theme, and the +desaturated paused style](assets/options.png) + +The layout adapts to the panel: stacked on 64×32, 128×32, and 128×64; label and +dots beside a large countdown on long panels like 256×32. On short panels the +session dots tuck in beside the label rather than taking a row of their own, so +the digits get the height instead. + +![The same work session on all four panel sizes — 64×32, 128×32, 128×64 and +256×32 — drawn at the same scale so the relative sizes are true](assets/sizes.png) + +| Phase | Default color | +|---|---| +| Work | red | +| Short break | green | +| Long break | blue | +| Paused | amber | +| Idle | grey | + +When a phase ends the panel flashes for a few seconds so you notice even if you +weren't looking — that alert length is configurable, and can be turned off. + +--- + +## Quick Start + +1. **Install the plugin** from the LEDMatrix plugin store and open its + configuration tab. +2. **Set your MQTT broker host** (typically your Home Assistant IP or + `homeassistant.local`). +3. **Set your durations** — work, short break, long break, and sessions per set. +4. **Save.** The plugin connects, subscribes, and announces itself to Home + Assistant. +5. **Test it:** + +```bash +mosquitto_pub -h -t ledmatrix/pomodoro/set -m START +mosquitto_pub -h -t ledmatrix/pomodoro/set -m PAUSE +mosquitto_pub -h -t ledmatrix/pomodoro/set -m RESET +``` + +Not using MQTT at all? Turn **Enable MQTT Control** off and turn **Start Timer +When Plugin Is Enabled** on — the timer then runs the cycle on its own from the +durations you configured. + +--- + +## Plugin Configuration + +### Timer + +| Field | Default | Description | +|---|---|---| +| **Work Session (minutes)** | `25` | Length of a focus session. Also settable live from Home Assistant. | +| **Short Break (minutes)** | `5` | Break after each work session. | +| **Long Break (minutes)** | `15` | Break after a full set of work sessions. | +| **Sessions Before Long Break** | `4` | How many work sessions make a set — this is the number of dots on screen. | +| **Auto-Start Breaks** | `true` | Roll straight into the break when work ends. Off makes the break wait for a Start/Resume. | +| **Auto-Start Next Work Session** | `false` | Roll straight back into work when a break ends. Off by default so you decide when to go again. | +| **Start Timer When Plugin Is Enabled** | `false` | Begin a work session as soon as the plugin loads. Handy without MQTT. | + +### MQTT + +| Field | Default | Description | +|---|---|---| +| **Enable MQTT Control** | `true` | Connect to a broker so the timer can be driven remotely. | +| **Broker Address** | `localhost` | IP or hostname of your MQTT broker. | +| **Broker Port** | `1883` | Use 8883 for TLS. | +| **Username / Password** | *(blank)* | Leave blank for an anonymous broker. | +| **Command Topic** | `ledmatrix/pomodoro/set` | Everything the plugin publishes is derived from this topic's base. | +| **State Topic** | `ledmatrix/pomodoro/state` | `ON` while a session is active, `OFF` when idle. | +| **Enable Home Assistant Auto-Discovery** | `true` | Announce the device to HA over MQTT. | +| **HA Discovery Prefix** | `homeassistant` | Only change this if you changed it in HA. | +| **Device Name in Home Assistant** | `LED Matrix — Pomodoro` | How the device is labelled in HA. | +| **State Publish Interval (seconds)** | `1` | How often the countdown is published while running. Raise it to cut broker traffic — state *changes* are always published immediately. | + +### Appearance + +| Field | Default | Description | +|---|---|---| +| **Countdown Color** | `phase` | `phase` colors the countdown by what's running; `fixed` always uses the Countdown Text Color. | +| **Colour Theme** | `classic` | `classic` uses the individual phase colours below. `calm` overrides them with a softer palette — warm terracotta, sage, soft indigo. | +| **Work / Short Break / Long Break Color** | red / green / blue | Phase colors. | +| **Idle / Paused Color** | grey / amber | Used when nothing is running and when paused. | +| **How Paused Looks** | `amber` | `amber` switches the countdown to the Paused Colour. `desaturate` keeps the phase's own hue but drains it, so a held timer reads as halted without changing which phase you're in. | +| **Countdown Text Color** | white | Only used when Countdown Color is `fixed`. | +| **Background Color** | black | Panel background. | +| **Countdown Digits** | `seven_segment` | `seven_segment` draws clock-radio style segments sized to the panel. `pixel` uses the display's pixel font. | +| **Show Unlit Segments** | `false` | Faintly light the unlit segments, like a real LED clock. Authentic, but it costs legibility — every digit gains a faint `8` behind it. Ignored when the stroke is only one pixel wide, where the effect would just be noise. | +| **Burndown Indicator** | `perimeter` | `perimeter` drains a ring around the edge of the panel; `bar` empties a bar along the bottom; `segments` puts out a row of blocks one at a time; `none` hides it. | +| **Show Phase Label** | `true` | The phase name above the countdown. | +| **Show Session Dots** | `true` | One pip per session in the set. Completed are solid, upcoming are hollow, and the one you're in is picked out. | +| **Pulse the Current Session Dot** | `true` | Slowly blink the pip for the session you're in, so the row reads as "two done, on the third" rather than just a count. | +| **Work / Short Break / Long Break / Idle / Paused Label** | `FOCUS` / `BREAK` / `LONG BREAK` / `POMODORO` / `PAUSED` | The on-screen text for each state. Blank the Paused Label to keep showing the phase name while paused. | +| **Font** | *(blank)* | Path to a TTF relative to the LEDMatrix root, e.g. `assets/fonts/PressStart2P-Regular.ttf`. Blank uses the display's default font. | +| **Font Size (px)** | `0` | Fix the countdown height in pixels. `0` sizes it automatically to the panel. | + +### Behavior + +| Field | Default | Description | +|---|---|---| +| **Hold the Display While Running** | `true` | Take over the matrix while a session is active instead of rotating. Off lets the timer take a normal turn in the rotation. | +| **Phase-Change Alert (seconds)** | `8` | How long the timer grabs the display when a phase ends. `0` disables it. | +| **Flash on Phase Change** | `true` | Flash the panel during that alert. | +| **Display Duration (seconds)** | `10` | Time on screen per rotation cycle when it isn't holding the display. | + +--- + +## MQTT Reference + +All topics below are derived from the **Command Topic**. With the default +`ledmatrix/pomodoro/set`, the base is `ledmatrix/pomodoro`. + +### Commands + +Publish any of these as a plain string to the command topic (case-insensitive): + +| Payload | Effect | +|---|---| +| `START` / `ON` | Start a work session — or resume a paused one. | +| `PAUSE` | Freeze the countdown where it is. | +| `RESUME` | Continue a paused countdown. | +| `TOGGLE` | Start, pause, or resume depending on the current state. Perfect for a single physical button. | +| `SKIP` | Abandon the current phase and move to the next one. A skipped work session doesn't count toward the set. | +| `STOP` / `OFF` | Stop and go idle, keeping the session count. | +| `RESET` | Stop, go idle, and zero the session count. | +| `WORK` | Start a work session immediately, whatever was running. | +| `SHORT_BREAK` / `BREAK` | Start a short break immediately. | +| `LONG_BREAK` | Start a long break immediately. | + +Anything unrecognized is ignored, so a stray message can never be mistaken for a +stop. + +### JSON Payloads + +Send an object to the same topic for one-off durations, labels, and setting +changes: + +```jsonc +// A 50-minute deep-work block with its own on-screen label. +// The configured Work Session length is left alone. +{"command": "start", "phase": "work", "duration_minutes": 50, "label": "DEEP WORK"} + +// Change the configured durations without starting anything. +{"work_minutes": 45, "short_break_minutes": 10, "sessions_before_long_break": 3} + +// Change a duration and start in one message. +{"command": "start", "work_minutes": 30} +``` + +| Key | Meaning | +|---|---| +| `command` (or `action`, `state`) | Any command from the table above. | +| `phase` | `work`, `short_break`, or `long_break` — which phase `start` should begin. | +| `duration_minutes` / `duration_seconds` | Length of *this* phase only. | +| `label` | On-screen text for this phase only. | +| `work_minutes`, `short_break_minutes`, `long_break_minutes`, `sessions_before_long_break` | Update the configured values. | + +> Duration and label overrides last for the phase they start. Changes to +> `work_minutes` and friends apply from that moment on but are **not** written +> back to the plugin's saved configuration — set them in the web UI to make them +> permanent. + +### Published Topics + +| Topic | Payload | +|---|---| +| `/state` | `ON` when a session is active, `OFF` when idle | +| `/status` | `running`, `paused`, or `idle` | +| `/phase` | `idle`, `work`, `short_break`, or `long_break` | +| `/remaining` | `MM:SS` | +| `/remaining_seconds` | Integer seconds | +| `/session` | Completed work sessions | +| `/attributes` | JSON snapshot of everything above plus `cycle_position`, `elapsed_fraction`, and the configured durations | +| `/event` | JSON `{"event_type": …}` on every transition — `started`, `paused`, `resumed`, `stopped`, `reset`, `skipped`, `phase_started`, `work_complete`, `break_complete` | +| `/task` | The current task label | +| `/available` | `online` / `offline` (also the last-will message) | +| `/work_minutes` etc. | Current value of each duration setting | + +The task label is set by publishing to `/task/set` (max 32 characters; +publish an empty payload to clear it). `RESET` clears it too. + +Each duration also has a matching command topic — `/work_minutes/set`, +`/short_break_minutes/set`, `/long_break_minutes/set`, +`/sessions_before_long_break/set` — which is what the Home Assistant +number boxes write to. + +--- + +## Home Assistant Setup + +### Auto-Discovery (recommended) + +1. Make sure the **MQTT integration** is set up in Home Assistant + (*Settings → Devices & Services → Add Integration → MQTT*) and pointed at the + same broker. +2. Leave **Enable Home Assistant Auto-Discovery** on in the plugin config. +3. Save. The device appears under *Settings → Devices & Services → MQTT* within + a few seconds. + +### What You Get + +| Entity | Type | What it does | +|---|---|---| +| **Timer** | Switch | On starts a session, off stops it. Carries the full state as attributes. | +| **Start / Pause / Resume / Skip Phase / Reset** | Buttons | One press per command. | +| **Work Minutes / Short Break Minutes / Long Break Minutes / Sessions Before Long Break** | Numbers | Set the durations from HA — the matrix picks them up immediately. | +| **Phase** | Sensor | `idle`, `work`, `short_break`, `long_break` | +| **Status** | Sensor | `running`, `paused`, `idle` | +| **Time Remaining** | Sensor | `MM:SS` | +| **Seconds Remaining** | Sensor | Numeric, for gauges and templates | +| **Completed Sessions** | Sensor | Running count, resets on `RESET` | +| **Task** | Text | What you're working on. Type it in HA and it appears on the panel. | +| **Timer Event** | Event | Fires on every transition — the clean hook for automations | +| **MQTT Connected** | Binary sensor | Connectivity | + +### Automation Examples + +**Announce the end of a work session on a speaker:** + +```yaml +automation: + - alias: "Pomodoro — time for a break" + triggers: + - trigger: state + entity_id: event.led_matrix_pomodoro_timer_event + attribute: event_type + to: work_complete + actions: + - action: tts.speak + target: + entity_id: tts.piper + data: + media_player_entity_id: media_player.office + message: "Nice work. Take a break." +``` + +**Start a session when you sit down at your desk:** + +```yaml +automation: + - alias: "Pomodoro — start on desk occupancy" + triggers: + - trigger: state + entity_id: binary_sensor.desk_occupancy + to: "on" + for: "00:02:00" + conditions: + - condition: state + entity_id: sensor.led_matrix_pomodoro_phase + state: idle + actions: + - action: button.press + target: + entity_id: button.led_matrix_pomodoro_start +``` + +**Turn on do-not-disturb for the length of a work session:** + +```yaml +automation: + - alias: "Pomodoro — focus mode" + triggers: + - trigger: state + entity_id: sensor.led_matrix_pomodoro_phase + actions: + - action: "switch.turn_{{ 'on' if trigger.to_state.state == 'work' else 'off' }}" + target: + entity_id: switch.office_do_not_disturb +``` + +**Dashboard card:** + +```yaml +type: entities +title: Pomodoro +entities: + - entity: sensor.led_matrix_pomodoro_time_remaining + - entity: sensor.led_matrix_pomodoro_phase + - entity: sensor.led_matrix_pomodoro_completed_sessions + - type: buttons + entities: + - entity: button.led_matrix_pomodoro_start + - entity: button.led_matrix_pomodoro_pause + - entity: button.led_matrix_pomodoro_skip_phase + - entity: button.led_matrix_pomodoro_reset + - entity: number.led_matrix_pomodoro_work_minutes + - entity: number.led_matrix_pomodoro_short_break_minutes +``` + +> Entity IDs are generated from the device name — check +> *Settings → Devices & Services → MQTT → LED Matrix — Pomodoro* for the exact +> ones on your system. + +### Manual YAML (if you turn discovery off) + +```yaml +mqtt: + switch: + - name: "Pomodoro" + command_topic: "ledmatrix/pomodoro/set" + state_topic: "ledmatrix/pomodoro/state" + payload_on: "START" + payload_off: "STOP" + state_on: "ON" + state_off: "OFF" + json_attributes_topic: "ledmatrix/pomodoro/attributes" + availability_topic: "ledmatrix/pomodoro/available" + icon: mdi:timer-play-outline + + sensor: + - name: "Pomodoro Time Remaining" + state_topic: "ledmatrix/pomodoro/remaining" + availability_topic: "ledmatrix/pomodoro/available" + - name: "Pomodoro Phase" + state_topic: "ledmatrix/pomodoro/phase" + availability_topic: "ledmatrix/pomodoro/available" + + button: + - name: "Pomodoro Skip" + command_topic: "ledmatrix/pomodoro/set" + payload_press: "SKIP" + + number: + - name: "Pomodoro Work Minutes" + command_topic: "ledmatrix/pomodoro/work_minutes/set" + state_topic: "ledmatrix/pomodoro/work_minutes" + min: 1 + max: 180 + unit_of_measurement: "min" +``` + +--- + +## Testing Without Home Assistant + +```bash +# Watch everything the plugin publishes +mosquitto_sub -h -t 'ledmatrix/pomodoro/#' -v + +# Drive it +mosquitto_pub -h -t ledmatrix/pomodoro/set -m START +mosquitto_pub -h -t ledmatrix/pomodoro/set -m PAUSE +mosquitto_pub -h -t ledmatrix/pomodoro/set -m SKIP + +# A one-off 50-minute block +mosquitto_pub -h -t ledmatrix/pomodoro/set \ + -m '{"command":"start","duration_minutes":50,"label":"DEEP WORK"}' + +# Change the configured work length +mosquitto_pub -h -t ledmatrix/pomodoro/work_minutes/set -m 45 +``` + +The plugin's own test suite runs without a broker or a LEDMatrix checkout: + +```bash +python plugins/pomodoro-timer/test_pomodoro_timer.py +``` + +The screenshots above are generated the same way the safety harness renders the +plugin, so they can't drift from what the display actually does. Regenerate them +against a LEDMatrix checkout with: + +```bash +python plugins/pomodoro-timer/test/render_readme_assets.py --core /path/to/LEDMatrix +``` + +--- + +## Troubleshooting + +| Symptom | Cause / fix | +|---|---| +| Nothing appears in Home Assistant | The MQTT integration isn't set up, or it points at a different broker. Check *Settings → Devices & Services → MQTT*, and confirm the discovery prefix matches (`homeassistant` unless you changed it). | +| Log says `paho-mqtt is not installed` | Install it on the Pi: `pip install paho-mqtt`. The plugin still renders; only remote control is unavailable. | +| Log says `MQTT connect failed rc=5` | Bad username/password, or the broker requires authentication. | +| Log says `MQTT connect error` and retries | Wrong host or port, or the broker isn't reachable from the Pi. The plugin retries with backoff, so fixing the config is enough. | +| The timer never appears on the matrix | Check the plugin is enabled and, if **Hold the Display While Running** is off, that it has a turn in your display rotation. | +| The timer takes over and won't give the panel back | That's **Hold the Display While Running**. Turn it off to keep the timer in the normal rotation, or send `STOP`. | +| The countdown looks tiny on a big panel | The seven-segment digits are sized to the space left after the label and dots. Turning off **Show Phase Label**, or switching **Burndown Indicator** to `perimeter`, frees the most room. **Font Size (px)** applies to the `pixel` digit style only. | +| The digits look squashed or the zeros read as two stacked boxes | The digits refuse to go below a readable width-to-height ratio and step down in size instead, so this should not happen — if it does on your panel shape, please open an issue with the dimensions. | +| Text is cut off | Shorten the phase labels — long labels are truncated to fit rather than drawn past the panel edge. | +| Durations changed in HA revert after a restart | Number-box changes are runtime-only by design. Set them in the plugin's web UI config to make them permanent. | +| The session count reset itself | `RESET` zeroes it (that's the difference between `STOP` and `RESET`), and so does a long break for the on-screen dots. | + +--- + +## License + +GPL-3.0 — see [LICENSE](LICENSE). diff --git a/plugins/pomodoro-timer/assets/hero.png b/plugins/pomodoro-timer/assets/hero.png new file mode 100644 index 0000000..30ee5c5 Binary files /dev/null and b/plugins/pomodoro-timer/assets/hero.png differ diff --git a/plugins/pomodoro-timer/assets/options.png b/plugins/pomodoro-timer/assets/options.png new file mode 100644 index 0000000..4d2c9c6 Binary files /dev/null and b/plugins/pomodoro-timer/assets/options.png differ diff --git a/plugins/pomodoro-timer/assets/phases.png b/plugins/pomodoro-timer/assets/phases.png new file mode 100644 index 0000000..09785f8 Binary files /dev/null and b/plugins/pomodoro-timer/assets/phases.png differ diff --git a/plugins/pomodoro-timer/assets/sizes.png b/plugins/pomodoro-timer/assets/sizes.png new file mode 100644 index 0000000..4dacc7f Binary files /dev/null and b/plugins/pomodoro-timer/assets/sizes.png differ diff --git a/plugins/pomodoro-timer/config_schema.json b/plugins/pomodoro-timer/config_schema.json new file mode 100644 index 0000000..1e29d49 --- /dev/null +++ b/plugins/pomodoro-timer/config_schema.json @@ -0,0 +1,486 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "title": "Pomodoro Timer", + "description": "A focus/break timer on the matrix, controlled over MQTT or Home Assistant.", + "x-propertyOrder": [ + "enabled", + "work_minutes", + "short_break_minutes", + "long_break_minutes", + "sessions_before_long_break", + "auto_start_breaks", + "auto_start_work", + "auto_start_on_enable", + "mqtt_enabled", + "mqtt_host", + "mqtt_port", + "mqtt_username", + "mqtt_password", + "command_topic", + "state_topic", + "ha_discovery", + "discovery_prefix", + "device_name", + "publish_interval_seconds", + "color_theme", + "color_mode", + "work_color", + "short_break_color", + "long_break_color", + "idle_color", + "paused_color", + "paused_style", + "time_color", + "background_color", + "digit_style", + "show_ghost_segments", + "show_phase_label", + "show_session_dots", + "pulse_active_pip", + "progress_style", + "work_label", + "short_break_label", + "long_break_label", + "idle_label", + "paused_label", + "font_path", + "font_size", + "pin_while_running", + "alert_seconds", + "alert_flash", + "display_duration" + ], + "properties": { + "enabled": { + "type": "boolean", + "default": true, + "title": "Enabled" + }, + "work_minutes": { + "type": "integer", + "default": 25, + "minimum": 1, + "maximum": 180, + "title": "Work Session (minutes)", + "description": "Length of a focus session. Also settable at runtime from Home Assistant." + }, + "short_break_minutes": { + "type": "integer", + "default": 5, + "minimum": 1, + "maximum": 60, + "title": "Short Break (minutes)", + "description": "Break taken after each work session." + }, + "long_break_minutes": { + "type": "integer", + "default": 15, + "minimum": 1, + "maximum": 120, + "title": "Long Break (minutes)", + "description": "Longer break taken after a full set of work sessions." + }, + "sessions_before_long_break": { + "type": "integer", + "default": 4, + "minimum": 1, + "maximum": 12, + "title": "Sessions Before Long Break", + "description": "How many work sessions make up one set. Shown on screen as the row of session dots." + }, + "auto_start_breaks": { + "type": "boolean", + "default": true, + "title": "Auto-Start Breaks", + "description": "Start the break automatically when a work session ends. Turn this off to make the break wait for a Start/Resume command." + }, + "auto_start_work": { + "type": "boolean", + "default": false, + "title": "Auto-Start Next Work Session", + "description": "Start the next work session automatically when a break ends. Off by default so you decide when to get back to it." + }, + "auto_start_on_enable": { + "type": "boolean", + "default": false, + "title": "Start Timer When Plugin Is Enabled", + "description": "Begin a work session as soon as the plugin loads. Useful if you don't use MQTT.", + "x-advanced": true + }, + "mqtt_enabled": { + "type": "boolean", + "default": true, + "title": "Enable MQTT Control", + "description": "Connect to an MQTT broker so the timer can be driven from Home Assistant or any MQTT client. Turn off to run the timer purely from this configuration." + }, + "mqtt_host": { + "type": "string", + "default": "localhost", + "title": "Broker Address", + "description": "Hostname or IP of your MQTT broker — usually the same address as your Home Assistant instance." + }, + "mqtt_port": { + "type": "integer", + "default": 1883, + "minimum": 1, + "maximum": 65535, + "title": "Broker Port", + "description": "Default is 1883. Use 8883 for TLS." + }, + "mqtt_username": { + "type": "string", + "default": "", + "title": "Username", + "description": "MQTT broker username. Leave blank if your broker does not require authentication." + }, + "mqtt_password": { + "type": "string", + "default": "", + "title": "Password", + "description": "MQTT broker password. Leave blank if your broker does not require authentication.", + "x-sensitive": true + }, + "command_topic": { + "type": "string", + "default": "ledmatrix/pomodoro/set", + "title": "Command Topic", + "description": "The plugin subscribes here. Publish START, PAUSE, RESUME, TOGGLE, SKIP, STOP or RESET — or JSON for per-message durations and labels. Every other topic is derived from this one." + }, + "state_topic": { + "type": "string", + "default": "ledmatrix/pomodoro/state", + "title": "State Topic", + "description": "The plugin publishes ON (a session is active) or OFF (idle) here so the Home Assistant switch stays in sync.", + "x-advanced": true + }, + "ha_discovery": { + "type": "boolean", + "default": true, + "title": "Enable Home Assistant Auto-Discovery", + "description": "Announce the timer to Home Assistant over MQTT. HA creates a device with the switch, start/pause/skip/reset buttons, duration number boxes, and phase/remaining sensors — no configuration.yaml entry needed." + }, + "discovery_prefix": { + "type": "string", + "default": "homeassistant", + "title": "HA Discovery Prefix", + "description": "The MQTT topic prefix your HA MQTT integration listens on. Leave as 'homeassistant' unless you changed it in HA.", + "x-advanced": true + }, + "device_name": { + "type": "string", + "default": "LED Matrix — Pomodoro", + "title": "Device Name in Home Assistant", + "description": "How this device appears under Settings → Devices & Services → MQTT in Home Assistant.", + "x-advanced": true + }, + "publish_interval_seconds": { + "type": "integer", + "default": 1, + "minimum": 1, + "maximum": 60, + "title": "State Publish Interval (seconds)", + "description": "How often the countdown is published while running. Raise this to cut broker traffic; state changes (start, pause, phase change) are always published immediately.", + "x-advanced": true + }, + "color_mode": { + "type": "string", + "enum": [ + "phase", + "fixed" + ], + "default": "phase", + "title": "Countdown Color", + "description": "'phase' colors the countdown by what's running (work / break / long break / paused). 'fixed' always uses the Countdown Text Color below.", + "x-widget": "radio" + }, + "work_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 255, + 70, + 50 + ], + "title": "Work Color", + "x-widget": "color-picker" + }, + "short_break_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 60, + 200, + 110 + ], + "title": "Short Break Color", + "x-widget": "color-picker" + }, + "long_break_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 70, + 150, + 255 + ], + "title": "Long Break Color", + "x-widget": "color-picker" + }, + "idle_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 110, + 110, + 110 + ], + "title": "Idle Color", + "x-widget": "color-picker", + "x-advanced": true + }, + "paused_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 255, + 176, + 0 + ], + "title": "Paused Color", + "x-widget": "color-picker", + "x-advanced": true + }, + "time_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 255, + 255, + 255 + ], + "title": "Countdown Text Color", + "description": "Used only when Countdown Color is set to 'fixed'.", + "x-widget": "color-picker", + "x-advanced": true + }, + "background_color": { + "type": "array", + "items": { + "type": "integer", + "minimum": 0, + "maximum": 255 + }, + "minItems": 3, + "maxItems": 3, + "default": [ + 0, + 0, + 0 + ], + "title": "Background Color", + "x-widget": "color-picker", + "x-advanced": true + }, + "show_phase_label": { + "type": "boolean", + "default": true, + "title": "Show Phase Label", + "description": "Show the current phase name above the countdown." + }, + "show_session_dots": { + "type": "boolean", + "default": true, + "title": "Show Session Dots", + "description": "Show one dot per work session in the set, filled in as you complete them." + }, + "work_label": { + "type": "string", + "default": "FOCUS", + "maxLength": 24, + "title": "Work Label", + "x-advanced": true + }, + "short_break_label": { + "type": "string", + "default": "BREAK", + "maxLength": 24, + "title": "Short Break Label", + "x-advanced": true + }, + "long_break_label": { + "type": "string", + "default": "LONG BREAK", + "maxLength": 24, + "title": "Long Break Label", + "x-advanced": true + }, + "idle_label": { + "type": "string", + "default": "POMODORO", + "maxLength": 24, + "title": "Idle Label", + "description": "Shown when no timer is running, above the next work session's length.", + "x-advanced": true + }, + "paused_label": { + "type": "string", + "default": "PAUSED", + "maxLength": 24, + "title": "Paused Label", + "description": "Shown in place of the phase name while the timer is paused. Leave blank to keep showing the phase name.", + "x-advanced": true + }, + "font_path": { + "type": "string", + "default": "", + "title": "Font", + "description": "Path to a TTF font file relative to the LEDMatrix root, e.g. assets/fonts/PressStart2P-Regular.ttf. Leave blank to use the display's default font.", + "x-advanced": true + }, + "font_size": { + "type": "integer", + "default": 0, + "minimum": 0, + "maximum": 64, + "title": "Font Size (px)", + "description": "Fix the countdown font height in pixels. Leave at 0 to size the countdown automatically to the panel.", + "x-advanced": true + }, + "pin_while_running": { + "type": "boolean", + "default": true, + "title": "Hold the Display While Running", + "description": "Take over the matrix for as long as a session is active, instead of rotating through your other plugins. Turn off to let the timer take its normal turn in the rotation." + }, + "alert_seconds": { + "type": "integer", + "default": 8, + "minimum": 0, + "maximum": 60, + "title": "Phase-Change Alert (seconds)", + "description": "How long the timer grabs the display when a phase ends. Set to 0 to disable the alert." + }, + "alert_flash": { + "type": "boolean", + "default": true, + "title": "Flash on Phase Change", + "description": "Flash the panel during the phase-change alert.", + "x-advanced": true + }, + "display_duration": { + "type": "number", + "default": 10, + "minimum": 1, + "maximum": 300, + "title": "Display Duration (seconds)", + "description": "How long the timer stays on screen each rotation cycle when it isn't holding the display.", + "x-advanced": true + }, + "digit_style": { + "type": "string", + "enum": [ + "seven_segment", + "pixel" + ], + "default": "seven_segment", + "title": "Countdown Digits", + "description": "'seven_segment' draws the countdown as clock-radio style segments sized to the panel. 'pixel' uses the display's pixel font.", + "x-widget": "radio" + }, + "show_ghost_segments": { + "type": "boolean", + "default": false, + "title": "Show Unlit Segments", + "description": "Show the unlit segments faintly, the way a real LED clock shows its dark segments. It looks authentic but costs legibility — every digit gains a faint 8 behind it — so it is off by default. Ignored when the stroke is only one pixel wide.", + "x-advanced": true + }, + "progress_style": { + "type": "string", + "enum": [ + "perimeter", + "bar", + "segments", + "none" + ], + "default": "perimeter", + "title": "Burndown Indicator", + "description": "How the time left in the phase is shown. 'perimeter' drains a ring around the edge of the panel and leaves the most room for the digits; 'bar' empties a bar along the bottom; 'segments' puts out a row of blocks one at a time; 'none' hides it.", + "x-widget": "radio" + }, + "color_theme": { + "type": "string", + "enum": [ + "classic", + "calm" + ], + "default": "classic", + "title": "Colour Theme", + "description": "'classic' uses the individual phase colours below. 'calm' overrides them with a softer palette — warm terracotta, sage, and a soft indigo — for a panel that doesn't shout at you.", + "x-widget": "radio" + }, + "paused_style": { + "type": "string", + "enum": [ + "amber", + "desaturate" + ], + "default": "amber", + "title": "How Paused Looks", + "description": "'amber' switches the countdown to the Paused Colour. 'desaturate' keeps the phase's own hue but drains it, so a held timer reads as halted without changing what phase you're in.", + "x-widget": "radio", + "x-advanced": true + }, + "pulse_active_pip": { + "type": "boolean", + "default": true, + "title": "Pulse the Current Session Dot", + "description": "Slowly blink the dot for the session you're in, so the row reads as 'two done, on the third' rather than just a count. Completed sessions are solid and upcoming ones are hollow either way.", + "x-advanced": true + } + }, + "required": [ + "enabled", + "work_minutes", + "short_break_minutes", + "long_break_minutes", + "sessions_before_long_break" + ], + "additionalProperties": false +} diff --git a/plugins/pomodoro-timer/manager.py b/plugins/pomodoro-timer/manager.py new file mode 100644 index 0000000..717aa6e --- /dev/null +++ b/plugins/pomodoro-timer/manager.py @@ -0,0 +1,1736 @@ +""" +Pomodoro Timer Plugin for LEDMatrix + +A configurable focus/break (Pomodoro) timer rendered on the matrix and driven +over MQTT, with Home Assistant auto-discovery so the whole timer — start, +pause, resume, skip, reset, plus the work and break durations — appears as a +device in Home Assistant with no YAML. + +The timer runs in its own thread, so it keeps counting (and keeps publishing +state) whether or not the plugin is the screen currently on the panel. + +MQTT topics (all derived from the command topic's base): + /set — subscribe: START | PAUSE | RESUME | TOGGLE | STOP | + RESET | SKIP | WORK | SHORT_BREAK | LONG_BREAK, + or a JSON object (see _parse_command) + /state — publish: ON / OFF (HA switch) + /status — publish: running / paused / idle + /phase — publish: idle | work | short_break | long_break + /remaining — publish: MM:SS + /remaining_seconds — publish: integer seconds + /session — publish: completed work sessions + /attributes — publish: JSON snapshot of the full state + /event — publish: JSON {"event_type": ...} on transitions + /available — publish: online / offline (also the LWT) + //set — subscribe: work_minutes, short_break_minutes, + / long_break_minutes, sessions_before_long_break + +HA MQTT Auto-Discovery (enabled by default) publishes device + entity configs +to homeassistant//... on connect, so Home Assistant builds the +device automatically. +""" + +import json +import math +import os +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from PIL import Image, ImageDraw, ImageFont + +try: + import paho.mqtt.client as mqtt +except ImportError: + mqtt = None # type: ignore + +from src.plugin_system.base_plugin import BasePlugin + +_PLUGIN_VERSION = "1.0.0" + +PHASE_IDLE = "idle" +PHASE_WORK = "work" +PHASE_SHORT_BREAK = "short_break" +PHASE_LONG_BREAK = "long_break" + +# attr name, HA entity name, min, max, step, icon +_NUMBER_FIELDS: Tuple[Tuple[str, str, int, int, int, str], ...] = ( + ("work_minutes", "Work Minutes", 1, 180, 1, "mdi:briefcase-clock"), + ("short_break_minutes", "Short Break Minutes", 1, 60, 1, "mdi:coffee-outline"), + ("long_break_minutes", "Long Break Minutes", 1, 120, 1, "mdi:sofa-outline"), + ("sessions_before_long_break", "Sessions Before Long Break", 1, 12, 1, "mdi:counter"), +) + +# key suffix, HA entity name, payload published to the command topic, icon +_BUTTONS: Tuple[Tuple[str, str, str, str], ...] = ( + ("start", "Start", "START", "mdi:play"), + ("pause", "Pause", "PAUSE", "mdi:pause"), + ("resume", "Resume", "RESUME", "mdi:play"), + ("skip", "Skip Phase", "SKIP", "mdi:skip-next"), + ("reset", "Reset", "RESET", "mdi:restart"), +) + +_EVENT_TYPES = [ + "started", "paused", "resumed", "stopped", "reset", "skipped", + "phase_started", "work_complete", "break_complete", +] + +_PHASE_ORDER = (PHASE_WORK, PHASE_SHORT_BREAK, PHASE_LONG_BREAK) + +# Calmer hues for people who don't want the panel shouting at them. Chosen to +# still separate cleanly on an LED matrix, where heavily muted colours turn to +# mud: warm terracotta, sage, and a soft indigo. +_CALM_THEME = { + "work_color": (230, 108, 76), + "short_break_color": (122, 176, 150), + "long_break_color": (118, 132, 210), + "idle_color": (120, 124, 140), +} + + +def _rgb(value, default: Tuple[int, int, int]) -> Tuple[int, int, int]: + """Coerce a config/JSON colour into a clamped RGB tuple.""" + try: + parts = [max(0, min(255, int(c))) for c in value] + except (TypeError, ValueError): + return default + if len(parts) != 3: + return default + return (parts[0], parts[1], parts[2]) + + +def _dim(color: Tuple[int, int, int], factor: float) -> Tuple[int, int, int]: + return tuple(max(0, min(255, int(c * factor))) for c in color) # type: ignore[return-value] + + +def _desaturate(color: Tuple[int, int, int], amount: float) -> Tuple[int, int, int]: + """Pull `color` toward its own luminance — keeps the hue, drops the punch.""" + grey = int(0.299 * color[0] + 0.587 * color[1] + 0.114 * color[2]) + return tuple( # type: ignore[return-value] + max(0, min(255, int(c + (grey - c) * amount))) for c in color) + + +def _lighten(color: Tuple[int, int, int], factor: float) -> Tuple[int, int, int]: + """Blend `factor` of white into `color` — used for the burndown's head.""" + return tuple( # type: ignore[return-value] + max(0, min(255, int(c + (255 - c) * factor))) for c in color) + + +def _fmt_clock(seconds: float) -> str: + """Format remaining seconds as MM:SS (minutes are not wrapped at 60).""" + total = int(math.ceil(max(0.0, seconds) - 1e-6)) + return f"{total // 60:02d}:{total % 60:02d}" + + +class PomodoroTimerPlugin(BasePlugin): + """Pomodoro focus/break timer with MQTT control and HA auto-discovery.""" + + def __init__(self, plugin_id: str, config: Dict[str, Any], + display_manager, cache_manager, plugin_manager): + super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager) + + # Runtime state (guarded by state_lock). Created before settings are + # loaded so _load_settings can seed the idle countdown. + self.state_lock = threading.RLock() + self.phase = PHASE_IDLE + self.running = False + self.phase_total = 25 * 60.0 + self.remaining = 25 * 60.0 + self.deadline: Optional[float] = None + self.completed_sessions = 0 + self.cycle_position = 0 + self.custom_label: Optional[str] = None + self.task = "" + self.alert_until = 0.0 + self.alert_color: Optional[Tuple[int, int, int]] = None + + self._load_settings(config) + self.remaining = self.phase_total = float(self.work_minutes * 60) + + # Fonts + self._font_cache: Dict[int, Any] = {} + self._ttf_path_cache: Optional[str] = None + self._ttf_path_resolved = False + + # Display pinning + self._pinned = False + + # Timer thread + self.timer_thread: Optional[threading.Thread] = None + self.timer_stop_event = threading.Event() + + # MQTT internals + self.mqtt_client_id = f"ledmatrix-pomodoro-{plugin_id}-{uuid.uuid4().hex[:8]}" + self.mqtt_client: Optional["mqtt.Client"] = None + self.mqtt_thread: Optional[threading.Thread] = None + self.mqtt_connected = False + self.mqtt_connecting = False + self.mqtt_reconnect_delay = 1.0 + self.mqtt_max_reconnect_delay = 60.0 + self.mqtt_stop_event = threading.Event() + self._last_publish = 0.0 + self._last_published_key: Optional[Tuple] = None + + if self.mqtt_enabled and mqtt is None: + self.logger.error( + "paho-mqtt is not installed — MQTT control is disabled. " + "Install it with: pip install paho-mqtt") + + self.logger.info("Pomodoro timer ready — %d/%d/%d min, command topic: %s", + self.work_minutes, self.short_break_minutes, + self.long_break_minutes, self.command_topic) + + # ── Configuration ─────────────────────────────────────────────────────────── + + def _load_settings(self, config: Dict[str, Any]) -> None: + """Read every configurable value out of `config` onto self.""" + def _int(key: str, default: int, low: int, high: int) -> int: + try: + return max(low, min(high, int(config.get(key, default)))) + except (TypeError, ValueError): + return default + + # Timer + self.work_minutes = _int("work_minutes", 25, 1, 180) + self.short_break_minutes = _int("short_break_minutes", 5, 1, 60) + self.long_break_minutes = _int("long_break_minutes", 15, 1, 120) + self.sessions_before_long_break = _int("sessions_before_long_break", 4, 1, 12) + self.auto_start_breaks = bool(config.get("auto_start_breaks", True)) + self.auto_start_work = bool(config.get("auto_start_work", False)) + self.auto_start_on_enable = bool(config.get("auto_start_on_enable", False)) + + # Labels + self.work_label = str(config.get("work_label", "FOCUS")) + self.short_break_label = str(config.get("short_break_label", "BREAK")) + self.long_break_label = str(config.get("long_break_label", "LONG BREAK")) + self.idle_label = str(config.get("idle_label", "POMODORO")) + self.paused_label = str(config.get("paused_label", "PAUSED")) + + # Appearance + self.color_mode = str(config.get("color_mode", "phase")) + self.time_color = _rgb(config.get("time_color", [255, 255, 255]), (255, 255, 255)) + self.work_color = _rgb(config.get("work_color", [255, 70, 50]), (255, 70, 50)) + self.short_break_color = _rgb(config.get("short_break_color", [60, 200, 110]), (60, 200, 110)) + self.long_break_color = _rgb(config.get("long_break_color", [70, 150, 255]), (70, 150, 255)) + self.idle_color = _rgb(config.get("idle_color", [110, 110, 110]), (110, 110, 110)) + self.paused_color = _rgb(config.get("paused_color", [255, 176, 0]), (255, 176, 0)) + self.background_color = _rgb(config.get("background_color", [0, 0, 0]), (0, 0, 0)) + self.color_theme = str(config.get("color_theme", "classic")) + if self.color_theme == "calm": + for key, value in _CALM_THEME.items(): + setattr(self, key.replace("_color", "") + "_color", value) + self.paused_style = str(config.get("paused_style", "amber")) + if self.paused_style not in ("amber", "desaturate"): + self.paused_style = "amber" + self.pulse_active_pip = bool(config.get("pulse_active_pip", True)) + self.show_phase_label = bool(config.get("show_phase_label", True)) + self.show_session_dots = bool(config.get("show_session_dots", True)) + self.digit_style = str(config.get("digit_style", "seven_segment")) + if self.digit_style not in ("seven_segment", "pixel"): + self.digit_style = "seven_segment" + self.show_ghost_segments = bool(config.get("show_ghost_segments", False)) + self.progress_style = str(config.get("progress_style", "perimeter")) + if self.progress_style not in ("perimeter", "bar", "segments", "none"): + self.progress_style = "perimeter" + self.font_path = str(config.get("font_path", "") or "") + try: + self.font_size = max(0, min(64, int(config.get("font_size", 0)))) + except (TypeError, ValueError): + self.font_size = 0 + + # Behaviour + self.pin_while_running = bool(config.get("pin_while_running", True)) + try: + self.alert_seconds = max(0, min(60, int(config.get("alert_seconds", 8)))) + except (TypeError, ValueError): + self.alert_seconds = 8 + self.alert_flash = bool(config.get("alert_flash", True)) + + # MQTT + self.mqtt_enabled = bool(config.get("mqtt_enabled", True)) + self.mqtt_host = str(config.get("mqtt_host", "localhost")) + try: + self.mqtt_port = int(config.get("mqtt_port", 1883)) + except (TypeError, ValueError): + self.mqtt_port = 1883 + self.mqtt_username = str(config.get("mqtt_username", "") or "") + self.mqtt_password = str(config.get("mqtt_password", "") or "") + self.command_topic = str(config.get("command_topic", "ledmatrix/pomodoro/set")) + self.state_topic = str(config.get("state_topic", "ledmatrix/pomodoro/state")) + self.ha_discovery = bool(config.get("ha_discovery", True)) + self.discovery_prefix = str(config.get("discovery_prefix", "homeassistant")) + self.device_name = str(config.get("device_name", "LED Matrix — Pomodoro")) + try: + self.publish_interval = max(1, min(60, int(config.get("publish_interval_seconds", 1)))) + except (TypeError, ValueError): + self.publish_interval = 1 + + self._derive_topics() + + def _derive_topics(self) -> None: + """Rebuild every published topic from the command topic's base.""" + base = self.command_topic + if base.endswith("/set"): + base = base[:-4] + self.topic_base = base + self.availability_topic = f"{base}/available" + self.status_topic = f"{base}/status" + self.phase_topic = f"{base}/phase" + self.remaining_topic = f"{base}/remaining" + self.remaining_secs_topic = f"{base}/remaining_seconds" + self.session_topic = f"{base}/session" + self.attributes_topic = f"{base}/attributes" + self.event_topic = f"{base}/event" + self.task_topic = f"{base}/task" + self.task_set_topic = f"{base}/task/set" + self.number_state_topics = {f: f"{base}/{f}" for f, *_ in _NUMBER_FIELDS} + self.number_set_topics = {f"{base}/{f}/set": f for f, *_ in _NUMBER_FIELDS} + + def validate_config(self) -> bool: + """Check the plugin-specific config on top of the base validation.""" + if not super().validate_config(): + return False + if self.color_mode not in ("phase", "fixed"): + self.logger.error("Invalid color_mode: %s (expected 'phase' or 'fixed')", + self.color_mode) + return False + if not self.command_topic.strip(): + self.logger.error("command_topic must not be empty") + return False + if not 1 <= self.mqtt_port <= 65535: + self.logger.error("Invalid mqtt_port: %s", self.mqtt_port) + return False + return True + + # ── Timer state ───────────────────────────────────────────────────────────── + + def _phase_seconds(self, phase: str) -> float: + """Configured length of `phase`, in seconds.""" + return { + PHASE_WORK: self.work_minutes * 60.0, + PHASE_SHORT_BREAK: self.short_break_minutes * 60.0, + PHASE_LONG_BREAK: self.long_break_minutes * 60.0, + }.get(phase, self.work_minutes * 60.0) + + def _remaining_locked(self) -> float: + """Seconds left in the current phase. Caller must hold state_lock.""" + if self.running and self.deadline is not None: + return max(0.0, self.deadline - time.monotonic()) + return max(0.0, self.remaining) + + def _begin_phase_locked(self, phase: str, seconds: Optional[float] = None, + run: bool = True, label: Optional[str] = None) -> None: + """Start `phase`, optionally paused. Caller must hold state_lock.""" + total = float(seconds) if seconds else self._phase_seconds(phase) + self.phase = phase + self.phase_total = max(1.0, total) + self.remaining = self.phase_total + self.running = bool(run) and phase != PHASE_IDLE + self.deadline = time.monotonic() + self.phase_total if self.running else None + self.custom_label = label or None + + def _go_idle_locked(self, reset_counters: bool = False) -> None: + """Stop the timer. Caller must hold state_lock.""" + self.phase = PHASE_IDLE + self.running = False + self.deadline = None + self.phase_total = self._phase_seconds(PHASE_WORK) + self.remaining = self.phase_total + self.custom_label = None + if reset_counters: + self.completed_sessions = 0 + self.cycle_position = 0 + + def _next_phase_locked(self, completed: bool = True) -> str: + """Which phase follows the current one, without mutating counters. + + A skipped work session doesn't count toward the set, so it never earns + the long break. + """ + if self.phase == PHASE_WORK: + position = self.cycle_position + (1 if completed else 0) + return (PHASE_LONG_BREAK if position >= self.sessions_before_long_break + else PHASE_SHORT_BREAK) + return PHASE_WORK + + def _advance_locked(self, completed: bool) -> List[Dict[str, Any]]: + """Move to the next phase. `completed` distinguishes a finished phase + from a skipped one (only finished work counts toward the cycle).""" + events: List[Dict[str, Any]] = [] + previous = self.phase + nxt = self._next_phase_locked(completed) + + if previous == PHASE_WORK and completed: + self.completed_sessions += 1 + self.cycle_position = min(self.sessions_before_long_break, + self.cycle_position + 1) + elif previous == PHASE_LONG_BREAK: + # A long break closes the set, so the session dots start over. + self.cycle_position = 0 + auto = self.auto_start_work if nxt == PHASE_WORK else self.auto_start_breaks + + if completed: + events.append({ + "event_type": "work_complete" if previous == PHASE_WORK else "break_complete", + "phase": previous, + "next_phase": nxt, + "completed_sessions": self.completed_sessions, + }) + if self.alert_seconds: + self.alert_until = time.monotonic() + self.alert_seconds + self.alert_color = self._phase_color(previous) + else: + events.append({"event_type": "skipped", "phase": previous, "next_phase": nxt}) + + self._begin_phase_locked(nxt, run=auto) + events.append({ + "event_type": "phase_started", + "phase": nxt, + "auto_started": bool(auto), + "duration_seconds": int(self.phase_total), + }) + return events + + def _tick(self) -> None: + """Advance the state machine and publish. Safe to call from any thread.""" + events: List[Dict[str, Any]] = [] + with self.state_lock: + # A phase can only roll over once per tick in practice, but loop + # defensively in case the process was suspended for a long time. + for _ in range(8): + if not self.running or self.deadline is None: + break + if time.monotonic() < self.deadline: + break + events.extend(self._advance_locked(completed=True)) + for event in events: + self._publish_event(event) + if events: + self._sync_pin() + self._publish_state(force=True) + else: + self._publish_state() + + # ── Commands ──────────────────────────────────────────────────────────────── + + def _apply_command(self, command: str, payload: Dict[str, Any]) -> None: + """Apply a normalised command plus any JSON options that came with it.""" + command = (command or "").strip().upper().replace("-", "_").replace(" ", "_") + events: List[Dict[str, Any]] = [] + + with self.state_lock: + # Duration overrides that arrive alongside the command. + for field, *_ in _NUMBER_FIELDS: + if field in payload: + self._set_number_locked(field, payload[field]) + label = payload.get("label") + if label is not None: + # Bounded exactly like the task label: it lands in the same + # trim-a-character-at-a-time loop in the renderer, so an + # unbounded string is a per-frame cost, and a non-string would + # reach the draw call untouched. + label = str(label).strip()[:32] or None + explicit = payload.get("duration_minutes") or payload.get("minutes") + duration = None + if explicit is not None: + try: + duration = max(1.0, float(explicit)) * 60.0 + except (TypeError, ValueError): + duration = None + if duration is None and payload.get("duration_seconds") is not None: + try: + duration = max(1.0, float(payload["duration_seconds"])) + except (TypeError, ValueError): + duration = None + + if command in ("START", "ON", "PLAY"): + started = True + phase = str(payload.get("phase", "") or "").strip().lower() + if phase in _PHASE_ORDER: + self._begin_phase_locked(phase, duration, run=True, label=label) + elif self.phase == PHASE_IDLE or self.remaining <= 0: + self._begin_phase_locked(PHASE_WORK, duration, run=True, label=label) + elif not self.running: + self._resume_locked() + # Resuming keeps the phase it was holding, so nothing else + # would pick up a label sent with the same message. + if label: + self.custom_label = label + elif duration is not None: + self._begin_phase_locked(self.phase, duration, run=True, label=label) + else: + # Already running with nothing to restart. Pressing Start + # mid-session must not re-fire "started" automations, so + # emit no event — but an explicit label is a real change + # the caller asked for, so apply it rather than drop it. + # Settings in the same payload were applied above. + if label: + self.custom_label = label + self.logger.debug("START ignored: timer already running") + started = False + if started: + events.append({"event_type": "started", "phase": self.phase, + "duration_seconds": int(self.phase_total)}) + + elif command in ("WORK", "FOCUS"): + self._begin_phase_locked(PHASE_WORK, duration, run=True, label=label) + events.append({"event_type": "started", "phase": PHASE_WORK, + "duration_seconds": int(self.phase_total)}) + + elif command in ("SHORT_BREAK", "BREAK"): + self._begin_phase_locked(PHASE_SHORT_BREAK, duration, run=True, label=label) + events.append({"event_type": "started", "phase": PHASE_SHORT_BREAK, + "duration_seconds": int(self.phase_total)}) + + elif command == "LONG_BREAK": + self._begin_phase_locked(PHASE_LONG_BREAK, duration, run=True, label=label) + events.append({"event_type": "started", "phase": PHASE_LONG_BREAK, + "duration_seconds": int(self.phase_total)}) + + elif command == "PAUSE": + if self.running: + self.remaining = self._remaining_locked() + self.running = False + self.deadline = None + events.append({"event_type": "paused", "phase": self.phase, + "remaining_seconds": int(self.remaining)}) + + elif command == "RESUME": + if not self.running and self.phase != PHASE_IDLE: + self._resume_locked() + events.append({"event_type": "resumed", "phase": self.phase, + "remaining_seconds": int(self.remaining)}) + + elif command in ("TOGGLE", "PAUSE_RESUME"): + if self.phase == PHASE_IDLE: + self._begin_phase_locked(PHASE_WORK, duration, run=True, label=label) + events.append({"event_type": "started", "phase": PHASE_WORK, + "duration_seconds": int(self.phase_total)}) + elif self.running: + self.remaining = self._remaining_locked() + self.running = False + self.deadline = None + events.append({"event_type": "paused", "phase": self.phase, + "remaining_seconds": int(self.remaining)}) + else: + self._resume_locked() + events.append({"event_type": "resumed", "phase": self.phase, + "remaining_seconds": int(self.remaining)}) + + elif command in ("STOP", "OFF"): + self._go_idle_locked() + self.alert_until = 0.0 + events.append({"event_type": "stopped"}) + + elif command in ("RESET", "CLEAR"): + self._go_idle_locked(reset_counters=True) + self.task = "" + self.alert_until = 0.0 + events.append({"event_type": "reset"}) + + elif command in ("SKIP", "NEXT"): + if self.phase != PHASE_IDLE: + events.extend(self._advance_locked(completed=False)) + + elif command in ("SET", "CONFIGURE", ""): + # Duration-only payload — already applied above. + pass + + else: + self.logger.debug("Ignoring unrecognised command: %r", command) + return + + for event in events: + self._publish_event(event) + self._sync_pin() + self._publish_state(force=True) + with self.state_lock: + self.logger.info("Pomodoro %s → phase=%s running=%s remaining=%s", + command, self.phase, self.running, + _fmt_clock(self._remaining_locked())) + + def _resume_locked(self) -> None: + """Continue a paused phase from where it stopped.""" + self.remaining = max(1.0, self.remaining) + self.running = True + self.deadline = time.monotonic() + self.remaining + + def _set_number_locked(self, field: str, raw: Any) -> bool: + """Apply one of the numeric settings; returns True when it changed.""" + spec = next((s for s in _NUMBER_FIELDS if s[0] == field), None) + if spec is None: + return False + _, _, low, high, _, _ = spec + try: + value = int(round(float(raw))) + except (TypeError, ValueError): + self.logger.warning("Invalid value for %s: %r", field, raw) + return False + value = max(low, min(high, value)) + if getattr(self, field) == value: + return False + setattr(self, field, value) + # Keep the idle screen (and a not-yet-started phase) in step with the + # new duration so HA edits are visible immediately. + if self.phase == PHASE_IDLE and field == "work_minutes": + self.phase_total = self.remaining = float(value * 60) + elif (not self.running and self.phase != PHASE_IDLE + and abs(self.remaining - self.phase_total) < 0.5): + phase_field = { + PHASE_WORK: "work_minutes", + PHASE_SHORT_BREAK: "short_break_minutes", + PHASE_LONG_BREAK: "long_break_minutes", + }.get(self.phase) + if phase_field == field: + self.phase_total = self.remaining = float(value * 60) + if field == "sessions_before_long_break": + self.cycle_position = min(self.cycle_position, value) + return True + + def _parse_command(self, raw: bytes) -> Tuple[Optional[str], Dict[str, Any]]: + """Return (command, options) for a command-topic payload. + + Accepts a bare string (``START``) or a JSON object such as + ``{"command": "start", "phase": "work", "duration_minutes": 50}``. + Returns (None, {}) for anything unrecognised so a stray message can + never be mistaken for a stop. + """ + try: + text = raw.decode("utf-8").strip() + except (UnicodeDecodeError, AttributeError): + return None, {} + if not text: + return None, {} + + try: + data = json.loads(text) + except (json.JSONDecodeError, ValueError): + return text, {} + + if isinstance(data, dict): + command = data.get("command") or data.get("action") or data.get("state") + if command is None: + # A payload with only settings is a configuration update. + if any(f in data for f in (f for f, *_ in _NUMBER_FIELDS)): + return "SET", data + return None, {} + return str(command), data + if isinstance(data, (int, float)): + return ("START" if data else "STOP"), {} + return str(data), {} + + # ── BasePlugin interface ──────────────────────────────────────────────────── + + def update(self) -> None: + # No network data to fetch; just keep the state machine honest even if + # the timer thread is not running (e.g. under the safety harness). + """Keep the state machine honest even when the timer thread is not + running (the safety harness constructs the plugin without enabling it). + There is no network data to fetch. + """ + self._tick() + + def display(self, force_clear: bool = False) -> bool: + """Render the current frame. Ticks first so a phase that expired between + renders rolls over before it is drawn. + """ + try: + self._tick() + snapshot = self._snapshot() + width, height = self._panel_size() + canvas, draw = self._frame(width, height) + self._render(draw, width, height, snapshot) + self.display_manager.image = canvas + self.display_manager.draw = draw + self.display_manager.update_display() + return True + except Exception as e: + self.logger.error("Error displaying pomodoro timer: %s", e, exc_info=True) + try: + self.display_manager.draw_text("Pomodoro Error", x=2, y=2, color=(255, 0, 0)) + self.display_manager.update_display() + except Exception: + self.logger.debug("Fallback display failed", exc_info=True) + return False + + def get_display_duration(self) -> float: + try: + return float(self.config.get("display_duration", 10)) + except (TypeError, ValueError): + return 10.0 + + def on_enable(self) -> None: + super().on_enable() + if self.timer_thread is None or not self.timer_thread.is_alive(): + self.timer_stop_event.clear() + self.timer_thread = threading.Thread( + target=self._timer_loop, name=f"{self.plugin_id}-timer", daemon=True) + self.timer_thread.start() + if self.auto_start_on_enable: + with self.state_lock: + idle = self.phase == PHASE_IDLE + if idle: + self._apply_command("START", {}) + if self.mqtt_enabled and mqtt is not None: + if self.mqtt_thread is None or not self.mqtt_thread.is_alive(): + self.mqtt_stop_event.clear() + self.mqtt_thread = threading.Thread( + target=self._mqtt_loop, name=f"{self.plugin_id}-mqtt", daemon=True) + self.mqtt_thread.start() + self.logger.info("MQTT thread started") + + def on_disable(self) -> None: + super().on_disable() + self._graceful_shutdown() + + def cleanup(self) -> None: + self._graceful_shutdown() + self.logger.info("Pomodoro timer cleaned up") + + def on_config_change(self, new_config: Dict[str, Any]) -> None: + """Reload every setting, clearing anything retained under topics we are + about to abandon, and reconnect only if the broker wiring changed. + """ + super().on_config_change(new_config) + previous = (self.mqtt_enabled, self.mqtt_host, self.mqtt_port, self.mqtt_username, + self.mqtt_password, self.command_topic, self.state_topic, + self.ha_discovery, self.discovery_prefix) + was_idle = self.phase == PHASE_IDLE + + # Anything retained under the OLD topics has to be cleared before + # _load_settings overwrites them, while we still know where it went. + # A plain restart deliberately leaves discovery retained — that is what + # lets HA keep the device across reboots, with the LWT marking it + # unavailable in the meantime — but these two cases orphan it: + # + # discovery off the config topics would linger and HA would keep + # entities for a device that no longer announces itself + # prefix change the configs are republished under the new prefix, so + # the ones at the old prefix are never touched again + # + # A topic change is different: the discovery path is keyed on the prefix + # and plugin id, so republishing overwrites it in place and HA follows. + # What it does strand is the retained payloads on the state topics being + # left behind, so those get cleared instead. command_topic is the base + # every other published topic derives from, and state_topic is settable + # on its own, so either one moving abandons something. + new_prefix = str(new_config.get("discovery_prefix", "homeassistant")) + new_command = str(new_config.get("command_topic", "ledmatrix/pomodoro/set")) + new_state = str(new_config.get("state_topic", "ledmatrix/pomodoro/state")) + discovery_off = not bool(new_config.get("ha_discovery", True)) + if self.ha_discovery and (discovery_off or new_prefix != self.discovery_prefix): + self._remove_discovery() + if new_command != self.command_topic or new_state != self.state_topic: + self._clear_retained_state() + + self._load_settings(new_config) + self._font_cache = {} + self._ttf_path_resolved = False + if was_idle: + with self.state_lock: + self._go_idle_locked() + + current = (self.mqtt_enabled, self.mqtt_host, self.mqtt_port, self.mqtt_username, + self.mqtt_password, self.command_topic, self.state_topic, + self.ha_discovery, self.discovery_prefix) + if previous != current: + self._graceful_shutdown() + self.on_enable() + else: + self._publish_state(force=True) + + def get_info(self) -> Dict[str, Any]: + """State plus connection details for the web UI.""" + info = super().get_info() + info.update(self._snapshot()) + info.update({ + "mqtt_enabled": self.mqtt_enabled, + "mqtt_connected": self.mqtt_connected, + "mqtt_host": self.mqtt_host, + "command_topic": self.command_topic, + "ha_discovery": self.ha_discovery, + }) + return info + + # ── State snapshot ────────────────────────────────────────────────────────── + + def _snapshot(self) -> Dict[str, Any]: + """A consistent view of the timer for rendering and publishing. + + Taken under the lock so a frame can never mix a phase from before a + rollover with a countdown from after it. + """ + with self.state_lock: + remaining = self._remaining_locked() + return { + "phase": self.phase, + "status": ("idle" if self.phase == PHASE_IDLE + else "running" if self.running else "paused"), + "running": self.running, + "remaining_seconds": int(round(remaining)), + "remaining": _fmt_clock(remaining), + "phase_total_seconds": int(self.phase_total), + "elapsed_fraction": (0.0 if self.phase_total <= 0 + else max(0.0, min(1.0, 1.0 - remaining / self.phase_total))), + "completed_sessions": self.completed_sessions, + "cycle_position": self.cycle_position, + "sessions_before_long_break": self.sessions_before_long_break, + "work_minutes": self.work_minutes, + "short_break_minutes": self.short_break_minutes, + "long_break_minutes": self.long_break_minutes, + "task": self.task, + # A task name is what you actually want to see; the phase is + # already carried by the colour, so the task takes the label + # row rather than fighting for a second one on a 32px panel. + "label": (self.custom_label or self.task + or self._phase_label(self.phase, self.running)), + "alerting": time.monotonic() < self.alert_until, + "alert_color": self.alert_color, + } + + def _phase_label(self, phase: str, running: bool) -> str: + """The on-screen name for a phase, or the paused label when held.""" + if phase == PHASE_IDLE: + return self.idle_label + if not running and self.paused_label: + return self.paused_label + return { + PHASE_WORK: self.work_label, + PHASE_SHORT_BREAK: self.short_break_label, + PHASE_LONG_BREAK: self.long_break_label, + }.get(phase, self.work_label) + + def _phase_color(self, phase: str) -> Tuple[int, int, int]: + """The configured colour for a phase.""" + return { + PHASE_WORK: self.work_color, + PHASE_SHORT_BREAK: self.short_break_color, + PHASE_LONG_BREAK: self.long_break_color, + }.get(phase, self.idle_color) + + # ── Rendering ─────────────────────────────────────────────────────────────── + + def _panel_size(self) -> Tuple[int, int]: + """The panel's logical size, however the core exposes it.""" + dm = self.display_manager + width = getattr(dm, "width", None) or getattr(getattr(dm, "matrix", None), "width", 128) + height = getattr(dm, "height", None) or getattr(getattr(dm, "matrix", None), "height", 32) + return int(width), int(height) + + def _frame(self, width: int, height: int): + """Return (image, draw) to render into. + + Reuses the display manager's own buffer when it is at least panel-sized + so anything drawn past the edge stays visible to the safety harness's + bounds check, instead of being silently clipped by a fresh canvas. + """ + canvas = getattr(self.display_manager, "image", None) + if (not isinstance(canvas, Image.Image) or canvas.mode != "RGB" + or canvas.size[0] < width or canvas.size[1] < height): + canvas = Image.new("RGB", (width, height), (0, 0, 0)) + return canvas, ImageDraw.Draw(canvas) + draw = getattr(self.display_manager, "draw", None) + if draw is None: + draw = ImageDraw.Draw(canvas) + return canvas, draw + + def _ttf_path(self) -> Optional[str]: + """Best available TrueType font path, or None to use the built-ins.""" + if self._ttf_path_resolved: + return self._ttf_path_cache + self._ttf_path_resolved = True + self._ttf_path_cache = None + + configured = self.font_path.strip() + if configured: + candidates = [configured] + if not os.path.isabs(configured): + candidates.append(os.path.join(os.getcwd(), configured)) + root = Path(__file__).resolve().parent.parent.parent + candidates.append(str(root / configured)) + for candidate in candidates: + if os.path.exists(candidate): + self._ttf_path_cache = candidate + return self._ttf_path_cache + self.logger.warning("Font not found: %s — falling back to the default font", + configured) + + # Borrow whatever TTF the core already loaded (usually PressStart2P). + # PIL's built-in bitmap font also carries a `.path`, but it is a file + # object rather than a filename — only take real paths. + for attr in ("small_font", "regular_font", "extra_small_font"): + path = getattr(getattr(self.display_manager, attr, None), "path", None) + if isinstance(path, (str, bytes, os.PathLike)) and os.path.exists(path): + self._ttf_path_cache = os.fspath(path) + return self._ttf_path_cache + return None + + def _font(self, size: int): + """A TTF at `size` px, cached; None when no TTF is available.""" + size = max(5, int(size)) + if size in self._font_cache: + return self._font_cache[size] + path = self._ttf_path() + font = None + if path: + try: + font = ImageFont.truetype(path, size) + except Exception as e: + self.logger.debug("Could not load %s @ %dpx: %s", path, size, e) + self._font_cache[size] = font + return font + + def _fallback_font(self, max_height: int): + """A built-in font for when no TrueType face is available.""" + dm = self.display_manager + if max_height >= 8: + return getattr(dm, "small_font", None) or getattr(dm, "regular_font", None) + return getattr(dm, "extra_small_font", None) or getattr(dm, "small_font", None) + + @staticmethod + def _measure(draw, text: str, font) -> Tuple[int, int, int, int]: + """(width, height, x_offset, y_offset) of `text` in `font`.""" + try: + x0, y0, x1, y1 = draw.textbbox((0, 0), text, font=font) + return x1 - x0, y1 - y0, x0, y0 + except Exception: + approx = 6 * len(text) + return approx, 8, 0, 0 + + def _fit_font(self, draw, text: str, max_w: int, max_h: int): + """Largest available font that renders `text` inside (max_w, max_h).""" + if self.font_size: + return self._font(self.font_size) or self._fallback_font(max_h) + if self._ttf_path() is None: + font = self._fallback_font(max_h) + if font is not None and max_h < 8: + return getattr(self.display_manager, "extra_small_font", font) + return font + best = None + low, high = 5, max(5, min(max_h, 96)) + while low <= high: + mid = (low + high) // 2 + font = self._font(mid) + if font is None: + break + tw, th, _, _ = self._measure(draw, text, font) + if tw <= max_w and th <= max_h: + best = font + low = mid + 1 + else: + high = mid - 1 + return best or self._font(5) or self._fallback_font(max_h) + + def _draw_in_box(self, draw, text: str, font, box: Tuple[int, int, int, int], + color: Tuple[int, int, int], align: str = "center") -> None: + """Draw `text` centred (or left-aligned) inside box=(x, y, w, h). + + Text that still doesn't fit at the smallest font is truncated rather + than allowed to spill past the panel edge. + """ + if not text or font is None: + return + bx, by, bw, bh = box + tw, th, ox, oy = self._measure(draw, text, font) + while tw > bw and len(text) > 1: + text = text[:-1] + tw, th, ox, oy = self._measure(draw, text, font) + if tw > bw or th > bh: + return + x = bx if align == "left" else bx + max(0, (bw - tw) // 2) + y = by + max(0, (bh - th) // 2) + + # PIL antialiases TrueType text, which on an LED panel shows up as + # fringe pixels at partial brightness around every glyph. The label face + # is a pixel font rendered at integer sizes, so that grey is pure + # artefact — render to a mask, threshold it, and paint one flat colour + # so every lit pixel is fully lit. + try: + mask = Image.new("L", (tw, th), 0) + ImageDraw.Draw(mask).text((-ox, -oy), text, font=font, fill=255) + draw.bitmap((x, y), mask.point(lambda v: 255 if v >= 128 else 0), + fill=color) + except Exception as e: + self.logger.debug("Crisp text failed, falling back: %s", e) + draw.text((x - ox, y - oy), text, font=font, fill=color) + + def _render(self, draw, width: int, height: int, snap: Dict[str, Any]) -> None: + """Compose one frame: background, phase content, then the indicator.""" + phase = snap["phase"] + accent = self._phase_color(phase) + if snap["status"] == "paused": + accent = (_dim(_desaturate(accent, 0.6), 0.75) + if self.paused_style == "desaturate" else self.paused_color) + background = self.background_color + text_color = accent if self.color_mode == "phase" else self.time_color + + # Completion flash: swap foreground and background twice a second. + if snap["alerting"] and self.alert_flash: + if int(time.monotonic() * 2) % 2 == 0: + flash = snap["alert_color"] or accent + background = flash + text_color = self.background_color + accent = self.background_color + + draw.rectangle([0, 0, width - 1, height - 1], fill=background) + + label = snap["label"] + remaining = max(0.0, 1.0 - snap["elapsed_fraction"]) + # Hold the label back so the eye lands on the countdown first. It is + # still plainly the phase colour, just not competing at full brightness. + # During the alert flash everything is already inverted, so leave it. + label_color = accent if snap["alerting"] else _dim(accent, 0.7) + + # The perimeter ring lives on the outermost pixels, so the content + # box steps in to clear it. Every other style sits below the content. + ring = self.progress_style == "perimeter" + inset = 2 if ring else 0 + strip_h = 0 if ring else self._strip_height(height) + dot_h = self._dot_size(height) if self.show_session_dots else 0 + + box = (inset, inset, width - 2 * inset, height - 2 * inset - strip_h) + if width >= 192 and height <= 48: + self._render_wide(draw, box, snap, label, text_color, accent, + label_color, dot_h) + else: + self._render_stacked(draw, box, snap, label, text_color, accent, + label_color, dot_h) + + if ring: + self._draw_perimeter(draw, width, height, remaining, accent) + elif strip_h: + self._draw_strip(draw, 0, height - strip_h, width, strip_h, remaining, accent) + + def _strip_height(self, height: int) -> int: + if self.progress_style == "none": + return 0 + return 3 if height >= 64 else 2 + + @staticmethod + def _dot_size(height: int) -> int: + return 4 if height >= 64 else 3 + + def _render_stacked(self, draw, box, snap, label, text_color, accent, + label_color, dot_h) -> None: + """Label above the countdown, for panels that are not extremely wide.""" + bx, by, bw, bh = box + label_h = (10 if bh >= 54 else 7) if (self.show_phase_label and label) else 0 + label_font = self._fit_font(draw, label, bw - 2, label_h) if label_h else None + + # On a 32px panel a dedicated dots row costs a quarter of the height the + # digits could have. When the label leaves enough width, the dots tuck in + # beside it instead and the countdown gets those rows back. + dots_inline = False + if dot_h and label_font is not None and bh < 40: + label_w = self._measure(draw, label, label_font)[0] + if label_w + self._dots_span(snap, dot_h) + 6 <= bw: + dots_inline = True + + top = by + label_h + (1 if label_h else 0) + bottom = by + bh - (0 if (dots_inline or not dot_h) else dot_h + 1) + # On a very short panel the label and the countdown cannot both fit. + # The countdown is the point, so the label goes. Flooring box_h instead + # would push the digits straight off the bottom edge. + if label_h and bottom - top < 7: + label_h = 0 + label_font = None + top = by + bottom = by + bh - (0 if (dots_inline or not dot_h) else dot_h + 1) + box_h = max(1, bottom - top) + + if label_h: + self._draw_in_box(draw, label, label_font, (bx + 1, by, bw - 2, label_h), + label_color) + + self._draw_time(draw, snap["remaining"], (bx + 2, top, bw - 4, box_h), text_color) + + if dot_h and dots_inline: + self._draw_session_dots(draw, bx, by + max(0, (label_h - dot_h) // 2), + bw - 1, dot_h, snap, accent, align="right") + elif dot_h: + self._draw_session_dots(draw, bx, bottom + 1, bw, dot_h, snap, accent, + align="center") + + @staticmethod + def _dots_span(snap: Dict[str, Any], size: int) -> int: + total = max(1, int(snap["sessions_before_long_break"])) + return total * (size + 1) - 1 + + def _render_wide(self, draw, box, snap, label, text_color, accent, + label_color, dot_h) -> None: + """Side-by-side layout for very wide panels (e.g. 256x32).""" + bx, by, bw, bh = box + left_w = max(40, int(bw * 0.30)) + label_h = (min(12, bh // 2) if (self.show_phase_label and label) else 0) + + if label_h: + label_font = self._fit_font(draw, label, left_w - 4, label_h) + self._draw_in_box(draw, label, label_font, (bx + 2, by + 1, left_w - 4, label_h), + label_color, align="left") + if dot_h: + dots_y = by + ((label_h + 3) if label_h else max(1, (bh - dot_h) // 2)) + dots_y = min(dots_y, by + max(0, bh - dot_h - 1)) + self._draw_session_dots(draw, bx + 2, dots_y, left_w - 4, dot_h, snap, accent, + align="left") + + self._draw_time(draw, snap["remaining"], + (bx + left_w + 2, by, bw - left_w - 4, max(1, bh - 1)), text_color) + + def _draw_session_dots(self, draw, x: int, y: int, avail_w: int, size: int, + snap: Dict[str, Any], accent: Tuple[int, int, int], + align: str = "center") -> None: + total = max(1, int(snap["sessions_before_long_break"])) + filled = max(0, min(total, int(snap["cycle_position"]))) + gap = 1 + while total * (size + gap) - gap > avail_w and size > 1: + size -= 1 + span = total * (size + gap) - gap + if span > avail_w or size < 1: + return + if align == "left": + start = x + elif align == "right": + start = x + max(0, avail_w - span) + else: + start = x + max(0, (avail_w - span) // 2) + + dim = _dim(accent, 0.28) + # The session you are actually in gets its own state, so the row reads + # as "two done, on the third, one to go" rather than just a count. + active = filled if snap["phase"] == PHASE_WORK else -1 + pulse_on = int(time.monotonic()) % 2 == 0 + for index in range(total): + left = start + index * (size + gap) + box = [left, y, left + size - 1, y + size - 1] + if index < filled: + draw.rectangle(box, fill=accent) + elif index == active: + lit = accent if (pulse_on or not self.pulse_active_pip) else dim + draw.rectangle(box, fill=lit) + elif size >= 3: + # Hollow: clearly "not yet" without reading as a dim filled dot. + draw.rectangle(box, outline=dim) + else: + draw.rectangle(box, fill=dim) + + # ── Countdown digits ──────────────────────────────────────────────────────── + + def _draw_time(self, draw, text: str, box: Tuple[int, int, int, int], + color: Tuple[int, int, int]) -> None: + if self.digit_style == "seven_segment": + if self._draw_seven_segment(draw, text, box, color): + return + # Too little room for legible segments — fall back to the pixel font. + bx, by, bw, bh = box + font = self._fit_font(draw, text, bw, bh) + self._draw_in_box(draw, text, font, box, color) + + def _seven_segment_metrics(self, text: str, bw: int, bh: int): + """Largest segment geometry for `text` that fits (bw, bh), or None. + + Returns (digit_h, digit_w, stroke, gap, colon_w, total_w). + """ + digits = sum(c.isdigit() for c in text) + colons = sum(c == ":" for c in text) + if not digits: + return None + for h in range(bh, 4, -1): + # A heavier stroke puts more lit pixels on the panel, which is what + # legibility comes down to at a distance. + stroke = max(1, round(h * 0.17)) + gap = max(1, round(h * 0.10)) + colon_w = stroke + spare = bw - colons * colon_w - (digits + colons - 1) * gap + if spare <= 0: + continue + width_cap = spare // digits + # A digit narrower than two strokes plus a gap has no interior left, + # so the vertical segments would fuse into a solid block. The height + # is usually the binding constraint, so spend spare width on wider + # digits — longer horizontal segments read further away. + digit_w = min(round(h * 0.78), width_cap) + if digit_w < 2 * stroke + 1 or h < 3 * stroke + 2: + continue + # Keep the classic clock-radio proportion. Without this a tall, + # narrow box drives the digits to a sliver and a 0 stops reading as + # a digit at all — it becomes two stacked boxes. Falling through to + # a shorter h trades height for a shape that still reads. + if digit_w < round(h * 0.64): + continue + total = digits * digit_w + colons * colon_w + (digits + colons - 1) * gap + return h, digit_w, stroke, gap, colon_w, total + return None + + def _draw_seven_segment(self, draw, text: str, box: Tuple[int, int, int, int], + color: Tuple[int, int, int]) -> bool: + """Draw `text` as seven-segment digits. False if it cannot be done legibly.""" + bx, by, bw, bh = box + metrics = self._seven_segment_metrics(text, bw, bh) + if metrics is None: + return False + h, digit_w, stroke, gap, colon_w, total = metrics + + # Unlit segments stay faintly visible, the way the dark segments of a + # real LED clock are: it anchors the digits so they don't appear to + # jump around as the numbers change. At a one-pixel stroke there is no + # room for the effect to read as anything but noise, so it drops out. + ghost = (_dim(color, 0.17) + if self.show_ghost_segments and stroke >= 2 else None) + x = bx + max(0, (bw - total) // 2) + y = by + max(0, (bh - h) // 2) + + for index, char in enumerate(text): + if index: + x += gap + if char == ":": + dot = max(1, colon_w) + for cy in (y + h // 3 - dot // 2, y + 2 * h // 3 - dot // 2): + draw.rectangle([x, cy, x + dot - 1, cy + dot - 1], fill=color) + x += colon_w + continue + if char.isdigit(): + self._draw_digit(draw, char, x, y, digit_w, h, stroke, color, ghost) + x += digit_w + return True + + # Lit segments per digit, labelled clockwise from the top bar plus the middle. + _SEGMENTS = { + "0": "abcdef", "1": "bc", "2": "abdeg", "3": "abcdg", "4": "bcfg", + "5": "acdfg", "6": "acdefg", "7": "abc", "8": "abcdefg", "9": "abcdfg", + } + + @staticmethod + def _segment_boxes(x: int, y: int, w: int, h: int, t: int) -> Dict[str, Tuple]: + """Pixel rectangles for each of the seven segments of a digit cell.""" + mid = y + (h - t) // 2 + # Verticals stop at the middle bar so corners meet cleanly. + upper_h = mid - y + t + lower_h = (y + h) - mid + return { + "a": (x, y, w, t), + "b": (x + w - t, y, t, upper_h), + "c": (x + w - t, mid, t, lower_h), + "d": (x, y + h - t, w, t), + "e": (x, mid, t, lower_h), + "f": (x, y, t, upper_h), + "g": (x, mid, w, t), + } + + def _draw_digit(self, draw, char: str, x: int, y: int, w: int, h: int, t: int, + color: Tuple[int, int, int], + ghost: Optional[Tuple[int, int, int]]) -> None: + """Draw one seven-segment digit, lit segments over the ghost layer.""" + lit = self._SEGMENTS.get(char, "") + boxes = self._segment_boxes(x, y, w, h, t) + + if ghost is not None: + for name, (sx, sy, sw, sh) in boxes.items(): + if name not in lit: + draw.rectangle([sx, sy, sx + sw - 1, sy + sh - 1], fill=ghost) + elif char == "1": + # A 1 lights only the right-hand verticals, so it hangs off the edge + # of its cell and opens a ragged gap before the next digit — "11:11" + # reads as four scattered bars. Centring it in the cell evens the + # spacing without changing cell widths, so digits never jitter as the + # time changes. With the ghost layer on, the unlit 8 already anchors + # the cell and a right-hung 1 is what real hardware looks like. + boxes = self._segment_boxes(x + (w - t) // 2, y, t, h, t) + + for name in lit: + sx, sy, sw, sh = boxes[name] + draw.rectangle([sx, sy, sx + sw - 1, sy + sh - 1], fill=color) + + # ── Burndown indicator ────────────────────────────────────────────────────── + + def _draw_perimeter(self, draw, width: int, height: int, remaining: float, + accent: Tuple[int, int, int]) -> None: + """Trace the remaining time around the edge of the panel. + + The whole border is lit at the start of a phase and drains clockwise + from the top-left, so the ring itself is the countdown. It costs no + interior space, which is what lets the digits be as large as they are. + """ + path = [] + path += [(x, 0) for x in range(width)] + path += [(width - 1, y) for y in range(1, height)] + path += [(x, height - 1) for x in range(width - 2, -1, -1)] + path += [(0, y) for y in range(height - 2, 0, -1)] + if not path: + return + + track = _dim(accent, 0.14) + for point in path: + draw.point(point, fill=track) + + lit = int(round(max(0.0, min(1.0, remaining)) * len(path))) + for point in path[:lit]: + draw.point(point, fill=accent) + if 0 < lit <= len(path): + # A brighter head makes the direction of travel readable, and gives + # the last minute of a phase something to watch. + draw.point(path[lit - 1], fill=_lighten(accent, 0.55)) + + def _draw_strip(self, draw, x: int, y: int, width: int, height: int, + remaining: float, accent: Tuple[int, int, int]) -> None: + if self.progress_style == "segments": + self._draw_segments(draw, x, y, width, height, remaining, accent) + return + self._draw_bar(draw, x, y, width, height, remaining, accent) + + def _draw_bar(self, draw, x: int, y: int, width: int, height: int, + remaining: float, accent: Tuple[int, int, int]) -> None: + draw.rectangle([x, y, x + width - 1, y + height - 1], fill=_dim(accent, 0.18)) + lit = int(round(max(0.0, min(1.0, remaining)) * width)) + if lit <= 0: + return + draw.rectangle([x, y, x + lit - 1, y + height - 1], fill=accent) + draw.rectangle([x + lit - 1, y, x + lit - 1, y + height - 1], + fill=_lighten(accent, 0.55)) + + def _draw_segments(self, draw, x: int, y: int, width: int, height: int, + remaining: float, accent: Tuple[int, int, int]) -> None: + """A row of discrete blocks that go out one by one, like a fuel gauge.""" + count = max(4, min(32, width // 8)) + gap = 1 + block = (width - (count - 1) * gap) // count + if block < 1: + # Too narrow for discrete blocks. Fall back for this frame only — + # rewriting self.progress_style here would silently overwrite the + # user's setting from the render thread. + self._draw_bar(draw, x, y, width, height, remaining, accent) + return + span = count * block + (count - 1) * gap + start = x + max(0, (width - span) // 2) + lit = int(round(max(0.0, min(1.0, remaining)) * count)) + dark = _dim(accent, 0.18) + for index in range(count): + left = start + index * (block + gap) + draw.rectangle([left, y, left + block - 1, y + height - 1], + fill=accent if index < lit else dark) + + # ── Display pinning ───────────────────────────────────────────────────────── + + def _sync_pin(self) -> None: + """Take over (or release) the panel to match the timer's state. + + Called from the timer thread, the MQTT thread, and the display thread, + so the whole compare-and-set — including writing the request — runs + under state_lock. Otherwise two callers can interleave and publish the + requests out of order (a stop landing after a start), and because the + method early-returns once _pinned matches, the panel would stay stuck + that way until the next transition. + """ + with self.state_lock: + active = self.phase != PHASE_IDLE + alerting = time.monotonic() < self.alert_until + want = (self.pin_while_running and active) or alerting + if want == self._pinned: + return + self._pinned = want + request: Dict[str, Any] = { + "request_id": str(uuid.uuid4()), + "plugin_id": self.plugin_id, + "action": "start" if want else "stop", + "timestamp": time.time(), + } + if want: + request.update({"mode": "pomodoro", "duration": None, "pinned": True}) + try: + self.cache_manager.set("display_on_demand_request", request) + except Exception as e: + self.logger.debug("On-demand display request failed: %s", e) + self._pinned = not want # let the next tick retry + + def _timer_loop(self) -> None: + """Drive the state machine independently of the render loop, so the + timer keeps counting while another plugin owns the panel. + """ + while not self.timer_stop_event.wait(0.25): + try: + self._tick() + # Idempotent: releases the panel once the alert window closes + # or the pin setting is turned off. + self._sync_pin() + except Exception as e: + self.logger.error("Timer loop error: %s", e, exc_info=True) + self.logger.info("Timer loop stopped") + + # ── MQTT publishing ───────────────────────────────────────────────────────── + + def _publish(self, topic: str, payload: str, retain: bool = True) -> None: + if not self.mqtt_client or not self.mqtt_connected: + return + try: + self.mqtt_client.publish(topic, payload, retain=retain) + except Exception as e: + self.logger.debug("Publish to %s failed: %s", topic, e) + + def _publish_availability(self, online: bool, wait: bool = False) -> None: + """Publish the availability state. + + Pass wait=True when shutting down: the publish has to reach the broker + before we disconnect, otherwise it sits in paho's outbound queue and + dies with the client, leaving the device stuck 'online' in HA until the + keepalive lapses. + """ + client = self.mqtt_client + if client is None: + return + try: + info = client.publish(self.availability_topic, + "online" if online else "offline", + qos=1, retain=True) + if wait and info is not None and hasattr(info, "wait_for_publish"): + info.wait_for_publish(timeout=2.0) + except Exception as e: + self.logger.debug("Availability publish failed: %s", e) + + def _publish_event(self, event: Dict[str, Any]) -> None: + self.logger.info("Pomodoro event: %s", event.get("event_type")) + self._publish(self.event_topic, json.dumps(event), retain=False) + + def _publish_state(self, force: bool = False) -> None: + """Publish the full state, throttled to publish_interval while ticking.""" + if not self.mqtt_client or not self.mqtt_connected: + return + now = time.monotonic() + snap = self._snapshot() + key = (snap["phase"], snap["status"], snap["remaining"], + snap["completed_sessions"], snap["cycle_position"], snap["task"]) + if not force: + if key == self._last_published_key: + return + if now - self._last_publish < self.publish_interval: + return + self._last_publish = now + self._last_published_key = key + + self._publish(self.state_topic, "ON" if snap["phase"] != PHASE_IDLE else "OFF") + self._publish(self.status_topic, snap["status"]) + self._publish(self.phase_topic, snap["phase"]) + self._publish(self.remaining_topic, snap["remaining"]) + self._publish(self.remaining_secs_topic, str(snap["remaining_seconds"])) + self._publish(self.session_topic, str(snap["completed_sessions"])) + self._publish(self.task_topic, snap["task"]) + self._publish(self.attributes_topic, json.dumps({ + k: v for k, v in snap.items() if k != "alert_color" + })) + for field, *_ in _NUMBER_FIELDS: + self._publish(self.number_state_topics[field], str(getattr(self, field))) + + # ── HA MQTT auto-discovery ────────────────────────────────────────────────── + + def _discovery_uid(self) -> str: + return f"ledmatrix_pomodoro_{self.plugin_id}" + + def _discovery_device(self) -> Dict[str, Any]: + return { + "identifiers": [self._discovery_uid()], + "name": self.device_name, + "model": "Pomodoro Timer", + "manufacturer": "LEDMatrix", + "sw_version": _PLUGIN_VERSION, + } + + def _discovery_entities(self) -> List[Tuple[str, Dict[str, Any]]]: + """(discovery topic, config payload) for every entity we announce.""" + uid = self._discovery_uid() + device = self._discovery_device() + avail = [{"topic": self.availability_topic, + "payload_available": "online", + "payload_not_available": "offline"}] + common = {"availability": avail, "device": device} + entities: List[Tuple[str, Dict[str, Any]]] = [] + + entities.append((f"switch/{uid}/config", { + "name": "Timer", "unique_id": f"{uid}_switch", + "command_topic": self.command_topic, "state_topic": self.state_topic, + "payload_on": "START", "payload_off": "STOP", + "state_on": "ON", "state_off": "OFF", + "json_attributes_topic": self.attributes_topic, + "icon": "mdi:timer-play-outline", **common, + })) + + for key, name, payload, icon in _BUTTONS: + entities.append((f"button/{uid}_{key}/config", { + "name": name, "unique_id": f"{uid}_{key}", + "command_topic": self.command_topic, "payload_press": payload, + "icon": icon, **common, + })) + + for field, name, low, high, step, icon in _NUMBER_FIELDS: + entities.append((f"number/{uid}_{field}/config", { + "name": name, "unique_id": f"{uid}_{field}", + "command_topic": f"{self.topic_base}/{field}/set", + "state_topic": self.number_state_topics[field], + "min": low, "max": high, "step": step, "mode": "box", + "unit_of_measurement": None if field == "sessions_before_long_break" else "min", + "icon": icon, "entity_category": "config", **common, + })) + + entities.append((f"sensor/{uid}_phase/config", { + "name": "Phase", "unique_id": f"{uid}_phase", + "state_topic": self.phase_topic, "icon": "mdi:progress-clock", **common, + })) + entities.append((f"sensor/{uid}_status/config", { + "name": "Status", "unique_id": f"{uid}_status", + "state_topic": self.status_topic, "icon": "mdi:play-pause", **common, + })) + entities.append((f"sensor/{uid}_remaining/config", { + "name": "Time Remaining", "unique_id": f"{uid}_remaining", + "state_topic": self.remaining_topic, "icon": "mdi:timer-sand", **common, + })) + entities.append((f"sensor/{uid}_remaining_seconds/config", { + "name": "Seconds Remaining", "unique_id": f"{uid}_remaining_seconds", + "state_topic": self.remaining_secs_topic, + "unit_of_measurement": "s", "device_class": "duration", + "state_class": "measurement", "icon": "mdi:timer-outline", **common, + })) + entities.append((f"sensor/{uid}_sessions/config", { + "name": "Completed Sessions", "unique_id": f"{uid}_sessions", + "state_topic": self.session_topic, "state_class": "total_increasing", + "icon": "mdi:check-circle-outline", **common, + })) + entities.append((f"text/{uid}_task/config", { + "name": "Task", "unique_id": f"{uid}_task", + "command_topic": self.task_set_topic, "state_topic": self.task_topic, + "max": 32, "icon": "mdi:format-text", **common, + })) + entities.append((f"event/{uid}_event/config", { + "name": "Timer Event", "unique_id": f"{uid}_event", + "state_topic": self.event_topic, "event_types": _EVENT_TYPES, + "icon": "mdi:bell-ring-outline", **common, + })) + entities.append((f"binary_sensor/{uid}_connected/config", { + "name": "MQTT Connected", "unique_id": f"{uid}_connected", + "state_topic": self.availability_topic, + "payload_on": "online", "payload_off": "offline", + "device_class": "connectivity", "icon": "mdi:wifi", + "device": device, + })) + return entities + + def _publish_discovery(self) -> None: + """Announce every entity to Home Assistant.""" + if not self.ha_discovery or not self.mqtt_client: + return + for suffix, payload in self._discovery_entities(): + self._publish(f"{self.discovery_prefix}/{suffix}", json.dumps(payload)) + self.logger.info("Published HA MQTT discovery — device: %s", self.device_name) + + def _remove_discovery(self) -> None: + """Clear the retained discovery configs so HA drops the entities.""" + if not self.ha_discovery or not self.mqtt_client: + return + for suffix, _ in self._discovery_entities(): + self._publish(f"{self.discovery_prefix}/{suffix}", "") + + def _clear_retained_state(self) -> None: + """Empty the retained payloads on the topics we are about to abandon.""" + if not self.mqtt_client: + return + topics = [self.state_topic, self.status_topic, self.phase_topic, + self.remaining_topic, self.remaining_secs_topic, + self.session_topic, self.attributes_topic, self.availability_topic, + self.task_topic] + topics += list(self.number_state_topics.values()) + for topic in topics: + self._publish(topic, "") + + # ── MQTT callbacks ────────────────────────────────────────────────────────── + + def _on_mqtt_connect(self, client, userdata, flags, rc): # pylint: disable=unused-argument + """Subscribe and announce ourselves once the broker accepts us.""" + if rc != 0: + self.mqtt_connecting = False + self.mqtt_connected = False + self.logger.error("MQTT connect failed rc=%s", rc) + return + self.mqtt_connected = True + self.mqtt_connecting = False + self.mqtt_reconnect_delay = 1.0 + client.subscribe(self.command_topic, qos=1) + for topic in self.number_set_topics: + client.subscribe(topic, qos=1) + client.subscribe(self.task_set_topic, qos=1) + self._publish_availability(True) + self._publish_discovery() + self._publish_state(force=True) + self.logger.info("MQTT connected — subscribed to %s", self.command_topic) + + def _on_mqtt_disconnect(self, client, userdata, rc): # pylint: disable=unused-argument + """Mark the connection down so the supervisor loop redials.""" + self.mqtt_connected = False + self.mqtt_connecting = False + if rc != 0: + self.logger.warning("MQTT disconnected unexpectedly rc=%s", rc) + + def _on_mqtt_message(self, client, userdata, msg): # pylint: disable=unused-argument + """Route an incoming message to a duration setter or a command.""" + try: + field = self.number_set_topics.get(msg.topic) + if field: + try: + raw = msg.payload.decode("utf-8").strip() + except (UnicodeDecodeError, AttributeError): + return + with self.state_lock: + changed = self._set_number_locked(field, raw) + if changed: + self.logger.info("%s set to %s via MQTT", field, getattr(self, field)) + self._publish_state(force=True) + return + + if msg.topic == self.task_set_topic: + try: + task = msg.payload.decode("utf-8").strip()[:32] + except (UnicodeDecodeError, AttributeError): + return + with self.state_lock: + self.task = task + self.logger.info("Task set to %r", task) + self._publish_state(force=True) + return + + command, payload = self._parse_command(msg.payload) + if command is None: + self.logger.debug("Ignoring unrecognised payload on %s", msg.topic) + return + self._apply_command(command, payload if isinstance(payload, dict) else {}) + except Exception as e: + self.logger.error("Error handling MQTT message: %s", e, exc_info=True) + + # ── MQTT connect / loop ───────────────────────────────────────────────────── + + def _connect_mqtt(self) -> bool: + """Build and dial a client. False if it could not be reached.""" + try: + try: + self.mqtt_client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION1, + client_id=self.mqtt_client_id, clean_session=True) + except (TypeError, AttributeError): + self.mqtt_client = mqtt.Client( + client_id=self.mqtt_client_id, clean_session=True) + self.mqtt_client.on_connect = self._on_mqtt_connect + self.mqtt_client.on_disconnect = self._on_mqtt_disconnect + self.mqtt_client.on_message = self._on_mqtt_message + if self.mqtt_username: + self.mqtt_client.username_pw_set(self.mqtt_username, self.mqtt_password) + self.mqtt_client.will_set(self.availability_topic, payload="offline", + qos=1, retain=True) + self.mqtt_client.connect(self.mqtt_host, self.mqtt_port, keepalive=60) + return True + except Exception as e: + self.logger.error("MQTT connect error: %s", e) + return False + + def _teardown_client(self) -> None: + """Stop the network loop, disconnect, and release the paho client. + + The single place a client is torn down: the reconnect path, the loop's + error handler, and shutdown all funnel through here so a refused or + half-open connection can never leave a live client (and its network + thread) behind. Idempotent, and never raises. + """ + client, self.mqtt_client = self.mqtt_client, None + self.mqtt_connected = False + self.mqtt_connecting = False + if client is None: + return + try: + client.loop_stop() + except Exception as e: + self.logger.debug("MQTT loop_stop during teardown failed: %s", e) + try: + client.disconnect() + except Exception as e: + self.logger.debug("MQTT disconnect during teardown failed: %s", e) + + def _backoff(self) -> bool: + """Sleep out the current reconnect delay, then double it. + + Returns True when the plugin is shutting down and the loop should stop. + """ + wait = min(self.mqtt_reconnect_delay, self.mqtt_max_reconnect_delay) + self.logger.info("Retrying MQTT in %.0fs", wait) + if self.mqtt_stop_event.wait(wait): + return True + self.mqtt_reconnect_delay = min( + self.mqtt_reconnect_delay * 2, self.mqtt_max_reconnect_delay) + return False + + def _mqtt_loop(self) -> None: + """Supervise the connection: dial, wait for CONNACK, redial on failure + with backoff, and tear the client down on the way out. + """ + while not self.mqtt_stop_event.is_set(): + try: + if self.mqtt_connected: + if self.mqtt_stop_event.wait(1.0): + break + self.mqtt_reconnect_delay = 1.0 + continue + + # Release whatever is left from a refused or half-open attempt + # before dialing again, so its network thread can't outlive it. + self._teardown_client() + self.mqtt_connecting = True + if not self._connect_mqtt(): + self._teardown_client() + if self._backoff(): + break + continue + + client = self.mqtt_client + if client is None: # torn down from another thread + continue + client.loop_start() + # Wait for CONNACK before looping so we never open a second + # connection. If it never arrives, drop this client and retry + # rather than sitting here connecting forever. + self.mqtt_stop_event.wait(5.0) + if not self.mqtt_connected and not self.mqtt_stop_event.is_set(): + self.logger.warning("No CONNACK from %s:%s within 5s — retrying", + self.mqtt_host, self.mqtt_port) + self._teardown_client() + if self._backoff(): + break + except Exception as e: + self.logger.error("MQTT loop error: %s", e, exc_info=True) + self._teardown_client() + if self._backoff(): + break + + self._publish_availability(False) + self._teardown_client() + self.logger.info("MQTT loop stopped") + + def _graceful_shutdown(self) -> None: + """Stop both threads and release the MQTT client. Safe to call twice.""" + self.timer_stop_event.set() + if self.timer_thread and self.timer_thread.is_alive(): + self.timer_thread.join(timeout=2.0) + self.timer_thread = None + + # Say goodbye while the socket is still up — once we disconnect below, + # the loop thread's own offline publish has nothing to send on. + self._publish_availability(False, wait=True) + + self.mqtt_stop_event.set() + if self.mqtt_thread and self.mqtt_thread.is_alive(): + # Drop the connection so the loop wakes out of its wait instead of + # sitting there for the full timeout. + client = self.mqtt_client + if client is not None: + try: + client.disconnect() + except Exception as e: + self.logger.debug("MQTT disconnect during shutdown failed: %s", e) + self.mqtt_thread.join(timeout=5.0) + self.mqtt_thread = None + # Unconditional: the loop may have died (or never started) with a live + # client still attached. + self._teardown_client() diff --git a/plugins/pomodoro-timer/manifest.json b/plugins/pomodoro-timer/manifest.json new file mode 100644 index 0000000..08b1b73 --- /dev/null +++ b/plugins/pomodoro-timer/manifest.json @@ -0,0 +1,90 @@ +{ + "id": "pomodoro-timer", + "name": "Pomodoro Timer", + "version": "1.3.3", + "author": "ChuckBuilds", + "description": "A configurable Pomodoro focus/break timer for your matrix. Set the work and break lengths, then start, pause, skip, or reset it over MQTT — with Home Assistant auto-discovery so the whole timer shows up as a device with no YAML.", + "entry_point": "manager.py", + "class_name": "PomodoroTimerPlugin", + "category": "productivity", + "tags": [ + "pomodoro", + "timer", + "focus", + "productivity", + "mqtt", + "home-assistant" + ], + "display_modes": [ + "pomodoro" + ], + "compatible_versions": [ + ">=2.0.0" + ], + "versions": [ + { + "released": "2026-07-29", + "version": "1.3.3", + "notes": "Review fixes. A label sent with START is now applied when the command resumes a paused timer, which was the one remaining branch that silently dropped it. Labels from a command payload are bounded to 32 characters and coerced to text, matching the task label — an unbounded string was paying for itself in the renderer's trim-to-fit loop on every frame, and a non-string reached the draw call untouched. Renaming only the state topic now clears the retained payload on the one being abandoned; it is settable independently of the command topic, so the old topic would otherwise sit on the broker forever showing a timer that no longer exists.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-29", + "version": "1.3.2", + "notes": "Screenshots. The README now shows what the plugin actually looks like: a work session on a 128x64 panel, the six on-screen states side by side, the appearance options (the four burndown styles, unlit segments, pixel digits, the calm theme, desaturated pause), and the same session on all four panel sizes at one scale. Every panel in those images is real plugin output rendered through the safety harness at the true panel size and scaled with nearest-neighbour, not a mockup, and test/render_readme_assets.py regenerates them so they can't drift from the display. No functional change.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.3.1", + "notes": "Crispness and bounds. The phase label is now rendered through a thresholded mask instead of PIL's antialiased text, so a frame contains only the palette colours rather than a grey halo around every glyph — 26-36 distinct colours per frame drops to 6, all intentional. Fixes content drawn past the bottom edge on panels under about 16px tall: the countdown box had a hard 6px floor that pushed the digits off-panel once the label had taken its row, so the label is now dropped instead when both cannot fit. Swept 256 panel shapes from 32x8 to 384x128 with no overflow, no crashes, and no blank frames.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.3.0", + "notes": "Session pips now have three states rather than two: completed are solid, upcoming are hollow outlines, and the session you are actually in is picked out and slowly pulses (pulse_active_pip), so the row reads as 'two done, on the third' instead of just a count. New task label, settable from Home Assistant via an MQTT text entity or by publishing to /task/set, which takes the label row since the phase is already carried by the colour; cleared by RESET. New 'calm' colour theme preset (warm terracotta, sage, soft indigo) as an alternative to the punchy defaults, and paused_style: desaturate, which drains the phase's own hue instead of switching to amber. A label sent with START to an already-running timer is now applied instead of silently dropped.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.2.0", + "notes": "Legibility pass on the countdown. The unlit 'ghost' segments are now off by default: they look authentic but put a faint 8 behind every digit, so a 15 read as an 85 at a glance. The segment stroke is heavier and the digits are allowed to spend spare panel width, since height is usually the binding constraint and longer horizontal segments read further away. The phase label is drawn at 70% brightness so the eye lands on the time first. A 1 is centred in its cell rather than hanging off the right, which evens out the spacing of times like 11:11 without changing cell widths, so the digits never jitter as the time changes; with the ghost layer enabled it stays right-hung, which is what real hardware looks like.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.1.1", + "notes": "Review fixes. The segments indicator no longer rewrites the configured progress_style from the render thread when a panel is too narrow for discrete blocks; it falls back to a bar for that frame only. Changing the Home Assistant discovery prefix now clears the retained configs at the old prefix, which would otherwise leave permanently-unavailable entities in HA, and renaming the command topic clears the retained payloads on the state topics it abandons. Docstrings added across the state machine, render path, and MQTT plumbing.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.1.0", + "notes": "Redrawn display. The countdown is now rendered as geometric seven-segment digits sized to the panel rather than set in a pixel font, with faintly-lit unlit segments the way a real LED clock shows them; digit_style: pixel keeps the old look. New burndown indicator styles replace the fill-up progress bar: perimeter (default) drains a ring around the edge of the panel with a brighter leading pixel and costs no interior space, bar and segments sit along the bottom edge, none hides it. On panels under 40px tall the session dots tuck in beside the phase label instead of taking their own row, giving the digits that height. Config: show_progress_bar is replaced by progress_style; adds digit_style and show_ghost_segments.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.0.1", + "notes": "Reliability fixes from review. Centralise MQTT client teardown so a refused or half-open connection can no longer leave a live paho client and its network thread behind, and retry instead of stalling forever when no CONNACK arrives. Shut down cleanly even when the MQTT thread already died, and publish the offline availability state before the socket closes so Home Assistant doesn't sit on a stale 'online'. Serialise the display-pin compare-and-set under the state lock so concurrent callers can't strand the panel pinned or unpinned. START on an already-running timer no longer re-fires the 'started' event. Clear retained discovery topics when Home Assistant discovery is turned off.", + "ledmatrix_min_version": "2.0.0" + }, + { + "released": "2026-07-28", + "version": "1.0.0", + "notes": "Initial release. Configurable work/short-break/long-break durations with session dots and a phase progress bar, MQTT command topic (START/PAUSE/RESUME/TOGGLE/SKIP/STOP/RESET plus JSON), and Home Assistant MQTT auto-discovery covering a switch, five control buttons, four duration number boxes, phase/status/remaining/session sensors, a timer-event entity, and connectivity.", + "ledmatrix_min_version": "2.0.0" + } + ], + "default_duration": 10, + "icon": "fa-hourglass-half", + "license": "GPL-3.0", + "homepage": "https://github.com/ChuckBuilds/ledmatrix-plugins/tree/main/plugins/pomodoro-timer", + "config_schema": "config_schema.json", + "last_updated": "2026-07-29", + "stars": 0, + "downloads": 0, + "verified": true, + "screenshot": "https://raw.githubusercontent.com/ChuckBuilds/ledmatrix-plugins/main/plugins/pomodoro-timer/assets/hero.png" +} diff --git a/plugins/pomodoro-timer/requirements.txt b/plugins/pomodoro-timer/requirements.txt new file mode 100644 index 0000000..7fa2a0a --- /dev/null +++ b/plugins/pomodoro-timer/requirements.txt @@ -0,0 +1 @@ +paho-mqtt>=1.6.0 diff --git a/plugins/pomodoro-timer/test/golden/128x32/pomodoro.png b/plugins/pomodoro-timer/test/golden/128x32/pomodoro.png new file mode 100644 index 0000000..b88b181 Binary files /dev/null and b/plugins/pomodoro-timer/test/golden/128x32/pomodoro.png differ diff --git a/plugins/pomodoro-timer/test/golden/128x64/pomodoro.png b/plugins/pomodoro-timer/test/golden/128x64/pomodoro.png new file mode 100644 index 0000000..3920f31 Binary files /dev/null and b/plugins/pomodoro-timer/test/golden/128x64/pomodoro.png differ diff --git a/plugins/pomodoro-timer/test/golden/256x32/pomodoro.png b/plugins/pomodoro-timer/test/golden/256x32/pomodoro.png new file mode 100644 index 0000000..4207c3f Binary files /dev/null and b/plugins/pomodoro-timer/test/golden/256x32/pomodoro.png differ diff --git a/plugins/pomodoro-timer/test/golden/64x32/pomodoro.png b/plugins/pomodoro-timer/test/golden/64x32/pomodoro.png new file mode 100644 index 0000000..98d8bf8 Binary files /dev/null and b/plugins/pomodoro-timer/test/golden/64x32/pomodoro.png differ diff --git a/plugins/pomodoro-timer/test/harness.json b/plugins/pomodoro-timer/test/harness.json new file mode 100644 index 0000000..f4d5ae5 --- /dev/null +++ b/plugins/pomodoro-timer/test/harness.json @@ -0,0 +1,31 @@ +{ + "_comment": "Deterministic settings the plugin safety harness renders pomodoro-timer with for golden-image comparison. The timer is idle on a fresh instance, so the golden shows the resting screen: the idle label above the configured work length in seven-segment digits, a full burndown ring, and an empty set of session dots. Everything that affects the frame is pinned here so a future change to a schema default can't silently invalidate the goldens.", + "config": { + "work_minutes": 25, + "short_break_minutes": 5, + "long_break_minutes": 15, + "sessions_before_long_break": 4, + "mqtt_enabled": false, + "auto_start_on_enable": false, + "color_mode": "phase", + "idle_color": [ + 110, + 110, + 110 + ], + "background_color": [ + 0, + 0, + 0 + ], + "idle_label": "POMODORO", + "show_phase_label": true, + "show_session_dots": true, + "digit_style": "seven_segment", + "show_ghost_segments": false, + "progress_style": "perimeter", + "font_path": "", + "font_size": 0 + }, + "mock_data": {} +} diff --git a/plugins/pomodoro-timer/test/render_readme_assets.py b/plugins/pomodoro-timer/test/render_readme_assets.py new file mode 100644 index 0000000..58683ff --- /dev/null +++ b/plugins/pomodoro-timer/test/render_readme_assets.py @@ -0,0 +1,223 @@ +"""Regenerate the images in ../assets/ that the README uses. + +Every panel in those sheets is real plugin output: rendered through the core's +bounds-checking display manager at the true panel size, then scaled up with +nearest-neighbour so the pixels stay pixels. Nothing is a mockup, and nothing is +touched up afterwards — so if the display changes, rerun this rather than +editing the PNGs. + +Needs a LEDMatrix (core) checkout for the harness: + + python plugins/pomodoro-timer/test/render_readme_assets.py --core /path/to/LEDMatrix + +The core path also comes from $LEDMATRIX_CORE, and defaults to a sibling +checkout of this repo. +""" +import argparse +import os +import sys +from pathlib import Path + +PDIR = Path(__file__).resolve().parent.parent +OUT = PDIR / "assets" + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument("--core", default=os.environ.get("LEDMATRIX_CORE"), + help="path to a LEDMatrix (core) checkout") +args = parser.parse_args() +core = Path(args.core) if args.core else PDIR.parents[1].parent / "LEDMatrix" +if not (core / "src" / "plugin_system").is_dir(): + sys.exit(f"no LEDMatrix core checkout at {core} — pass --core") + +sys.path.insert(0, str(core)) +os.environ["EMULATOR"] = "true" + +from PIL import Image, ImageDraw, ImageFont # noqa: E402 +from src.plugin_system.testing.loading import load_manifest # noqa: E402 +from src.plugin_system.testing.harness import ( # noqa: E402 + _instantiate, load_config_defaults) +from src.plugin_system.testing.bounds_display_manager import ( # noqa: E402 + BoundsCheckingDisplayManager) + +OUT.mkdir(parents=True, exist_ok=True) +MANIFEST = load_manifest(PDIR) +BASE = {"enabled": True, **load_config_defaults(PDIR)} +# The pip for the session you are in blinks once a second; hold it lit so the +# sheets come out the same every run. +BASE["pulse_active_pip"] = False + +FONT_DIRS = ["/usr/share/fonts/truetype/dejavu", "/Library/Fonts", + "/usr/share/fonts/TTF"] + + +def _font(name: str, size: int): + for directory in FONT_DIRS: + path = Path(directory) / name + if path.exists(): + return ImageFont.truetype(str(path), size) + return ImageFont.load_default() + + +F_NAME = _font("DejaVuSans-Bold.ttf", 15) +F_CAPTION = _font("DejaVuSans.ttf", 13) + +CARD = (18, 18, 22) +FG = (228, 228, 234) +DIM = (146, 146, 158) +EDGE = (58, 58, 68) +PAD = 24 + + +def render(width: int, height: int, setup=None, burn: float = 0.4, **overrides): + """One frame of the plugin at a real panel size, with the phase `burn` of + the way through so the burndown indicator has something to show.""" + display = BoundsCheckingDisplayManager(width=width, height=height, + overflow_extent=(width, height)) + plugin = _instantiate("pomodoro-timer", MANIFEST, PDIR, + {**BASE, **overrides}, {}, display) + if setup: + setup(plugin) + with plugin.state_lock: + if plugin.deadline is not None: + plugin.deadline -= plugin.phase_total * burn + else: + plugin.remaining = max(1.0, plugin.remaining - plugin.phase_total * burn) + plugin.display() + assert display.check_overflow() is None, f"overflow at {width}x{height}" + return display.get_image() + + +def scaled(image, factor: int): + return image.resize((image.width * factor, image.height * factor), Image.NEAREST) + + +def paste_panel(sheet, image, x: int, y: int) -> None: + """Drop a panel onto the card inside a hairline edge — the panel background + is black and the card is nearly black, so it needs the outline to read.""" + ImageDraw.Draw(sheet).rectangle( + [x - 1, y - 1, x + image.width, y + image.height], outline=EDGE) + sheet.paste(image, (x, y)) + + +def save(sheet, path: Path) -> None: + """Write the sheet, dropping to a palette when the whole card fits in 256 + colours — these are flat-colour images, so that is lossless and roughly + halves the file.""" + colours = sheet.getcolors(1 << 24) + if colours and len(colours) <= 256: + sheet = sheet.quantize(colors=len(colours), method=Image.MEDIANCUT) + sheet.save(path, optimize=True) + print(path.name, f"{path.stat().st_size // 1024}K") + + +def caption(draw, x: int, y: int, name: str, text: str, width: int) -> int: + """Name with its one-line explanation under it, both asserted to fit inside + the column so a longer caption can't silently run into the next one.""" + for font, string, colour, dy in ((F_NAME, name, FG, 0), + (F_CAPTION, text, DIM, 20)): + assert draw.textlength(string, font=font) <= width, f"{string!r} overflows" + draw.text((x, y + dy), string, font=font, fill=colour) + return 42 + + +# ── The states worth showing ─────────────────────────────────────────────────── + +def command(cmd: str, payload=None): + return lambda plugin: plugin._apply_command(cmd, payload or {}) + + +def paused(plugin) -> None: + plugin._apply_command("START", {}) + plugin._apply_command("PAUSE", {}) + + +def long_break(plugin) -> None: + with plugin.state_lock: + plugin.completed_sessions, plugin.cycle_position = 3, 3 + plugin._apply_command("LONG_BREAK", {}) + + +def working_on(task: str, position: int = 2): + def setup(plugin) -> None: + plugin._apply_command("START", {}) + with plugin.state_lock: + plugin.task = task + plugin.completed_sessions = plugin.cycle_position = position + return setup + + +WORK = working_on("WRITE SPEC") + + +def grid(path: Path, cells, scale: int = 3, panel=(128, 32), columns: int = 2) -> None: + """A card of captioned panels, laid out left to right, top to bottom.""" + pw, ph = panel[0] * scale, panel[1] * scale + gap_x, cell_h = 28, ph + 42 + 30 + rows = -(-len(cells) // columns) + sheet = Image.new("RGB", (PAD * 2 + pw * columns + gap_x * (columns - 1), + PAD * 2 + rows * cell_h - 30), CARD) + draw = ImageDraw.Draw(sheet) + for index, (name, text, setup, burn, extra) in enumerate(cells): + x = PAD + (index % columns) * (pw + gap_x) + y = PAD + (index // columns) * cell_h + y += caption(draw, x, y, name, text, pw) + paste_panel(sheet, scaled(render(*panel, setup, burn, **extra), scale), x, y) + save(sheet, path) + + +def main() -> None: + # The hero: a work session on a 128x64 panel. + image = scaled(render(128, 64, WORK), 6) + sheet = Image.new("RGB", (image.width + PAD * 2, image.height + PAD * 2), CARD) + paste_panel(sheet, image, PAD, PAD) + save(sheet, OUT / "hero.png") + + grid(OUT / "phases.png", [ + ("Idle", "nothing running — the next work length, in full", None, 0.0, {}), + ("Work", "a focus session, part-way through", command("START"), 0.4, {}), + ("Paused", "held where it was; the ring stops draining", paused, 0.4, {}), + ("Short break", "the breather between sessions", + command("SHORT_BREAK"), 0.4, {}), + ("Long break", "three sessions banked, one to go", long_break, 0.4, {}), + ("With a task", "a task name takes the label row", WORK, 0.4, {}), + ]) + + grid(OUT / "options.png", [ + ("Burndown: perimeter", "the default — a ring that drains clockwise", + WORK, 0.4, {"progress_style": "perimeter"}), + ("Burndown: bar", "an emptying bar along the bottom edge", + WORK, 0.4, {"progress_style": "bar"}), + ("Burndown: segments", "blocks that go out one at a time", + WORK, 0.4, {"progress_style": "segments"}), + ("Burndown: none", "no indicator at all", + WORK, 0.4, {"progress_style": "none"}), + ("Unlit segments on", "the clock-radio look, at some cost to legibility", + WORK, 0.4, {"show_ghost_segments": True}), + ("Pixel digits", "the display font instead of drawn segments", + WORK, 0.4, {"digit_style": "pixel"}), + ("Calm theme", "warm terracotta, sage and soft indigo", + WORK, 0.4, {"color_theme": "calm"}), + ("Paused: desaturate", "the phase's own hue drained, instead of amber", + paused, 0.4, {"paused_style": "desaturate"}), + ]) + + # One work session on every panel size, all at the same scale, so the sheet + # shows how much room each shape really has. + scale = 3 + tiles = [(f"{w}×{h}", scaled(render(w, h, WORK), scale)) + for w, h in ((64, 32), (128, 32), (128, 64), (256, 32))] + sheet = Image.new( + "RGB", (PAD * 2 + max(i.width for _, i in tiles), + PAD * 2 + sum(i.height + 40 for _, i in tiles) - 18), CARD) + draw = ImageDraw.Draw(sheet) + y = PAD + for label, image in tiles: + draw.text((PAD, y), label, font=F_NAME, fill=DIM) + y += 22 + paste_panel(sheet, image, PAD, y) + y += image.height + 18 + save(sheet, OUT / "sizes.png") + + +if __name__ == "__main__": + main() diff --git a/plugins/pomodoro-timer/test_pomodoro_timer.py b/plugins/pomodoro-timer/test_pomodoro_timer.py new file mode 100644 index 0000000..4a73ca8 --- /dev/null +++ b/plugins/pomodoro-timer/test_pomodoro_timer.py @@ -0,0 +1,773 @@ +#!/usr/bin/env python3 +""" +Pomodoro Timer plugin tests. + +Self-contained: the LEDMatrix core is stubbed out, so this runs from a plain +checkout of the plugins repo with only Pillow installed. + + python -m pytest plugins/pomodoro-timer/test_pomodoro_timer.py + python plugins/pomodoro-timer/test_pomodoro_timer.py # no pytest needed +""" + +import importlib.util +import json +import logging +import sys +import types +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +PLUGIN_DIR = Path(__file__).resolve().parent + + +# ── Core stubs ────────────────────────────────────────────────────────────── + +class _StubBasePlugin: + def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager): + self.plugin_id = plugin_id + self.config = config + self.display_manager = display_manager + self.cache_manager = cache_manager + self.plugin_manager = plugin_manager + self.enabled = True + self.global_config = {} + self.logger = logging.getLogger(plugin_id) + + def validate_config(self): + return True + + def get_info(self): + return {"plugin_id": self.plugin_id} + + def on_enable(self): + self.enabled = True + + def on_disable(self): + self.enabled = False + + def on_config_change(self, new_config): + self.config = new_config + + +def _install_core_stub(): + for name in ("src", "src.plugin_system", "src.plugin_system.base_plugin"): + sys.modules.pop(name, None) + src = types.ModuleType("src") + plugin_system = types.ModuleType("src.plugin_system") + base_plugin = types.ModuleType("src.plugin_system.base_plugin") + base_plugin.BasePlugin = _StubBasePlugin + plugin_system.base_plugin = base_plugin + src.plugin_system = plugin_system + sys.modules["src"] = src + sys.modules["src.plugin_system"] = plugin_system + sys.modules["src.plugin_system.base_plugin"] = base_plugin + + +class _Matrix: + def __init__(self, width, height): + self.width = width + self.height = height + + +class _DisplayManager: + def __init__(self, width=128, height=32): + self.width = width + self.height = height + self.matrix = _Matrix(width, height) + self.image = Image.new("RGB", (width, height), (0, 0, 0)) + self.draw = ImageDraw.Draw(self.image) + self.small_font = ImageFont.load_default() + self.regular_font = self.small_font + self.extra_small_font = self.small_font + self.updates = 0 + + def update_display(self): + self.updates += 1 + + def draw_text(self, text, x=None, y=None, color=(255, 255, 255), **kwargs): + self.draw.text((x or 0, y or 0), text, fill=color, font=self.small_font) + + +class _CacheManager: + def __init__(self): + self.store = {} + + def get(self, key, max_age=None): + return self.store.get(key) + + def set(self, key, value, ttl=None): + self.store[key] = value + + def delete(self, key): + self.store.pop(key, None) + + +class _PluginManager: + font_manager = None + config_manager = None + + +_MISSING = object() +_STUBBED_MODULES = ("src", "src.plugin_system", "src.plugin_system.base_plugin") + + +def _load_module(): + """Import manager.py against the stubbed core, then put sys.modules back. + + `manager.py` binds BasePlugin at class-creation time, so the stub only has + to be in place for the exec. Restoring afterwards keeps our fake `src` from + shadowing the real core for any other plugin's tests in the same session. + """ + originals = {name: sys.modules.get(name, _MISSING) for name in _STUBBED_MODULES} + try: + _install_core_stub() + spec = importlib.util.spec_from_file_location( + "pomodoro_timer_manager", PLUGIN_DIR / "manager.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + finally: + for name, original in originals.items(): + if original is _MISSING: + sys.modules.pop(name, None) + else: + sys.modules[name] = original + + +MODULE = _load_module() +DEFAULTS = { + k: v.get("default") + for k, v in json.loads((PLUGIN_DIR / "config_schema.json").read_text())["properties"].items() + if "default" in v +} + + +def make_plugin(width=128, height=32, **overrides): + config = {**DEFAULTS, "mqtt_enabled": False, **overrides} + return MODULE.PomodoroTimerPlugin( + "pomodoro-timer", config, _DisplayManager(width, height), + _CacheManager(), _PluginManager()) + + +# ── Tests ─────────────────────────────────────────────────────────────────── + +def test_defaults_start_idle(): + p = make_plugin() + snap = p._snapshot() + assert snap["phase"] == "idle" + assert snap["status"] == "idle" + assert snap["remaining"] == "25:00" + assert snap["elapsed_fraction"] == 0.0 + + +def test_clock_formatting(): + assert MODULE._fmt_clock(0) == "00:00" + assert MODULE._fmt_clock(59.2) == "01:00" + assert MODULE._fmt_clock(1500) == "25:00" + assert MODULE._fmt_clock(-5) == "00:00" + + +def test_start_pause_resume(): + p = make_plugin() + p._apply_command("START", {}) + assert p._snapshot()["status"] == "running" + assert p._snapshot()["phase"] == "work" + + p._apply_command("PAUSE", {}) + snap = p._snapshot() + assert snap["status"] == "paused" + assert snap["label"] == "PAUSED" + + p._apply_command("RESUME", {}) + assert p._snapshot()["status"] == "running" + + p._apply_command("TOGGLE", {}) + assert p._snapshot()["status"] == "paused" + + +def test_stop_keeps_counters_reset_clears_them(): + p = make_plugin() + p._apply_command("START", {}) + with p.state_lock: + p.deadline = 0.0 # force the phase to be due + p._tick() # work completes -> break + assert p._snapshot()["completed_sessions"] == 1 + + p._apply_command("STOP", {}) + assert p._snapshot()["phase"] == "idle" + assert p._snapshot()["completed_sessions"] == 1 + + p._apply_command("RESET", {}) + assert p._snapshot()["completed_sessions"] == 0 + assert p._snapshot()["cycle_position"] == 0 + + +def test_full_cycle_reaches_long_break_and_wraps(): + p = make_plugin(auto_start_breaks=True, auto_start_work=True) + p._apply_command("START", {}) + seen = [] + for _ in range(8): + with p.state_lock: + p.deadline = 0.0 + p._tick() + seen.append(p._snapshot()["phase"]) + assert seen == ["short_break", "work", "short_break", "work", + "short_break", "work", "long_break", "work"] + assert p._snapshot()["completed_sessions"] == 4 + # The long break closed the set, so the dots start over. + assert p._snapshot()["cycle_position"] == 0 + + +def test_break_waits_when_auto_start_is_off(): + p = make_plugin(auto_start_breaks=False) + p._apply_command("START", {}) + with p.state_lock: + p.deadline = 0.0 + p._tick() + snap = p._snapshot() + assert snap["phase"] == "short_break" + assert snap["status"] == "paused" + assert snap["remaining"] == "05:00" + + +def test_skipped_work_does_not_earn_a_long_break(): + p = make_plugin() + with p.state_lock: + p.cycle_position = 3 # one work session away from the long break + p._apply_command("WORK", {}) + p._apply_command("SKIP", {}) + snap = p._snapshot() + assert snap["phase"] == "short_break" + assert snap["completed_sessions"] == 0 + assert snap["cycle_position"] == 3 + + +def test_json_command_with_custom_duration_and_label(): + p = make_plugin() + command, payload = p._parse_command( + b'{"command": "start", "phase": "work", "duration_minutes": 50, "label": "DEEP"}') + p._apply_command(command, payload) + snap = p._snapshot() + assert snap["phase"] == "work" + assert snap["remaining"] == "50:00" + assert snap["label"] == "DEEP" + # The custom duration is per-session; the configured length is untouched. + assert p.work_minutes == 25 + + +def test_settings_only_payload_updates_durations(): + p = make_plugin() + command, payload = p._parse_command(b'{"work_minutes": 45, "short_break_minutes": 10}') + assert command == "SET" + p._apply_command(command, payload) + assert p.work_minutes == 45 + assert p.short_break_minutes == 10 + # Idle screen follows the new work length. + assert p._snapshot()["remaining"] == "45:00" + + +def test_number_values_are_clamped(): + p = make_plugin() + with p.state_lock: + p._set_number_locked("work_minutes", "9999") + p._set_number_locked("sessions_before_long_break", 0) + assert p.work_minutes == 180 + assert p.sessions_before_long_break == 1 + + +def test_unrecognised_payloads_are_ignored(): + p = make_plugin() + for raw in (b"", b"gibberish-command", b'{"nothing": 1}', b"\xff\xfe"): + command, payload = p._parse_command(raw) + if command is not None: + p._apply_command(command, payload) + assert p._snapshot()["phase"] == "idle" + + +def test_pin_request_follows_the_timer(): + p = make_plugin(pin_while_running=True) + p._apply_command("START", {}) + request = p.cache_manager.get("display_on_demand_request") + assert request["action"] == "start" and request["pinned"] is True + + p._apply_command("STOP", {}) + assert p.cache_manager.get("display_on_demand_request")["action"] == "stop" + + +def test_pin_is_not_requested_when_disabled(): + p = make_plugin(pin_while_running=False, alert_seconds=0) + p._apply_command("START", {}) + assert p.cache_manager.get("display_on_demand_request") is None + + +def test_display_renders_every_panel_size(): + for width, height in ((64, 32), (128, 32), (128, 64), (256, 32)): + p = make_plugin(width, height) + p._apply_command("START", {}) + assert p.display() is True + assert p.display_manager.updates >= 1 + assert p.display_manager.image.size == (width, height) + assert p.display_manager.image.getbbox() is not None, "rendered a blank frame" + + +def _ring_path(width, height): + return ([(x, 0) for x in range(width)] + + [(width - 1, y) for y in range(1, height)] + + [(x, height - 1) for x in range(width - 2, -1, -1)] + + [(0, y) for y in range(height - 2, 0, -1)]) + + +def _lit_ring_fraction(plugin): + width, height = plugin.display_manager.image.size + pixels = plugin.display_manager.image.convert("RGB").load() + path = _ring_path(width, height) + lit = sum(1 for point in path if sum(pixels[point]) > 200) + return lit / len(path) + + +def test_burndown_ring_drains_with_the_phase(): + for elapsed in (0.0, 0.25, 0.5, 0.75): + p = make_plugin(128, 32, progress_style="perimeter") + p._apply_command("START", {}) + with p.state_lock: + p.deadline -= p.phase_total * elapsed + p.display() + got = _lit_ring_fraction(p) + assert abs(got - (1 - elapsed)) < 0.02, \ + f"{elapsed:.0%} elapsed should leave {1 - elapsed:.0%} of the ring lit, got {got:.0%}" + + +def test_seven_segment_digits_keep_a_readable_shape(): + """A digit that gets too narrow for its height stops reading as a digit.""" + for width, height in ((64, 32), (128, 32), (128, 64), (256, 32)): + p = make_plugin(width, height) + for text in ("25:00", "05:00", "180:00"): + metrics = p._seven_segment_metrics(text, width - 6, height - 10) + if metrics is None: + continue + h, digit_w, stroke, _gap, _colon_w, total = metrics + assert digit_w >= round(h * 0.64), f"{width}x{height} {text}: digit too narrow" + assert digit_w >= 2 * stroke + 1, f"{width}x{height} {text}: segments would fuse" + assert total <= width - 6, f"{width}x{height} {text}: digits overrun the box" + + +def test_seven_segment_gives_up_when_there_is_no_room(): + p = make_plugin() + assert p._seven_segment_metrics("25:00", 10, 4) is None + # ...and display() still renders, having fallen back to the pixel font. + tiny = make_plugin(64, 16) + tiny._apply_command("START", {}) + assert tiny.display() is True + assert tiny.display_manager.image.getbbox() is not None + + +def test_every_indicator_and_digit_style_renders(): + for style in ("perimeter", "bar", "segments", "none"): + for digits in ("seven_segment", "pixel"): + p = make_plugin(128, 32, progress_style=style, digit_style=digits) + p._apply_command("START", {}) + assert p.display() is True, f"{style}/{digits} failed to render" + assert p.display_manager.image.getbbox() is not None + + +def test_a_task_replaces_the_phase_label_and_survives_phases(): + p = make_plugin() + client = _wire_client(p) + p._apply_command("START", {}) + assert p._snapshot()["label"] == "FOCUS" + + p._on_mqtt_message(client, None, _Message(p.task_set_topic, "Editing specs")) + assert p._snapshot()["task"] == "Editing specs" + assert p._snapshot()["label"] == "Editing specs" + assert dict(client.published)[p.task_topic] == "Editing specs" + + # It is what you are working on, not what phase you are in, so it carries + # across a phase change... + p._apply_command("SKIP", {}) + assert p._snapshot()["label"] == "Editing specs" + # ...but a per-session label sent with a start command still wins. + p._apply_command("START", {"label": "DEEP WORK"}) + assert p._snapshot()["label"] == "DEEP WORK" + # ...and RESET clears it. + p._apply_command("RESET", {}) + assert p._snapshot()["task"] == "" + assert p._snapshot()["label"] == "POMODORO" + + +def test_task_text_is_bounded(): + p = make_plugin() + client = _wire_client(p) + p._on_mqtt_message(client, None, _Message(p.task_set_topic, "x" * 200)) + assert len(p._snapshot()["task"]) == 32 + + +def test_a_label_sent_with_start_applies_however_the_timer_was_left(): + """START carries a label whichever branch it lands in. + + Idle, running and paused each restart differently, and a label the caller + explicitly asked for shouldn't depend on which one it hit. + """ + for name, setup in (("idle", []), + ("running", [("START", {})]), + ("paused", [("START", {}), ("PAUSE", {})])): + p = make_plugin() + for command, payload in setup: + p._apply_command(command, payload) + p._apply_command("START", {"label": "DEEP WORK"}) + assert p._snapshot()["label"] == "DEEP WORK", f"dropped when {name}" + # Resuming still resumes rather than restarting the phase. + p = make_plugin() + p._apply_command("START", {}) + with p.state_lock: + p.deadline -= 600 + p._apply_command("PAUSE", {}) + p._apply_command("START", {"label": "DEEP WORK"}) + assert p._snapshot()["remaining"] == "15:00" + + +def test_command_labels_are_bounded_like_the_task(): + p = make_plugin() + p._apply_command("START", {"label": "x" * 200}) + assert len(p._snapshot()["label"]) == 32 + # Anything that isn't a string still has to survive the render path. + p._apply_command("WORK", {"label": 12345}) + assert p._snapshot()["label"] == "12345" + assert p.display() is True + + +def test_calm_theme_and_desaturated_pause(): + classic = make_plugin() + calm = make_plugin(color_theme="calm") + assert calm.work_color != classic.work_color + for name in ("work_color", "short_break_color", "long_break_color"): + assert getattr(calm, name) != getattr(classic, name) + + p = make_plugin(paused_style="desaturate") + p._apply_command("START", {}) + p._apply_command("PAUSE", {}) + assert p.display() is True + assert p.display_manager.image.getbbox() is not None + + +def _panel_colours(plugin): + return {c for _, c in plugin.display_manager.image.convert("RGB").getcolors(65536)} + + +def test_nothing_is_antialiased(): + """Every pixel is one of the palette colours — no half-lit fringe. + + PIL antialiases TrueType text by default, which on an LED panel is a grey + halo around every glyph. The label face is a pixel font at integer sizes, + so any intermediate value is an artefact rather than a design choice. + """ + def dim(c, f): + return tuple(max(0, min(255, int(x * f))) for x in c) + + for width, height in ((64, 32), (128, 32), (128, 64), (256, 32)): + p = make_plugin(width, height) + p._apply_command("START", {}) + p.display() + accent = p.work_color + allowed = { + (0, 0, 0), accent, + dim(accent, 0.7), # label + dim(accent, 0.14), dim(accent, 0.28), # ring track, upcoming pip + tuple(max(0, min(255, int(c + (255 - c) * 0.55))) for c in accent), + } + stray = _panel_colours(p) - allowed + assert not stray, f"{width}x{height} has unintended colours: {sorted(stray)}" + + +def test_no_overflow_on_any_panel_shape(): + """A panel can be almost any size, so sweep well past the sampled ones. + + The label is dropped rather than squeezed once there is no room for both + it and the countdown; flooring the countdown box instead pushed the digits + straight off the bottom edge on anything under ~16px tall. + """ + for width in (32, 48, 64, 96, 128, 192, 256, 384): + for height in (8, 10, 12, 14, 16, 20, 32, 64, 128): + p = make_plugin(width, height) + p._apply_command("START", {}) + assert p.display() is True, f"{width}x{height} failed to render" + image = p.display_manager.image + assert image.size == (width, height) + assert image.getbbox() is not None, f"{width}x{height} rendered blank" + + +def test_unknown_styles_fall_back_to_the_defaults(): + p = make_plugin(progress_style="disco", digit_style="gothic") + assert p.progress_style == "perimeter" + assert p.digit_style == "seven_segment" + + +def test_rendering_never_rewrites_the_configured_style(): + """A panel too narrow for blocks falls back per-frame, not permanently.""" + p = make_plugin(24, 16, progress_style="segments") + p._apply_command("START", {}) + p.display() + assert p.progress_style == "segments", "the render path overwrote the user's setting" + + +def test_changing_the_discovery_prefix_clears_the_old_configs(): + p = make_plugin(command_topic="ledmatrix/pomodoro/set") + client = _wire_client(p) + p.on_config_change({**DEFAULTS, "mqtt_enabled": False, "discovery_prefix": "ha2"}) + cleared = [t for t, payload in client.published + if t.startswith("homeassistant/") and payload == ""] + assert len(cleared) == 18, "old prefix's retained configs would orphan HA entities" + + +def test_renaming_the_command_topic_clears_the_old_state_topics(): + p = make_plugin(command_topic="ledmatrix/pomodoro/set") + client = _wire_client(p) + old_state, old_avail = p.state_topic, p.availability_topic + p.on_config_change({**DEFAULTS, "mqtt_enabled": False, + "command_topic": "desk/focus/set"}) + cleared = {t for t, payload in client.published if payload == ""} + assert old_state in cleared and old_avail in cleared + # The discovery configs are keyed on the prefix, so they are republished in + # place rather than orphaned — nothing under homeassistant/ gets cleared. + assert not any(t.startswith("homeassistant/") for t in cleared) + assert p.topic_base == "desk/focus" + + +def test_renaming_only_the_state_topic_clears_what_it_left_behind(): + """`state_topic` is settable on its own, not just derived from the base. + + Changing it alone would otherwise leave the old topic retained on the + broker forever, showing a timer that stopped existing. + """ + p = make_plugin() + client = _wire_client(p) + old_state = p.state_topic + p.on_config_change({**DEFAULTS, "mqtt_enabled": False, + "state_topic": "desk/focus/running"}) + assert old_state in {t for t, payload in client.published if payload == ""} + assert p.state_topic == "desk/focus/running" + + +def test_discovery_payloads_are_valid_json_and_unique(): + p = make_plugin() + entities = p._discovery_entities() + assert len(entities) == 18 + uids = set() + for topic, payload in entities: + assert topic.count("/") == 2 and topic.endswith("/config") + json.dumps(payload) + assert payload["unique_id"] not in uids + uids.add(payload["unique_id"]) + topics = [t for t, _ in entities] + assert any(t.startswith("switch/") for t in topics) + assert sum(t.startswith("button/") for t in topics) == 5 + assert sum(t.startswith("number/") for t in topics) == 4 + assert any(t.startswith("text/") for t in topics) + + +class _RecordingClient: + """Stands in for paho's client so the MQTT plumbing can be exercised + without a broker (an end-to-end run against mosquitto is covered + separately).""" + + def __init__(self): + self.subscriptions = [] + self.published = [] + + def subscribe(self, topic, qos=0): + self.subscriptions.append(topic) + + def publish(self, topic, payload, retain=False, qos=0): + self.published.append((topic, payload)) + + +class _Message: + def __init__(self, topic, payload): + self.topic = topic + self.payload = payload if isinstance(payload, bytes) else payload.encode() + + +def _wire_client(plugin): + client = _RecordingClient() + plugin.mqtt_client = client + plugin.mqtt_connected = True + return client + + +def test_connect_subscribes_to_command_and_number_topics(): + p = make_plugin() + client = _RecordingClient() + p.mqtt_client = client + p._on_mqtt_connect(client, None, {}, 0) + assert p.mqtt_connected is True + assert p.command_topic in client.subscriptions + for topic in p.number_set_topics: + assert topic in client.subscriptions + published = {t for t, _ in client.published} + assert any(t.startswith("homeassistant/") for t in published) + assert p.state_topic in published + + +def test_incoming_messages_drive_the_timer(): + p = make_plugin() + client = _wire_client(p) + + p._on_mqtt_message(client, None, _Message(p.command_topic, "START")) + assert p._snapshot()["status"] == "running" + + p._on_mqtt_message(client, None, _Message(p.command_topic, "PAUSE")) + assert p._snapshot()["status"] == "paused" + + p._on_mqtt_message(client, None, _Message(f"{p.topic_base}/work_minutes/set", "45")) + assert p.work_minutes == 45 + + published = dict(client.published) + assert published[p.state_topic] == "ON" + assert published[p.status_topic] == "paused" + assert published[p.number_state_topics["work_minutes"]] == "45" + + +def test_a_failed_connection_does_not_look_connected(): + p = make_plugin() + client = _RecordingClient() + p.mqtt_client = client + p._on_mqtt_connect(client, None, {}, 5) # 5 = not authorised + assert p.mqtt_connected is False + assert client.subscriptions == [] + + +def test_start_on_a_running_timer_emits_no_event(): + p = make_plugin() + client = _wire_client(p) + p._apply_command("START", {}) + before = [t for t, _ in client.published if t == p.event_topic] + + p._apply_command("START", {}) # e.g. the HA Start button pressed again + after = [t for t, _ in client.published if t == p.event_topic] + assert after == before, "a redundant START re-fired the started event" + + # Settings sent alongside a redundant START still apply and get published. + p._apply_command("START", {"work_minutes": 30}) + assert p.work_minutes == 30 + assert dict(client.published)[p.number_state_topics["work_minutes"]] == "30" + + # A resume, by contrast, is a real transition and does emit. + p._apply_command("PAUSE", {}) + p._apply_command("START", {}) + assert p._snapshot()["status"] == "running" + assert len([t for t, _ in client.published if t == p.event_topic]) > len(after) + + +class _StubPahoClient: + """Records the teardown calls made against a live client.""" + + def __init__(self): + self.loop_stopped = 0 + self.disconnected = 0 + self.published = [] + + def loop_stop(self): + self.loop_stopped += 1 + + def disconnect(self): + self.disconnected += 1 + + def publish(self, topic, payload, retain=False, qos=0): + self.published.append((topic, payload)) + + +def test_teardown_releases_the_client_and_is_idempotent(): + p = make_plugin() + client = _StubPahoClient() + p.mqtt_client = client + p.mqtt_connected = True + p.mqtt_connecting = True + + p._teardown_client() + assert p.mqtt_client is None + assert p.mqtt_connected is False and p.mqtt_connecting is False + assert client.loop_stopped == 1 and client.disconnected == 1 + + p._teardown_client() # no client left — must not raise or re-call + assert client.loop_stopped == 1 + + +def test_teardown_survives_a_client_that_raises(): + p = make_plugin() + + class _Angry(_StubPahoClient): + def loop_stop(self): + raise RuntimeError("loop already stopped") + + def disconnect(self): + raise RuntimeError("socket gone") + + p.mqtt_client = _Angry() + p._teardown_client() # must swallow both failures + assert p.mqtt_client is None + + +def test_shutdown_releases_a_client_left_by_a_dead_thread(): + p = make_plugin() + client = _StubPahoClient() + p.mqtt_client = client # loop exited (or never ran) with a live client + p.mqtt_thread = None + + p._graceful_shutdown() + assert p.mqtt_stop_event.is_set() + assert p.mqtt_client is None + assert client.disconnected >= 1 + # HA must be told we're gone before the socket closes, or the device sits + # "online" until the keepalive lapses. + assert (p.availability_topic, "offline") in client.published + + +def test_pin_state_is_restored_when_the_request_fails(): + p = make_plugin() + + def _boom(*args, **kwargs): + raise RuntimeError("cache unavailable") + + p.cache_manager.set = _boom + p._apply_command("START", {}) + assert p._pinned is False, "a failed request must stay retryable" + + +def test_derived_topics_track_the_command_topic(): + p = make_plugin(command_topic="home/focus/set") + assert p.topic_base == "home/focus" + assert p.remaining_topic == "home/focus/remaining" + assert "home/focus/work_minutes/set" in p.number_set_topics + assert p.number_state_topics["work_minutes"] == "home/focus/work_minutes" + + +def test_validate_config_rejects_bad_values(): + assert make_plugin().validate_config() is True + assert make_plugin(color_mode="rainbow").validate_config() is False + assert make_plugin(mqtt_port=0).validate_config() is False + + +def test_config_change_keeps_a_running_session(): + p = make_plugin() + p._apply_command("START", {}) + before = p._snapshot()["remaining_seconds"] + p.on_config_change({**DEFAULTS, "mqtt_enabled": False, "work_color": [1, 2, 3]}) + snap = p._snapshot() + assert snap["phase"] == "work" + assert abs(snap["remaining_seconds"] - before) <= 1 + assert p.work_color == (1, 2, 3) + + +if __name__ == "__main__": + failures = 0 + for name, fn in sorted(globals().items()): + if not name.startswith("test_") or not callable(fn): + continue + try: + fn() + print(f"[ok] {name}") + except Exception as exc: # noqa: BLE001 - report and keep going + failures += 1 + print(f"[FAIL] {name}: {exc!r}") + print(f"\n{'FAILED' if failures else 'PASSED'} — {failures} failure(s)") + sys.exit(1 if failures else 0)