Initial Commit

README.md

@@ -1 +1,61 @@
-# timetagger_watcher
+# TimeTagger Watcher
+
+## Introduction
+
+TimeTagger Watcher is a simple Python script designed to track your current task from TimeTagger and log it to an ActivityWatch bucket. This seamless integration can help you monitor your productivity and understand how your time is being allocated.
+
+## Prerequisites
+
+Before running the script, please ensure you have the following installed on your system:
+
+- Python 3.x
+- Requests package: `pip install requests`
+- ActivityWatch Client package: `pip install aw-client`
+
+## Setting Up
+
+Set environment variables for your TimeTagger URL, token, and optionally, the log level:
+
+```shell
+export TIMETAGGER_URL="https://yourtimetagger.url/api/endpoint"
+export TIMETAGGER_TOKEN="yourtimetaggertoken"
+export LOG_LEVEL="INFO" # Default is WARNING, options: DEBUG, INFO, WARNING, ERROR, CRITICAL
+```
+
+For local or self-signed certificates, you may disable SSL verification in the script by keeping `VERIFY_CERTIFICATE = False`.
+
+## Basic Usage
+
+To start tracking your TimeTagger tasks with ActivityWatch, simply run the `timetagger_watcher.py` script:
+```shell
+python timetagger_watcher.py
+```
+
+## Example
+
+Assuming your TimeTagger URL is `https://timetagger.example.com/timetagger/api/v2/records` and you have received a token, the setup would be:
+
+```shell
+export TIMETAGGER_URL="https://timetagger.example.com/timetagger/api/v2/records"
+export TIMETAGGER_TOKEN="your_unique_token_here"
+export LOG_LEVEL="INFO"
+```
+
+After setting these environment variables, run the script:
+```shell
+python timetagger_watcher.py
+```
+
+The log level `INFO` allows you to see informative messages providing insight into the script's operations and any issues encountered.
+
+## Note
+
+This script is meant to run continuously. If you wish to execute it in a more robust environment (e.g., background service), you may need to take additional steps to daemonize the script or monitor it with a tool like `supervisor` or `systemd`.
+
+## Contribution
+
+Your contributions to improve TimeTagger Watcher are welcome! Please feel free to fork the repository, make your changes, and submit a pull request.
+
+## License
+
+This project is licensed under the MIT License. Please see the LICENSE.md file for details.

requirements.txt

@@ -0,0 +1,4 @@
+aw_client==0.5.13
+aw_core==0.5.16
+Requests==2.31.0
+urllib3==2.1.0

timetagger_watcher.py

@@ -0,0 +1,74 @@
+import os
+import logging
+import time
+from datetime import datetime, timezone
+
+import requests
+import urllib3
+from aw_client import ActivityWatchClient
+from aw_core.models import Event
+
+# Constants
+TIME_TAGGER_URL = os.environ.get("TIMETAGGER_URL", "https://timetagger.timetagger.orb.local/timetagger/api/v2/records")
+VERIFY_CERTIFICATE = False # Set it to true if you don't have a self-signed cert.
+SLEEP_TIME = 1  # seconds
+BUCKET_ID_FORMAT = "{}_timetagger-bucket"
+
+# Disable warnings
+if not VERIFY_CERTIFICATE:
+	urllib3.disable_warnings() 
+
+# Set up logging
+log_level = os.environ.get("LOG_LEVEL", logging.WARNING)
+logging.basicConfig(level=log_level)
+
+def get_timetagger_token():
+    token = os.environ.get("TIMETAGGER_TOKEN")
+    if not token:
+        raise ValueError("TIMETAGGER_TOKEN environment variable not set.")
+    return token
+
+def get_current_task_from_timetagger():
+    token = get_timetagger_token()
+    headers = {"authtoken": token}
+    current_unix_time = int(time.time())
+    timerange = f"{current_unix_time}-{current_unix_time}"
+    params = {"timerange": timerange}
+    try:
+        response = requests.get(TIME_TAGGER_URL, params=params, headers=headers, verify=VERIFY_CERTIFICATE)
+        response.raise_for_status()  # Raises an HTTPError if the HTTP request returned an unsuccessful status code.
+        records = response.json().get('records', [])
+        if not records:
+            logging.info("No records found for the current time.")
+            return None
+        return records[0].get('ds')
+    except requests.RequestException as e:
+        logging.error(f"Failed to retrieve current task from TimeTagger: {e}")
+        return None
+
+def main():
+    client = ActivityWatchClient("timetagger")
+    hostname = client.client_hostname
+    bucket_id = BUCKET_ID_FORMAT.format(hostname)
+    client.create_bucket(bucket_id, event_type="tag")
+
+    with client:
+        while True:
+            task = get_current_task_from_timetagger()
+            if task:
+                heartbeat_data = {"tag": task}
+                now = datetime.now(timezone.utc)
+                heartbeat_event = Event(timestamp=now, data=heartbeat_data)
+
+                client.heartbeat(bucket_id, heartbeat_event,
+                                 pulsetime=SLEEP_TIME + 1,
+                                 queued=True,
+                                 commit_interval=4.0)
+
+            time.sleep(SLEEP_TIME)
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        logging.info("Program interrupted by user.")