Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@ To install the latest git version directly from github without cloning, run
`pip install git+https://github.com/ActivityWatch/aw-watcher-window.git`

To install from a cloned version, cd into the directory and run
`poetry install` to install inside an virtualenv. If you want to install it
system-wide it can be installed with `pip install .`, but that has the issue
that it might not get the exact version of the dependencies due to not reading
the poetry.lock file.
`poetry install` to install inside an virtualenv. You can run the binary via `aw-watcher-window`.

If you want to install it system-wide it can be installed with `pip install .`, but that has the issue
that it might not get the exact version of the dependencies due to not reading the poetry.lock file.

## Note to macOS users

Expand Down
27 changes: 20 additions & 7 deletions aw_watcher_window/config.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,31 @@
from configparser import ConfigParser
import argparse

from aw_core.config import load_config as _load_config


def load_config():
default_client_config = ConfigParser()
default_client_config["aw-watcher-window"] = {
"exclude_title": False,
"poll_time": "1.0"
}
default_client_config["aw-watcher-window-testing"] = {
default_client_config["aw-watcher-window"] = default_client_config["aw-watcher-window-testing"] = {
"exclude_title": False,
"poll_time": "1.0"
"poll_time": "1.0",
"strategy_macos": "jxa"
}

# TODO: Handle so aw-watcher-window testing gets loaded instead of testing is on
return _load_config("aw-watcher-window", default_client_config)["aw-watcher-window"]

def parse_args():
config = load_config()

default_poll_time = config.getfloat("poll_time")
default_exclude_title = config.getboolean("exclude_title")
default_strategy_macos = config.get("strategy_macos")

parser = argparse.ArgumentParser("A cross platform window watcher for Activitywatch.\nSupported on: Linux (X11), macOS and Windows.")
parser.add_argument("--testing", dest="testing", action="store_true")
parser.add_argument("--exclude-title", dest="exclude_title", action="store_true", default=default_exclude_title)
parser.add_argument("--verbose", dest="verbose", action="store_true")
parser.add_argument("--poll-time", dest="poll_time", type=float, default=default_poll_time)
parser.add_argument("--strategy", dest="strategy", default=default_strategy_macos, choices=["jxa", "applescript"], help="(macOS only) strategy to use for retrieving the active window")
parsed_args = parser.parse_args()
return parsed_args
29 changes: 19 additions & 10 deletions aw_watcher_window/lib.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

def get_current_window_linux() -> Optional[dict]:
from . import xlib

window = xlib.get_current_window()

if window is None:
Expand All @@ -13,20 +14,28 @@ def get_current_window_linux() -> Optional[dict]:
cls = xlib.get_window_class(window)
name = xlib.get_window_name(window)

return {"appname": cls, "title": name}
return {"app": cls, "title": name}


def get_current_window_macos(strategy: str) -> Optional[dict]:
# TODO should we use unknown when the title is blank like the other platforms?
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think so yes, I'll elaborate in #54.


def get_current_window_macos() -> Optional[dict]:
from . import macos
info = macos.getInfo()
app = macos.getApp(info)
title = macos.getTitle(info)
# `jxa` is the default & preferred strategy. It includes the url + incognito status
if strategy == "jxa":
from . import macos_jxa

return {"title": title, "appname": app}
return macos_jxa.getInfo()
elif strategy == "applescript":
from . import macos_applescript

return macos_applescript.getInfo()
else:
raise ValueError(f"invalid strategy '{strategy}'")


def get_current_window_windows() -> Optional[dict]:
from . import windows

window_handle = windows.get_active_window_handle()
app = windows.get_app_name(window_handle)
title = windows.get_window_title(window_handle)
Expand All @@ -36,14 +45,14 @@ def get_current_window_windows() -> Optional[dict]:
if title is None:
title = "unknown"

return {"appname": app, "title": title}
return {"app": app, "title": title}


def get_current_window() -> Optional[dict]:
def get_current_window(strategy: str = None) -> Optional[dict]:
if sys.platform.startswith("linux"):
return get_current_window_linux()
elif sys.platform == "darwin":
return get_current_window_macos()
return get_current_window_macos(strategy)
elif sys.platform in ["win32", "cygwin"]:
return get_current_window_windows()
else:
Expand Down
45 changes: 0 additions & 45 deletions aw_watcher_window/macos.py

This file was deleted.

31 changes: 31 additions & 0 deletions aw_watcher_window/macos_applescript.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
import os
import subprocess
from subprocess import PIPE
from typing import Dict


# the applescript version of the macos strategy is kept here until the jxa
# approach is proven out in production environments
# https://github.com/ActivityWatch/aw-watcher-window/pull/52


def getInfo() -> Dict[str, str]:
cmd = [
"osascript",
os.path.join(os.path.dirname(os.path.realpath(__file__)), "printAppTitle.scpt"),
]
p = subprocess.run(cmd, stdout=PIPE)
info = str(p.stdout, "utf8").strip()

app = getApp(info)
title = getTitle(info)

return {"app": app, "title": title}


def getApp(info: str) -> str:
return info.split('","')[0][1:]


def getTitle(info: str) -> str:
return info.split('","')[1][:-1]
97 changes: 97 additions & 0 deletions aw_watcher_window/macos_jxa.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import os
import json
import logging
from typing import Dict

logger = logging.getLogger(__name__)
script = None


def compileScript():
# https://stackoverflow.com/questions/44209057/how-can-i-run-jxa-from-swift
# https://stackoverflow.com/questions/16065162/calling-applescript-from-python-without-using-osascript-or-appscript
from OSAKit import OSAScript, OSALanguage

scriptPath = os.path.join(
os.path.dirname(os.path.realpath(__file__)), "printAppStatus.jxa"
)
scriptContents = open(scriptPath, mode="r").read()
javascriptLanguage = OSALanguage.languageForName_("JavaScript")

script = OSAScript.alloc().initWithSource_language_(
scriptContents, javascriptLanguage
)
(success, err) = script.compileAndReturnError_(None)

# should only occur if jxa was modified incorrectly
if not success:
raise Exception("error compiling jxa script")

return script


def getInfo() -> Dict[str, str]:
# use a global variable to cache the compiled script for performance
global script
if not script:
script = compileScript()

(result, err) = script.executeAndReturnError_(None)

if err:
# error structure:
# {
# NSLocalizedDescription = "Error: Error: Can't get object.";
# NSLocalizedFailureReason = "Error: Error: Can't get object.";
# OSAScriptErrorBriefMessageKey = "Error: Error: Can't get object.";
# OSAScriptErrorMessageKey = "Error: Error: Can't get object.";
# OSAScriptErrorNumberKey = "-1728";
# OSAScriptErrorRangeKey = "NSRange: {0, 0}";
# }

raise Exception("jxa error: {}".format(err["NSLocalizedDescription"]))

return json.loads(result.stringValue())


def background_ensure_permissions() -> None:
from multiprocessing import Process

permission_process = Process(target=ensure_permissions, args=(()))
permission_process.start()
return


def ensure_permissions() -> None:
from ApplicationServices import AXIsProcessTrusted
from AppKit import NSAlert, NSAlertFirstButtonReturn, NSWorkspace, NSURL

accessibility_permissions = AXIsProcessTrusted()
if not accessibility_permissions:
title = "Missing accessibility permissions"
info = "To let ActivityWatch capture window titles grant it accessibility permissions. \n If you've already given ActivityWatch accessibility permissions and are still seeing this dialog, try removing and re-adding them."

alert = NSAlert.new()
alert.setMessageText_(title)
alert.setInformativeText_(info)

ok_button = alert.addButtonWithTitle_("Open accessibility settings")

alert.addButtonWithTitle_("Close")
choice = alert.runModal()
print(choice)
if choice == NSAlertFirstButtonReturn:
NSWorkspace.sharedWorkspace().openURL_(
NSURL.URLWithString_(
"x-apple.systempreferences:com.apple.preference.security?Privacy_Accessibility"
)
)


if __name__ == "__main__":
print(getInfo())
print("Waiting 5 seconds...")
import time

time.sleep(5)
print(getInfo())
43 changes: 14 additions & 29 deletions aw_watcher_window/main.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import argparse
import logging
import traceback
import sys
Expand All @@ -11,18 +10,17 @@
from aw_client import ActivityWatchClient

from .lib import get_current_window
from .config import load_config
from .config import parse_args

logger = logging.getLogger(__name__)

# run with LOG_LEVEL=DEBUG
log_level = os.environ.get('LOG_LEVEL')
if log_level:
logger.setLevel(logging.__getattribute__(log_level.upper()))

def main():
# Read settings from config
config = load_config()
args = parse_args(
default_poll_time=config.getfloat("poll_time"),
default_exclude_title=config.getboolean("exclude_title"),
)
args = parse_args()

if sys.platform.startswith("linux") and ("DISPLAY" not in os.environ or not os.environ["DISPLAY"]):
raise Exception("DISPLAY environment variable not set")
Expand All @@ -45,43 +43,30 @@ def main():

sleep(1) # wait for server to start
with client:
heartbeat_loop(client, bucket_id, poll_time=args.poll_time, exclude_title=args.exclude_title)
heartbeat_loop(client, bucket_id, poll_time=args.poll_time, strategy=args.strategy, exclude_title=args.exclude_title)


def parse_args(default_poll_time: float, default_exclude_title: bool):
"""config contains defaults loaded from the config file"""
parser = argparse.ArgumentParser("A cross platform window watcher for Activitywatch.\nSupported on: Linux (X11), macOS and Windows.")
parser.add_argument("--testing", dest="testing", action="store_true")
parser.add_argument("--exclude-title", dest="exclude_title", action="store_true", default=default_exclude_title)
parser.add_argument("--verbose", dest="verbose", action="store_true")
parser.add_argument("--poll-time", dest="poll_time", type=float, default=default_poll_time)
return parser.parse_args()


def heartbeat_loop(client, bucket_id, poll_time, exclude_title=False):
def heartbeat_loop(client, bucket_id, poll_time, strategy, exclude_title=False):
while True:
if os.getppid() == 1:
logger.info("window-watcher stopped because parent process died")
break

try:
current_window = get_current_window()
current_window = get_current_window(strategy)
logger.debug(current_window)
except Exception as e:
logger.error("Exception thrown while trying to get active window: {}".format(e))
traceback.print_exc()
current_window = {"appname": "unknown", "title": "unknown"}
current_window = {"app": "unknown", "title": "unknown"}

now = datetime.now(timezone.utc)
if current_window is None:
logger.debug('Unable to fetch window, trying again on next poll')
else:
# Create current_window event
data = {
"app": current_window["appname"],
"title": current_window["title"] if not exclude_title else "excluded"
}
current_window_event = Event(timestamp=now, data=data)
if exclude_title:
current_window["title"] = "excluded"

current_window_event = Event(timestamp=now, data=current_window)

# Set pulsetime to 1 second more than the poll_time
# This since the loop takes more time than poll_time
Expand Down
Loading