felixm/aw-watcher-marvin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Call amazing marvin less often

Browse Source
This commit is contained in:
Felix Martin
2025-06-14 10:16:16 -04:00
parent ba1d3cdec3
commit a30b3cf76d
1 changed files with 12 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/awam/__main__.py
View File

@@ -32,9 +32,9 @@ def get_tracked_task(api_token: str) -> Optional[Dict[str, Any]]:
def setup_activitywatch_client() -> Tuple[ActivityWatchClient, str]: def setup_activitywatch_client() -> Tuple[ActivityWatchClient, str]:
"""Initialize ActivityWatch client and create bucket.""" """Initialize ActivityWatch client and create bucket."""
client = ActivityWatchClient("amazing-marvin-watcher", testing=False) client = ActivityWatchClient("aw-watcher-marvin", testing=False)
bucket_id = f"amazing-marvin-task_{client.client_hostname}" bucket_id = f"aw-watcher-marvin_{client.client_hostname}"
event_type = "amazing-marvin-task" event_type = "current-window"
try: try:
client.create_bucket(bucket_id, event_type=event_type) client.create_bucket(bucket_id, event_type=event_type)
@@ -45,16 +45,14 @@ def setup_activitywatch_client() -> Tuple[ActivityWatchClient, str]:
sys.exit(1) sys.exit(1)
def create_task_event(tracked_task: Optional[Dict[str, Any]]) -> Optional[Event]: def create_task_event(tracked_task: Dict[str, Any]) -> Event:
"""Create an ActivityWatch event from a tracked Amazing Marvin task.""" """Create an ActivityWatch event from a tracked Amazing Marvin task."""
if not tracked_task: if not tracked_task:
return None return None
event_data = { 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", "app": "Amazing Marvin",
"title": tracked_task.get("title", "Unknown Task"),
} }
return Event(timestamp=datetime.now(timezone.utc), data=event_data) return Event(timestamp=datetime.now(timezone.utc), data=event_data)
@@ -79,7 +77,7 @@ def main() -> None:
print("Press Ctrl+C to stop") print("Press Ctrl+C to stop")
current_task_id = None current_task_id = None
heartbeat_interval = 30 # seconds heartbeat_interval = 60
try: try:
with client: with client:
@@ -95,11 +93,10 @@ def main() -> None:
print(f"Now tracking: {task_title}") print(f"Now tracking: {task_title}")
event = create_task_event(tracked_task) event = create_task_event(tracked_task)
if event:
client.heartbeat( client.heartbeat(
bucket_id, bucket_id,
event, event,
pulsetime=int(heartbeat_interval * 1.5), pulsetime=int(heartbeat_interval * 2.1),
queued=True, queued=True,
) )
else: else: