ai checkin script

checkin.py

#!/usr/bin/env python

# open lxterminal app from app launcher
# python checkinscript.py

import glob
import json
import os
from pdb import run
import time
import subprocess
import anthropic
from anthropic.types import MessageParam
import threading

system = """\
Conduct brief check-ins with the user at 30-minute intervals. Your primary objective is to help the user stay focused and efficient in their tasks.

During each check-in, concisely inquire about the user's current activity and assess their progress. Ask questions like, "What are you working on right now?" and "How is your task progressing?" to gain a clear understanding of their status. Also **occasionally** ask questions like "What is the point of that?" or "Is there a better way to do that?" or "Are you overthinking things?" if it would be helpful to the user.

Focus on delivering value through concise questions that guide the user.

**Do not be too cheery. Talk like a serious math friend or something. DO NOT GIVE CLICHE ADVICE EVER.**

**Never use exclamation marks.**

**After the user has indicated their next steps**, end the checkin by including "ENDCHECKIN" anywhere in your message. **Do not end the check-in until the user has indicated an intent.** NEVER ASK A QUESTION AND END THE CHECK-IN IN THE SAME MESSAGE.

**Do not give suggestions unless specifically asked. Merely ask questions.**
"""

os.chdir(os.path.dirname(os.path.realpath(__file__)))

messages: list[MessageParam] = []


def run_both(foreground_func, background_func):
    # Create an event to signal the background thread to stop
    stop_event = threading.Event()

    # Create and start the background thread
    background_thread = threading.Thread(target=background_func, args=(stop_event,))
    background_thread.start()

    # Run the foreground function
    result = foreground_func()

    # Signal the background thread to stop
    stop_event.set()

    # Wait for the background thread to finish
    background_thread.join()

    return result


def refocus_loop(e: threading.Event):
    while not e.is_set():
        minimize_windows()
        open_lxterminal()
        # print("refocusing window")
        time.sleep(60)


# load file if it exists
def load_messages():
    global messages
    if os.path.exists("checkin_messages.json"):
        with open("checkin_messages.json", "r") as f:
            messages = json.load(f)


def minimize_windows():
    subprocess.call(["wmctrl", "-k", "on"])


def open_lxterminal():
    subprocess.call(["wmctrl", "-a", "LXTerminal"])


def play_sound():
    subprocess.run(["paplay", "/usr/share/sounds/freedesktop/stereo/complete.oga"])


def append(role: str, content: str):
    messages.append({"role": role, "content": content})
    with open("checkin.log", "a") as f:
        f.write(f"<{role}>\n{content}\n</{role}>\n")
    with open("checkin_messages.json", "w") as f:
        json.dump(messages, f, indent=4)


# messages: list[MessageParam]
async def query(temperature=0):
    if messages and messages[-1]["role"] == "assistant":
        print("WARNING: Adding junk ending user message")
        append("user", "<system></system>")
    # print("\n(running query...)", flush=True)
    client = anthropic.AsyncAnthropic(api_key=key)

    while True:
        try:
            async with client.messages.stream(
                max_tokens=200,
                messages=messages,
                model="claude-3-opus-20240229",
                system=system,
                temperature=temperature,
            ) as stream:
                print("Assistant: ", end="", flush=True)
                # print("(made it inside...\n)")
                async for text in stream.text_stream:
                    print(text, end="", flush=True)
                print("\n")
            accumulated = await stream.get_final_message()
            break
        except Exception as e:
            print("Error:", e)
            time.sleep(30)
            continue
    a = accumulated.content[0].text
    append("assistant", a)


async def onestep():
    load_messages()
    minimize_windows()
    play_sound()
    print("\n" * 50)
    time.sleep(1)
    open_lxterminal()
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
    append(
        "user",
        f"<system><time>{timestamp}</time>Please generate a short check-in question for the user. Remember not to end the check-in until after the user has indicated an intent.</system>",
    )
    await query(temperature=1)

    while True:
        answer = input("User: ")
        if not answer:
            answer = "<system>User provided empty response.</system>"
        append("user", answer)
        if messages[-1]["content"].lower() in ["q", "quit", "end", "endcheckin"]:
            break
        await query(temperature=0.3)
        if "endcheckin" in messages[-1]["content"].lower():
            break
    print("Check-in ended.")
    time.sleep(5)
    minimize_windows()


def onestep_sync():
    asyncio.run(onestep())


def main():
    while True:
        run_both(onestep_sync, refocus_loop)
        time.sleep(30 * 60)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())