351 lines
11 KiB
Python
Executable File
351 lines
11 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
|
|
|
|
class Status(TypedDict):
|
|
state: str
|
|
short_remaining: int
|
|
long_remaining: int
|
|
break_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"])
|
|
|
|
# If no events, assume just took breaks (timers at 0)
|
|
if not events:
|
|
return {
|
|
"state": "working",
|
|
"short_remaining": short_interval,
|
|
"long_remaining": long_interval,
|
|
}
|
|
|
|
# Track active time since each type of break
|
|
active_time_since_short_break = 0
|
|
active_time_since_long_break = 0
|
|
current_state = "working"
|
|
idle_start_time = None
|
|
|
|
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")]
|
|
|
|
# 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
|
|
|
|
# Process events to calculate active time and breaks
|
|
for i, (timestamp, event_type) in enumerate(idle_events):
|
|
if event_type == "idle_start":
|
|
current_state = "idle"
|
|
idle_start_time = timestamp
|
|
|
|
if i > 0 and events[i-1][1] == "idle_end":
|
|
# Active period just ended
|
|
active_end = timestamp
|
|
active_start = events[i-1][0]
|
|
duration = active_end - active_start
|
|
|
|
# Subtract any paused time within this active period
|
|
paused_time = get_paused_time_in_range(active_start, active_end)
|
|
duration -= paused_time
|
|
|
|
# Add to both counters, they reset independently
|
|
active_time_since_short_break += duration
|
|
active_time_since_long_break += duration
|
|
|
|
elif event_type == "idle_end":
|
|
current_state = "working"
|
|
|
|
if i > 0 and events[i-1][1] == "idle_start":
|
|
# Idle period just ended
|
|
idle_end = timestamp
|
|
idle_start = events[i-1][0]
|
|
idle_duration = idle_end - idle_start
|
|
|
|
# If the idle duration qualifies as a break, reset timers
|
|
if idle_duration >= long_duration:
|
|
active_time_since_long_break = 0
|
|
if idle_duration >= short_duration:
|
|
active_time_since_short_break = 0
|
|
|
|
if current_state == "working":
|
|
# If currently working, add time since last idle_end
|
|
if events and events[-1][1] == "idle_end":
|
|
last_active_start = events[-1][0]
|
|
current_active_duration = current_time - last_active_start
|
|
|
|
# Subtract any paused time in current active period
|
|
paused_time = get_paused_time_in_range(last_active_start, current_time)
|
|
current_active_duration -= paused_time
|
|
|
|
active_time_since_short_break += current_active_duration
|
|
active_time_since_long_break += current_active_duration
|
|
|
|
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 current_state == "idle" and idle_start_time:
|
|
# 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
|
|
return {
|
|
"state": "working",
|
|
"short_remaining": short_remaining,
|
|
"long_remaining": long_remaining,
|
|
}
|
|
|
|
|
|
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
|
|
if status["short_remaining"] <= 0 or status["long_remaining"] <= 0:
|
|
css_class = "overdue"
|
|
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}"
|
|
|
|
elif state == "short_break":
|
|
break_time = format_time(status["break_remaining"])
|
|
css_class = "break"
|
|
text = f"SB {break_time}"
|
|
tooltip = f"Short break ends in {break_time}"
|
|
|
|
elif state == "long_break":
|
|
break_time = format_time(status["break_remaining"])
|
|
css_class = "break"
|
|
text = f"LB {break_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)
|
|
|
|
print(json.dumps(output))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|