ActivityWatch client script that queries VSCode events and shows time spent on projects

list_project_times.py

#!/usr/bin/env python3

from time import sleep
from datetime import datetime, timedelta, timezone
from dateutil.parser import isoparse

from aw_core.models import Event
from aw_client import ActivityWatchClient

# Server settings
HOST = "localhost"
PORT = 5600

# Buckets
HOST_NAME = "torchwood"
AFK_WATCHER = f"aw-watcher-afk_{HOST_NAME}"
VSCODE_WATCHER = f"aw-watcher-vscode_{HOST_NAME}"

# Settings
MAX_HISTORY = timedelta(days=30)
MERGE_DELTA = timedelta(minutes=10)

def proc_vsc_event(ev):
    start = isoparse(ev["timestamp"])
    if ev["duration"] <= 0:
        return None
    end = start + timedelta(seconds=ev["duration"])

    data = ev["data"]
    project_dir = data["project"]
    #file = data["file"]
    #language = data["language"]
    return {"project_dir": project_dir, "start": start, "end": end}


def proc_events(ev):
    events = list(sorted(filter(lambda it: it is not None, map(
        proc_vsc_event, ev)), key=lambda it: it["start"]))
    for i in range(1, len(events)):
        if (events[i - 1]["project_dir"] == events[i]["project_dir"]):
            if (events[i - 1]["end"] + MERGE_DELTA >= events[i]["start"]):
                if events[i - 1]["start"] <= events[i]["start"]:
                    events[i]["start"] = events[i - 1]["start"]
                    if events[i - 1]["end"] > events[i]["end"]:
                        events[i]["end"] = events[i - 1]["end"]
                    events[i - 1] = None
    events = list(filter(lambda it: it is not None, events))
    return events

def weekday_name(weekday):
    if weekday == 1:
        return "MON"
    elif weekday == 2:
        return "TUE"
    elif weekday == 3:
        return "WED"
    elif weekday == 4:
        return "THR"
    elif weekday == 5:
        return "FRI"
    elif weekday == 6:
        return "SAT"
    elif weekday == 7:
        return "SUN"

def display_duration(duration: timedelta, show_seconds=False):
    total_seconds = duration.total_seconds()
    hours = int(total_seconds // 3600)
    minutes = int((total_seconds % 3600) // 60)
    seconds = int(total_seconds % 60)

    result = []
    if hours > 0:
        result.append(f"{hours}h")
    if minutes > 0:
        result.append(f"{minutes}min")
    if show_seconds and seconds > 0:
        result.append(f"{seconds}s")
    return " ".join(result)


def main():
    client = ActivityWatchClient(
        "test-client", testing=True, host=HOST, port=PORT)

    end = datetime.now(timezone.utc)
    start = end - MAX_HISTORY
    
    query = f"RETURN = query_bucket('{VSCODE_WATCHER}');"
    res = client.query(query, [(start, end)])
    events = proc_events(res[0])

    projects = {}

    for e in events:
        start_date = e["start"].date()
        if e["project_dir"] in projects:
            project = projects[e["project_dir"]]
            if start_date in project:
                project[start_date] += e["end"] - e["start"]
            else:
                project[start_date] = e["end"] - e["start"]
        else:
            projects[e["project_dir"]] = {
                start_date: e["end"] - e["start"]
            }

    for path in projects:
        print(f"{path}:")
        total = timedelta(seconds=0)
        for day in projects[path]:
            total += projects[path][day]
            if projects[path][day] > timedelta(minutes=1):
                print("\t-",
                      day.strftime('%d.%m.'), weekday_name(day.isoweekday()), "-", display_duration(projects[path][day]))
        print(f"\t[ TOTAL ]", display_duration(total, True))


if __name__ == "__main__":
    main()