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 947e34c9ec
1 changed files with 11 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/awam/__main__.py
View File

@@ -32,8 +32,8 @@ 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 = "amazing-marvin-task"
try: try:
@@ -45,13 +45,13 @@ 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"), "title": tracked_task.get("title", "Unknown Task"),
"task_id": tracked_task.get("_id", ""), "task_id": tracked_task.get("_id", ""),
"database": tracked_task.get("db", "Tasks"), "database": tracked_task.get("db", "Tasks"),
"app": "Amazing Marvin", "app": "Amazing Marvin",
@@ -79,7 +79,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,13 +95,12 @@ 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 * 2.1),
pulsetime=int(heartbeat_interval * 1.5), queued=True,
queued=True, )
)
else: else:
if current_task_id is not None: if current_task_id is not None:
current_task_id = None current_task_id = None