Simple libnotify pomodoro timer which enforces i3 workspace focus

sftpomo.py

#!/usr/bin/env python3
import argparse
import datetime
import math
import json
import os
import subprocess
import threading
import time
import typing

import dateparser

cli = argparse.ArgumentParser()
cli.add_argument("--forbidden-ws", type=int, action="append")
cli.add_argument("--window-minutes", type=float, default=30)
cli.add_argument("--pause-minutes", type=float, default=5)
cli.add_argument("start")

args = cli.parse_args()
work_start = dateparser.parse(args.start).timestamp()
if not work_start:
    cli.error(f"cannot parse start time {work_start!r}")


def notify(title: str, text: str, set_term_title: bool = False):
    subprocess.run(["notify-send", title, text])
    if set_term_title and os.isatty(0):
        print(f"\N{esc}]0;{title}: {text}\N{bel}", end="")
    print(f"{title}: {text}")


class WorkspaceEnforcer:
    def __init__(self, forbidden_ws: typing.Iterable[int]):
        self._forbidden_ws = set(forbidden_ws)
        self._proc = subprocess.Popen(
            ["i3-msg", "-t", "subscribe", "-m", '[ "workspace" ]'],
            stdout=subprocess.PIPE,
        )
        self.locked = threading.Event()
        self._thread = threading.Thread(target=self._run)
        self._thread.start()

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self._proc.kill()
        self._thread.join()

    def _run(self):
        for line in self._proc.stdout:
            msg = json.loads(line)
            if msg.get("change") != "focus":
                continue
            new = msg.get("current", {}).get("num")
            if new in self._forbidden_ws:
                if not self.locked.is_set():
                    continue
                old = msg.get("old", {}).get("num")
                if old is None:
                    continue
                try:
                    if time.time() < os.path.getmtime(f"/tmp/allow-{new}") + 10:
                        # lock overridden
                        continue
                except FileNotFoundError:
                    pass
                notify(
                    f"switching to WS {new} forbidden",
                    f"touch /tmp/allow-{new} to allow",
                )
                subprocess.run(["i3-msg", "workspace", str(old)], stdout=subprocess.DEVNULL)


def timestamp_to_str(timestamp: float):
    dt = datetime.datetime.fromtimestamp(timestamp)
    if dt.date() == datetime.date.today():
        return dt.strftime("%H:%M")
    else:
        return dt.strftime("%Y-%m-%d %H:%M")


def sleep_until(timestamp: float):
    now = time.time()
    time.sleep(timestamp - now)


window_len = args.window_minutes * 60
pause_len = args.pause_minutes * 60
work_until = work_start + window_len - pause_len
if work_until < time.time() + 1:
    extra_windows = math.ceil((time.time() + 1 - work_until) / window_len)
    print(
        f"work window until {timestamp_to_str(work_until)} is already over; forwarding by {extra_windows} windows"
    )
    work_until += window_len * extra_windows


with WorkspaceEnforcer(args.forbidden_ws or []) as wsenforcer:
    while True:
        wsenforcer.locked.set()
        notify("pomodoro", f"work until {timestamp_to_str(work_until)}", True)
        sleep_until(work_until)
        wsenforcer.locked.clear()
        pause_until = work_until + pause_len
        notify("pomodoro", f"pause until {timestamp_to_str(pause_until)}", True)
        sleep_until(pause_until)
        work_until = work_until + window_len