Initial commit

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+CLAUDE.md

.pre-commit-config.yaml

@@ -0,0 +1,23 @@
+repos:
+  - repo: https://github.com/psf/black
+    rev: 23.12.1
+    hooks:
+      - id: black
+        language_version: python3
+
+  - repo: https://github.com/pycqa/isort
+    rev: 5.13.2
+    hooks:
+      - id: isort
+
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.1.9
+    hooks:
+      - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.8.0
+    hooks:
+      - id: mypy
+        additional_dependencies: [types-requests]

README.md

@@ -0,0 +1,5 @@
+# ActivityWatch Watcher Amazing Marvin
+
+An Activity Watch Watcher that logs the task that is currently being tracked in
+Amazing Marvin.
+

pyproject.toml

@@ -0,0 +1,24 @@
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry]
+name = "awam"
+version = "0.1.0"
+description = "An ActivityWatch watcher that logs the tracked Amazing Marvin task."
+authors = ["Felix Martin"]
+readme = "README.md"
+packages = [{include = "awam", from = "src"}]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+aw-client = "*"
+aw-core = "*"
+requests = "*"
+keyring = "*"
+
+[tool.poetry.group.dev.dependencies]
+pre-commit = "*"
+
+[tool.poetry.scripts]
+awam = "awam.__main__:main"

src/awam/__main__.py

@@ -0,0 +1,119 @@
+import os
+import sys
+from datetime import datetime, timezone
+from time import sleep
+from typing import Any, Dict, Optional, Tuple
+
+import keyring
+import requests
+from aw_client import ActivityWatchClient
+from aw_core.models import Event
+
+
+def get_tracked_task(api_token: str) -> Optional[Dict[str, Any]]:
+    """Get the currently tracked task from Amazing Marvin API."""
+    url = "https://serv.amazingmarvin.com/api/trackedItem"
+    headers = {"X-API-Token": api_token, "Content-Type": "application/json"}
+
+    try:
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+
+        if response.status_code == 200:
+            return response.json()
+        else:
+            print(f"No tracked task found (status: {response.status_code})")
+            return None
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching tracked task: {e}")
+        return None
+
+
+def setup_activitywatch_client() -> Tuple[ActivityWatchClient, str]:
+    """Initialize ActivityWatch client and create bucket."""
+    client = ActivityWatchClient("amazing-marvin-watcher", testing=False)
+    bucket_id = f"amazing-marvin-task_{client.client_hostname}"
+    event_type = "amazing-marvin-task"
+
+    try:
+        client.create_bucket(bucket_id, event_type=event_type)
+        print(f"ActivityWatch bucket '{bucket_id}' ready")
+        return client, bucket_id
+    except Exception as e:
+        print(f"Error setting up ActivityWatch: {e}")
+        sys.exit(1)
+
+
+def create_task_event(tracked_task: Optional[Dict[str, Any]]) -> Optional[Event]:
+    """Create an ActivityWatch event from a tracked Amazing Marvin task."""
+    if not tracked_task:
+        return None
+
+    event_data = {
+        "task_title": tracked_task.get("title", "Unknown Task"),
+        "task_id": tracked_task.get("_id", ""),
+        "database": tracked_task.get("db", "Tasks"),
+        "app": "Amazing Marvin",
+    }
+
+    return Event(timestamp=datetime.now(timezone.utc), data=event_data)
+
+
+def main() -> None:
+    username = os.getlogin()
+    token = keyring.get_password("amazing-marvin-api-token", username)
+
+    if not token:
+        print(
+            "No API token found in keyring. Please store your Amazing Marvin API token:"
+        )
+        print(
+            f"Run: python -c \"import keyring; keyring.set_password('amazing-marvin-api-token', '{username}', 'YOUR_TOKEN_HERE')\""
+        )
+        sys.exit(1)
+
+    client, bucket_id = setup_activitywatch_client()
+
+    print("Starting Amazing Marvin task tracking...")
+    print("Press Ctrl+C to stop")
+
+    current_task_id = None
+    heartbeat_interval = 10  # seconds
+
+    try:
+        with client:
+            while True:
+                tracked_task = get_tracked_task(token)
+
+                if tracked_task:
+                    task_id = tracked_task.get("_id")
+                    task_title = tracked_task.get("title", "Unknown Task")
+
+                    if task_id != current_task_id:
+                        current_task_id = task_id
+                        print(f"Now tracking: {task_title}")
+
+                    event = create_task_event(tracked_task)
+                    if event:
+                        client.heartbeat(
+                            bucket_id,
+                            event,
+                            pulsetime=heartbeat_interval + 5,
+                            queued=True,
+                            commit_interval=30.0,
+                        )
+                else:
+                    if current_task_id is not None:
+                        current_task_id = None
+                        print("No task currently being tracked")
+
+                sleep(heartbeat_interval)
+
+    except KeyboardInterrupt:
+        print("\nStopping Amazing Marvin task tracking...")
+        sleep(1)  # Give time for queued events to be sent
+
+
+if __name__ == "__main__":
+    main()