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.
250 lines
7.9 KiB
Python
Executable File
250 lines
7.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
break-event - Log idle state transitions and auto-trim old events
|
|
|
|
Called by break-idle and swayidle to record when the user goes idle or resumes
|
|
activity, and by the Waybar menu. Automatically trims log entries before
|
|
qualifying break periods to keep log small.
|
|
|
|
In reading mode (a "reading" flag file next to the log), automatic idle is
|
|
ignored so hands-off reading counts as work. Breaks are started from the menu
|
|
with start-break and end on input.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import fcntl
|
|
import json
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
Event = tuple[int, str]
|
|
Events = list[Event]
|
|
|
|
|
|
class BreakConfig(TypedDict):
|
|
interval: str
|
|
duration: str
|
|
|
|
|
|
class Config(TypedDict):
|
|
short_break: BreakConfig
|
|
long_break: BreakConfig
|
|
|
|
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:
|
|
# Probably /run/user/$UID/break-timer
|
|
log_dir = Path(runtime_dir) / "break-timer"
|
|
else:
|
|
log_dir = Path.home() / ".cache" / "break-timer"
|
|
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
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 trim_log(events: Events, long_break_duration: int) -> Events:
|
|
"""
|
|
Trim events before qualifying long breaks.
|
|
Returns filtered list of events.
|
|
"""
|
|
if len(events) < 2:
|
|
return events
|
|
|
|
# Find the most recent idle period that qualifies as a long break
|
|
last_qualifying_idle_index = None
|
|
|
|
for i in range(len(events) - 1):
|
|
timestamp, event_type = events[i]
|
|
|
|
if event_type == "idle_start":
|
|
# Find the corresponding idle_end
|
|
for j in range(i + 1, len(events)):
|
|
next_timestamp, next_event_type = events[j]
|
|
if next_event_type == "idle_end":
|
|
idle_duration = next_timestamp - timestamp
|
|
if idle_duration >= long_break_duration:
|
|
last_qualifying_idle_index = j
|
|
break
|
|
elif next_event_type == "idle_start":
|
|
# No idle_end found, check if still idle
|
|
break
|
|
|
|
# Also check if currently idle and duration qualifies
|
|
if events and events[-1][1] == "idle_start":
|
|
idle_start_time = events[-1][0]
|
|
current_time = int(time.time())
|
|
if current_time - idle_start_time >= long_break_duration:
|
|
last_qualifying_idle_index = len(events) - 1
|
|
|
|
# Trim everything before the qualifying idle_start
|
|
if last_qualifying_idle_index is not None:
|
|
return events[last_qualifying_idle_index:]
|
|
|
|
return events
|
|
|
|
|
|
def write_log(log_path: Path, events: Events) -> None:
|
|
"""Write events to log file atomically"""
|
|
temp_path = log_path.with_suffix('.tmp')
|
|
|
|
try:
|
|
with open(temp_path, 'w') as f:
|
|
for timestamp, event_type in events:
|
|
f.write(f"{timestamp} {event_type}\n")
|
|
|
|
# Atomic rename
|
|
temp_path.rename(log_path)
|
|
except IOError as e:
|
|
print(f"Error writing log: {e}", file=sys.stderr)
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
|
|
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 in_break(events: Events) -> bool:
|
|
"""Whether the last idle-related event started a break. With none (e.g.
|
|
after skip_long) there is no work period yet, so idle_end must be logged."""
|
|
for _, event_type in reversed(events):
|
|
if event_type == "idle_start":
|
|
return True
|
|
if event_type == "idle_end":
|
|
return False
|
|
return True
|
|
|
|
|
|
COMMANDS = ("idle_start", "idle_end", "sleep", "wake", "start-break", "snooze",
|
|
"skip_long", "toggle-pause", "toggle-reading")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2 or sys.argv[1] not in COMMANDS:
|
|
print(f"Usage: break-event <{'|'.join(COMMANDS)}>", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
event_type = sys.argv[1]
|
|
config = load_config()
|
|
log_path = get_log_path()
|
|
lock_path = log_path.parent / "break-timer.lock"
|
|
reading_path = log_path.parent / "reading"
|
|
|
|
# Convert config to seconds
|
|
long_break_duration = parse_duration(config["long_break"]["duration"])
|
|
|
|
# Acquire lock for atomic operation
|
|
lock_path.touch()
|
|
with open(lock_path, 'w') as lock_file:
|
|
fcntl.flock(lock_file, fcntl.LOCK_EX)
|
|
|
|
# Read current events
|
|
events = parse_log(log_path)
|
|
|
|
try:
|
|
if event_type == "skip_long":
|
|
# Erase all events from the log, effectively resetting all timers.
|
|
write_log(log_path, [])
|
|
elif event_type == "toggle-pause":
|
|
current_time = int(time.time())
|
|
event_key = "paused" if not is_paused(events) else "unpaused"
|
|
events.append((current_time, event_key))
|
|
write_log(log_path, events)
|
|
elif event_type == "toggle-reading":
|
|
if reading_path.exists():
|
|
reading_path.unlink()
|
|
else:
|
|
reading_path.touch()
|
|
elif not is_paused(events): # Don't log idle events when paused
|
|
reading = reading_path.exists()
|
|
if event_type == "start-break":
|
|
logged = "idle_start"
|
|
elif event_type == "snooze":
|
|
logged = "snoozed"
|
|
elif event_type in ("sleep", "wake"):
|
|
logged = "idle_start" if event_type == "sleep" else "idle_end"
|
|
elif reading and event_type == "idle_start":
|
|
logged = None
|
|
elif reading and event_type == "idle_end" and not in_break(events):
|
|
logged = None
|
|
else:
|
|
logged = event_type
|
|
|
|
if logged:
|
|
events.append((int(time.time()), logged))
|
|
events = trim_log(events, long_break_duration)
|
|
write_log(log_path, events)
|
|
finally:
|
|
fcntl.flock(lock_file, fcntl.LOCK_UN)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|