tgrosinger
194029d713
Breaks started from the menu in reading mode used to end on their own once the due break's length passed, so work started counting before the user was back. They now last until input, which makes them identical to an ordinary idle_start, so the reading_break event and its cap logic are gone.
373 lines
12 KiB
Python
Executable File
373 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
break-status - Calculate and display break timer status for Waybar
|
|
|
|
Parses the activity log to determine time until next breaks.
|
|
Outputs JSON format compatible with Waybar custom modules.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
from pathlib import Path
|
|
from typing import TypedDict, NotRequired
|
|
|
|
|
|
Event = tuple[int, str]
|
|
Events = list[Event]
|
|
|
|
class BreakConfig(TypedDict):
|
|
interval: str
|
|
duration: str
|
|
|
|
|
|
class Config(TypedDict):
|
|
short_break: BreakConfig
|
|
long_break: BreakConfig
|
|
snooze: NotRequired[str]
|
|
|
|
class Status(TypedDict):
|
|
state: str
|
|
short_remaining: int
|
|
long_remaining: int
|
|
break_remaining: NotRequired[int]
|
|
snooze_remaining: NotRequired[int]
|
|
|
|
|
|
def parse_duration(duration: str) -> int:
|
|
"""Parse duration string like '5m' or '30s' to seconds"""
|
|
if isinstance(duration, (int, float)):
|
|
return int(duration)
|
|
|
|
duration = duration.strip()
|
|
if duration.endswith('m'):
|
|
return int(duration[:-1]) * 60
|
|
elif duration.endswith('s'):
|
|
return int(duration[:-1])
|
|
elif duration.endswith('h'):
|
|
return int(duration[:-1]) * 3600
|
|
else:
|
|
# Assume seconds if no unit
|
|
return int(duration)
|
|
|
|
|
|
def load_config() -> Config:
|
|
"""Load configuration from ~/.config/break-timer/config.json"""
|
|
config_path = Path.home() / ".config" / "break-timer" / "config.json"
|
|
|
|
# Default configuration
|
|
default_config = {
|
|
"short_break": {"interval": "20m", "duration": "20s"},
|
|
"long_break": {"interval": "45m", "duration": "5m"}
|
|
}
|
|
|
|
if not config_path.exists():
|
|
return default_config
|
|
|
|
try:
|
|
with open(config_path) as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
return default_config
|
|
|
|
|
|
def get_log_path() -> Path:
|
|
"""Get the path to the activity log file"""
|
|
runtime_dir = os.environ.get('XDG_RUNTIME_DIR')
|
|
if runtime_dir:
|
|
log_dir = Path(runtime_dir) / "break-timer"
|
|
else:
|
|
log_dir = Path.home() / ".cache" / "break-timer"
|
|
|
|
return log_dir / "activity.log"
|
|
|
|
|
|
def parse_log(log_path: Path) -> Events:
|
|
"""Parse the log file and return list of (timestamp, event_type) tuples"""
|
|
events = []
|
|
if not log_path.exists():
|
|
return events
|
|
|
|
try:
|
|
with open(log_path, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
parts = line.split(maxsplit=1)
|
|
if len(parts) == 2:
|
|
try:
|
|
timestamp = int(parts[0])
|
|
event_type = parts[1]
|
|
events.append((timestamp, event_type))
|
|
except ValueError:
|
|
continue # Skip malformed lines
|
|
except IOError:
|
|
pass
|
|
|
|
return events
|
|
|
|
|
|
def format_time(seconds: int) -> str:
|
|
"""Format seconds as MM:SS"""
|
|
if seconds < 0:
|
|
seconds = 0
|
|
minutes = seconds // 60
|
|
secs = seconds % 60
|
|
return f"{minutes}:{secs:02d}"
|
|
|
|
|
|
def is_paused(events: Events) -> bool:
|
|
"""Check if timer is currently paused (last event is 'paused')"""
|
|
return bool(events and events[-1][1] == "paused")
|
|
|
|
|
|
def calculate_status(events: Events, config: Config) -> Status:
|
|
"""
|
|
Calculate break timer status from events.
|
|
Returns dict with status information.
|
|
"""
|
|
current_time = int(time.time())
|
|
|
|
# Check if currently paused
|
|
if is_paused(events):
|
|
return {
|
|
"state": "paused",
|
|
"short_remaining": 0,
|
|
"long_remaining": 0,
|
|
}
|
|
|
|
# Convert config to seconds
|
|
short_interval = parse_duration(config["short_break"]["interval"])
|
|
short_duration = parse_duration(config["short_break"]["duration"])
|
|
long_interval = parse_duration(config["long_break"]["interval"])
|
|
long_duration = parse_duration(config["long_break"]["duration"])
|
|
snooze_duration = parse_duration(config.get("snooze", "5m"))
|
|
|
|
# If no events, assume just took breaks (timers at 0)
|
|
if not events:
|
|
return {
|
|
"state": "working",
|
|
"short_remaining": short_interval,
|
|
"long_remaining": long_interval,
|
|
}
|
|
|
|
active_time_since_short_break = 0
|
|
active_time_since_long_break = 0
|
|
|
|
pause_events = [e for e in events if e[1] in ("paused", "unpaused")]
|
|
idle_events = [e for e in events if e[1] in ("idle_start", "idle_end", "snoozed")]
|
|
snooze_until = None
|
|
|
|
# Helper to calculate paused time within a range
|
|
def get_paused_time_in_range(start_time: int, end_time: int) -> int:
|
|
"""Calculate total paused time within a time range"""
|
|
paused_time = 0
|
|
pause_start = None
|
|
|
|
for timestamp, event_type in pause_events:
|
|
if timestamp > end_time:
|
|
break
|
|
|
|
if event_type == "paused" and timestamp >= start_time:
|
|
pause_start = timestamp
|
|
elif event_type == "unpaused" and pause_start is not None:
|
|
# Calculate overlap between pause period and our range
|
|
overlap_start = max(pause_start, start_time)
|
|
overlap_end = min(timestamp, end_time)
|
|
if overlap_end > overlap_start:
|
|
paused_time += overlap_end - overlap_start
|
|
pause_start = None
|
|
|
|
# Handle case where still paused at end of range
|
|
if pause_start is not None and pause_start < end_time:
|
|
paused_time += end_time - max(pause_start, start_time)
|
|
return paused_time
|
|
|
|
def add_work(start: int, end: int) -> None:
|
|
nonlocal active_time_since_short_break, active_time_since_long_break
|
|
duration = end - start - get_paused_time_in_range(start, end)
|
|
active_time_since_short_break += duration
|
|
active_time_since_long_break += duration
|
|
|
|
def credit_break(start: int, end: int) -> None:
|
|
nonlocal active_time_since_short_break, active_time_since_long_break, snooze_until
|
|
if end - start >= long_duration:
|
|
active_time_since_long_break = 0
|
|
if end - start >= short_duration:
|
|
active_time_since_short_break = 0
|
|
snooze_until = None # Taking a break uses up the snooze
|
|
|
|
# Replay the log. Exactly one of active_start / idle_start_time is set once
|
|
# the first event is seen.
|
|
active_start = None
|
|
idle_start_time = None
|
|
|
|
for timestamp, event_type in idle_events:
|
|
if event_type == "snoozed":
|
|
# Only counts if a break is due; snoozing again restarts it
|
|
pending = 0
|
|
if active_start is not None:
|
|
pending = timestamp - active_start - get_paused_time_in_range(active_start, timestamp)
|
|
if (active_time_since_short_break + pending >= short_interval
|
|
or active_time_since_long_break + pending >= long_interval):
|
|
snooze_until = timestamp + snooze_duration
|
|
|
|
elif event_type == "idle_end":
|
|
if idle_start_time is not None:
|
|
credit_break(idle_start_time, timestamp)
|
|
idle_start_time = None
|
|
if active_start is None:
|
|
active_start = timestamp
|
|
|
|
elif idle_start_time is None: # idle_start; ignored if already on a break
|
|
if active_start is not None:
|
|
add_work(active_start, timestamp)
|
|
active_start = None
|
|
idle_start_time = timestamp
|
|
|
|
if active_start is not None:
|
|
add_work(active_start, current_time)
|
|
|
|
short_remaining = short_interval - active_time_since_short_break
|
|
long_remaining = long_interval - active_time_since_long_break
|
|
|
|
# Determine current state and calculate remaining times
|
|
if idle_start_time is not None:
|
|
# Currently idle - determine if in break
|
|
idle_elapsed = current_time - idle_start_time
|
|
|
|
if idle_elapsed >= long_duration:
|
|
# Long break completed, show time until next breaks
|
|
return {
|
|
"state": "break_complete",
|
|
"short_remaining": short_interval, # Reset
|
|
"long_remaining": long_interval, # Reset
|
|
}
|
|
elif idle_elapsed >= short_duration:
|
|
# Short break completed, long break in progress
|
|
long_break_remaining = long_duration - idle_elapsed
|
|
return {
|
|
"state": "long_break",
|
|
"break_remaining": long_break_remaining,
|
|
"short_remaining": short_interval, # Already reset
|
|
"long_remaining": long_remaining, # Will reset when break completed
|
|
}
|
|
else:
|
|
# Short break in progress
|
|
short_break_remaining = short_duration - idle_elapsed
|
|
return {
|
|
"state": "short_break",
|
|
"break_remaining": short_break_remaining,
|
|
"short_remaining": short_remaining, # Will reset when break completed
|
|
"long_remaining": long_remaining,
|
|
}
|
|
|
|
# Currently working
|
|
status: Status = {
|
|
"state": "working",
|
|
"short_remaining": short_remaining,
|
|
"long_remaining": long_remaining,
|
|
}
|
|
if snooze_until is not None and current_time < snooze_until:
|
|
status["snooze_remaining"] = snooze_until - current_time
|
|
return status
|
|
|
|
|
|
def format_output(status: Status):
|
|
"""
|
|
Format status as Waybar JSON output.
|
|
https://github.com/Alexays/Waybar/wiki/Module:-Custom
|
|
"""
|
|
|
|
state = status["state"]
|
|
|
|
if state == "paused":
|
|
css_class = "paused"
|
|
text = "⏸ PAUSED"
|
|
tooltip = "Timer is paused. Click to unpause."
|
|
|
|
elif state == "working":
|
|
short_time = format_time(status["short_remaining"])
|
|
long_time = format_time(status["long_remaining"])
|
|
|
|
# Determine urgency class
|
|
overdue = min(status["short_remaining"], status["long_remaining"])
|
|
if overdue <= 0 and "snooze_remaining" in status:
|
|
snooze_time = format_time(status["snooze_remaining"])
|
|
css_class = "snoozed"
|
|
text = f"{snooze_time} (-{format_time(-overdue)})"
|
|
elif status["long_remaining"] <= 0:
|
|
css_class = ["overdue", "overdue-long"]
|
|
text = f"BREAK! {short_time} / {long_time}"
|
|
elif status["short_remaining"] <= 0:
|
|
css_class = ["overdue", "overdue-short"]
|
|
text = f"BREAK! {short_time} / {long_time}"
|
|
elif status["long_remaining"] < 300: # Less than 5 minutes
|
|
css_class = "warning"
|
|
text = f"{short_time} / {long_time}"
|
|
else:
|
|
css_class = "normal"
|
|
text = f"{short_time} / {long_time}"
|
|
|
|
tooltip = f"Short break in {short_time}\nLong break in {long_time}"
|
|
if css_class == "snoozed":
|
|
tooltip = f"Snoozed for {snooze_time}\n" + tooltip
|
|
|
|
elif state == "short_break":
|
|
break_time = format_time(status["break_remaining"])
|
|
short_time = format_time(status["short_remaining"])
|
|
css_class = "break"
|
|
text = f"SB {break_time} ({short_time})"
|
|
tooltip = f"Short break ends in {break_time}"
|
|
|
|
elif state == "long_break":
|
|
break_time = format_time(status["break_remaining"])
|
|
long_time = format_time(status["long_remaining"])
|
|
css_class = "break"
|
|
text = f"LB {break_time} ({long_time})"
|
|
tooltip = f"Long break ends in {break_time}"
|
|
|
|
elif state == "break_complete":
|
|
short_time = format_time(status["short_remaining"])
|
|
long_time = format_time(status["long_remaining"])
|
|
css_class = "break"
|
|
text = f"{short_time} / {long_time}"
|
|
tooltip = "Break complete. Timers reset."
|
|
|
|
else:
|
|
css_class = "normal"
|
|
text = "???"
|
|
tooltip = "Unknown state"
|
|
|
|
return {
|
|
"text": text,
|
|
"tooltip": tooltip,
|
|
"class": css_class
|
|
}
|
|
|
|
|
|
def main():
|
|
config = load_config()
|
|
log_path = get_log_path()
|
|
events = parse_log(log_path)
|
|
|
|
status = calculate_status(events, config)
|
|
output = format_output(status)
|
|
|
|
# Waybar's format-icons picks the icon by "alt"
|
|
reading = (log_path.parent / "reading").exists()
|
|
if reading:
|
|
output["tooltip"] += "\nReading mode: idle time counts as work"
|
|
if output["class"] == "snoozed":
|
|
output["alt"] = "snoozed"
|
|
else:
|
|
output["alt"] = "reading" if reading else "normal"
|
|
|
|
print(json.dumps(output))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|